""" End-to-End (E2E) Integration Journey Test. Simulates the complete patient and operational lifecycle in a single cohesive story: 1. System Provisioning: Lab, Admin, Receptionist, Phlebotomist, Pathologist, Accountant. 2. Master Data & Catalog: Referral partner with 10% commission, Test Category, Lab Test, Wellness Package. 3. Patient & Appointment: Patient registration with code assignment, appointment booking and confirmation. 4. Order Booking: Conversion to VisitOrder with price snapshots and referral partner attribution. 5. Phlebotomy: Tube barcode generation, sample collection, order progression. 6. Clinical Execution: Result observations, abnormal flagging, draft report creation. 7. Verification & Release: Pathologist review, sign-off, WeasyPrint PDF report generation with QR code. 8. Client Access: Cryptographic 72h Redis link, public report view and PDF download. 9. Billing & Split Payments: Bill generation, discount, split tender settlement (Cash + Card). 10. Referral Settlement: Commission calculation and payout disbursement. 11. Executive Analytics: Operational funnel, financial ledgers, daily trends, and top performers. 12. System Health: Kubernetes/orchestration readiness probe (/health/ready). """ from datetime import UTC, date, datetime from decimal import Decimal from uuid import uuid4 import pytest import pytest_asyncio from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import create_access_token, hash_password from app.models.appointment import Appointment, AppointmentStatus, BookingChannel from app.models.billing import Bill, BillStatus, Payment, PaymentMode from app.models.lab import Lab from app.models.package import TestPackage as LabTestPackage from app.models.referral import CommissionEntry, Referral, ReferralType from app.models.report import ReportStatus from app.models.test import ( Test as LabTest, ) from app.models.test import ( TestCategory as LabTestCategory, ) from app.models.test import ( TestParameter as LabParameter, ) from app.models.test import ( TestParameterRange as LabParameterRange, ) from app.models.test import ( TestResultType as LabTestResultType, ) from app.models.user import User, UserRole from app.repositories.permission_repository import PermissionRepository from app.services.client_link_service import ClientLinkService @pytest_asyncio.fixture(autouse=True) async def seed_perms(db_session: AsyncSession) -> None: """Seed permissions for RBAC checks.""" repo = PermissionRepository() await repo.seed_permissions(db_session) @pytest_asyncio.fixture async def e2e_lab(db_session: AsyncSession) -> Lab: """Create test laboratory.""" lab = Lab(name="Munal Central Reference Lab", address="Tripureshwor, Kathmandu") db_session.add(lab) await db_session.commit() await db_session.refresh(lab) return lab @pytest_asyncio.fixture async def staff_tokens(db_session: AsyncSession, e2e_lab: Lab) -> dict[str, str]: """Create tokens for various staff roles.""" roles = { "admin": UserRole.SUPERADMIN, "receptionist": UserRole.RECEPTIONIST, "phlebotomist": UserRole.PHLEBOTOMIST, "pathologist": UserRole.PATHOLOGIST, "radiologist": UserRole.RADIOLOGIST, "accountant": UserRole.ACCOUNTANT, } tokens: dict[str, str] = {} for role_name, user_role in roles.items(): user = User( lab_id=e2e_lab.id, name=f"Staff {role_name.capitalize()}", phone=f"980000000{len(tokens) + 1}", email=f"{role_name}@munal.test", role=user_role, hashed_password=hash_password("Pass123!"), is_active=True, ) db_session.add(user) await db_session.commit() await db_session.refresh(user) tokens[role_name] = create_access_token( subject=str(user.id), extra_claims={ "phone": user.phone, "email": user.email, "role": user.role.value, "lab_id": str(e2e_lab.id), }, ) return tokens @pytest.mark.asyncio async def test_full_laboratory_lifecycle_e2e_journey( client: AsyncClient, db_session: AsyncSession, e2e_lab: Lab, staff_tokens: dict[str, str], ) -> None: """Execute complete end-to-end laboratory and clinical workflow.""" admin_auth = {"Authorization": f"Bearer {staff_tokens['admin']}"} receptionist_auth = {"Authorization": f"Bearer {staff_tokens['receptionist']}"} phlebotomist_auth = {"Authorization": f"Bearer {staff_tokens['phlebotomist']}"} pathologist_auth = {"Authorization": f"Bearer {staff_tokens['pathologist']}"} radiologist_auth = {"Authorization": f"Bearer {staff_tokens['radiologist']}"} accountant_auth = {"Authorization": f"Bearer {staff_tokens['accountant']}"} # ───────────────────────────────────────────────────────────────────────── # Step 1: Health & Readiness Probe # ───────────────────────────────────────────────────────────────────────── health_res = await client.get("/health") assert health_res.status_code == 200 assert health_res.json()["status"] == "ok" ready_res = await client.get("/health/ready") assert ready_res.status_code == 200 assert ready_res.json()["status"] == "ready" assert ready_res.json()["database"] == "connected" assert ready_res.json()["session_store"] == "ready" # ───────────────────────────────────────────────────────────────────────── # Step 2: Referral Partner Registration # ───────────────────────────────────────────────────────────────────────── doctor = Referral( lab_id=e2e_lab.id, name="Dr. Sandeep Giri", referral_type=ReferralType.DOCTOR, phone="9841887766", commission_rate=Decimal("10.00"), ) db_session.add(doctor) await db_session.commit() await db_session.refresh(doctor) # ───────────────────────────────────────────────────────────────────────── # Step 3: Test Catalog & Package Setup # ───────────────────────────────────────────────────────────────────────── category = LabTestCategory( lab_id=e2e_lab.id, name="Hematology", description="Blood cell counts and morphological examinations", ) db_session.add(category) await db_session.flush() test_cbc = LabTest( lab_id=e2e_lab.id, category_id=category.id, name="Complete Blood Count", code="CBC", price=Decimal("600.00"), result_type=LabTestResultType.COMPOSITE, ) db_session.add(test_cbc) await db_session.flush() param_wbc = LabParameter( test_id=test_cbc.id, name="Total Leukocyte Count (WBC)", unit="/cumm", decimal_places=0, sort_order=1, ) db_session.add(param_wbc) await db_session.flush() ref_range = LabParameterRange( parameter_id=param_wbc.id, gender=None, age_min=None, age_max=None, numeric_low=Decimal("4000"), numeric_high=Decimal("11000"), ) db_session.add(ref_range) pkg_wellness = LabTestPackage( lab_id=e2e_lab.id, name="Executive Health Screening", code="EHS01", price=Decimal("2000.00"), ) db_session.add(pkg_wellness) await db_session.commit() await db_session.refresh(test_cbc) await db_session.refresh(param_wbc) await db_session.refresh(pkg_wellness) # ───────────────────────────────────────────────────────────────────────── # Step 4: Patient Registration # ───────────────────────────────────────────────────────────────────────── patient_res = await client.post( "/api/v1/patients", json={ "lab_id": str(e2e_lab.id), "name": "Dinesh Sharma", "gender": "male", "age": 35, "phone": "9841999888", "email": "dinesh@example.com", "address": "Baneshwor, Kathmandu", }, headers=receptionist_auth, ) assert patient_res.status_code == 201 patient_data = patient_res.json()["data"] patient_id = patient_data["id"] assert patient_data["patient_code"].startswith("P26-") # ───────────────────────────────────────────────────────────────────────── # Step 5: Appointment Booking # ───────────────────────────────────────────────────────────────────────── apt = Appointment( lab_id=e2e_lab.id, patient_id=uuid4() if False else patient_data["id"], status=AppointmentStatus.CONFIRMED, booked_via=BookingChannel.PORTAL, ) db_session.add(apt) await db_session.commit() await db_session.refresh(apt) # ───────────────────────────────────────────────────────────────────────── # Step 6: Order Creation (Appointment to VisitOrder Handoff) # ───────────────────────────────────────────────────────────────────────── order_res = await client.post( "/api/v1/orders", json={ "lab_id": str(e2e_lab.id), "patient_id": patient_id, "appointment_id": str(apt.id), "referral_id": str(doctor.id), "test_ids": [str(test_cbc.id)], "package_ids": [str(pkg_wellness.id)], "notes": "Patient fasting overnight", }, headers=receptionist_auth, ) assert order_res.status_code == 201 order_data = order_res.json()["data"] order_id = order_data["id"] # Total price: CBC (600) + Package (2000) = 2600.00 assert Decimal(str(order_data["total_amount"])) == Decimal("2600.00") assert order_data["status"] == "pending" # Phase A proof: every order is born with exactly one DRAFT report, # so lab and radiology queues work uniformly (incl. X-ray-only orders). draft_list_res = await client.get( "/api/v1/reports", params={"order_id": order_id}, headers=receptionist_auth, ) assert draft_list_res.status_code == 200 drafts = draft_list_res.json()["data"] assert len(drafts) == 1 assert drafts[0]["status"] == "draft" auto_draft_id = drafts[0]["id"] # ───────────────────────────────────────────────────────────────────────── # Step 7: Sample Collection & Barcode Scan # ───────────────────────────────────────────────────────────────────────── sample_res = await client.post( f"/api/v1/orders/{order_id}/samples", json={"sample_type": "EDTA Whole Blood", "notes": "Purple top tube"}, headers=phlebotomist_auth, ) assert sample_res.status_code == 201 sample_data = sample_res.json()["data"] sample_id = sample_data["id"] barcode = sample_data["barcode"] assert barcode == sample_data["sample_code"] assert barcode.startswith("S") and barcode[1:].isdigit() # Mark sample collected collect_res = await client.post( f"/api/v1/samples/{sample_id}/collect", json={"notes": "Venipuncture clean"}, headers=phlebotomist_auth, ) assert collect_res.status_code == 200 assert collect_res.json()["data"]["status"] == "collected" # Phase A proof: a second specimen type gets its own distinct sample code urine_res = await client.post( f"/api/v1/orders/{order_id}/samples", json={"sample_type": "urine", "notes": "Midstream catch"}, headers=phlebotomist_auth, ) assert urine_res.status_code == 201 assert urine_res.json()["data"]["sample_code"] != barcode # ───────────────────────────────────────────────────────────────────────── # Step 8: Clinical Test Results Entry (fills the auto-created draft) # ───────────────────────────────────────────────────────────────────────── # Normal WBC: 7,500 /cumm from app.repositories.report_repository import ReportRepository report = await ReportRepository().get_by_id(db_session, auto_draft_id) assert report is not None assert report.status == ReportStatus.DRAFT report.result_data = { "order_number": order_data["order_number"], "patient_age_at_test": 35, "patient_gender": "male", "parameters": { str(param_wbc.id): { "parameter_id": str(param_wbc.id), "parameter_name": param_wbc.name, "unit": param_wbc.unit, "value": "7500", "reference_low": "4000", "reference_high": "11000", "flag": "N", "is_abnormal": False, "notes": "Normal cell morphology", } }, "last_updated_at": datetime.now(UTC).isoformat(), } report.notes = "Specimen analyzed on Sysmex automated hematology analyzer" await db_session.commit() await db_session.refresh(report) # ───────────────────────────────────────────────────────────────────────── # Step 9: Pathologist Review, Verification & Release # ───────────────────────────────────────────────────────────────────────── # Phase C proof: pathologist uploads a signature scan (whitened server-side) from io import BytesIO from PIL import Image sig_img = Image.new("RGB", (200, 80), "white") for x in range(40, 160): for y in range(30, 50): sig_img.putpixel((x, y), (10, 10, 10)) sig_buf = BytesIO() sig_img.save(sig_buf, format="PNG") sig_upload = await client.post( "/api/v1/users/me/signature", files={"file": ("sig.png", sig_buf.getvalue(), "image/png")}, headers=pathologist_auth, ) assert sig_upload.status_code == 200 assert sig_upload.json()["data"]["signature_image_path"] is not None # 1. Submit for verification submit_res = await client.post( f"/api/v1/reports/{report.id}/submit", json={"notes": "Ready for pathologist sign-off"}, headers=pathologist_auth, ) assert submit_res.status_code == 200 assert submit_res.json()["data"]["status"] == "pending_verification" # 2. Verify and electronically sign verify_res = await client.post( f"/api/v1/reports/{report.id}/verify", json={ "verification_reason": "Normal adult leukocyte distribution", "clinical_notes": "No morphological abnormalities observed", }, headers=pathologist_auth, ) assert verify_res.status_code == 200 assert verify_res.json()["data"]["status"] == "verified" # 3. Release report (triggers PDF generation & automated client notification) release_res = await client.post( f"/api/v1/reports/{report.id}/release", json={}, headers=pathologist_auth, ) assert release_res.status_code == 200 assert release_res.json()["data"]["status"] == "released" # 4. Verify WeasyPrint dynamic PDF generation pdf_res = await client.get( f"/api/v1/reports/{report.id}/pdf", headers=pathologist_auth, ) assert pdf_res.status_code == 200 assert pdf_res.headers["content-type"].startswith("application/pdf") assert pdf_res.content.startswith(b"%PDF-") # Phase C proof: the transparent signature lands on the exact sign-off block from app.repositories.report_repository import ReportRepository as _ReportRepo from app.services.pdf_service import PDFReportService as _PdfService db_session.expunge_all() fresh_report = await _ReportRepo().get_by_id(db_session, report.id) assert fresh_report is not None assert fresh_report.verifier is not None html = _PdfService().render_html(fresh_report) assert "data:image/png;base64," in html # ───────────────────────────────────────────────────────────────────────── # Step 9b: Radiologist Amendment + Per-Report Commission (Phase E) # ───────────────────────────────────────────────────────────────────────── amend_res = await client.post( f"/api/v1/reports/{report.id}/amend", json={"reason": "Addendum: correlate clinically"}, headers=pathologist_auth, ) assert amend_res.status_code == 200, amend_res.text v2_id = amend_res.json()["data"]["id"] # Superadmin sets the radiologist's 15% rate via staff update radiologist_user_res = await client.get("/api/v1/users", params={"limit": 100}, headers=admin_auth) assert radiologist_user_res.status_code == 200 radio_entry = next(u for u in radiologist_user_res.json()["data"] if u["role"] == "radiologist") rate_res = await client.patch( f"/api/v1/users/{radio_entry['id']}", json={"commission_rate": "15.00"}, headers=admin_auth, ) assert rate_res.status_code == 200, rate_res.text assert str(rate_res.json()["data"]["commission_rate"]) == "15.00" submit_v2 = await client.post( f"/api/v1/reports/{v2_id}/submit", json={"notes": "Radiologist addendum ready"}, headers=radiologist_auth, ) assert submit_v2.status_code == 200 verify_v2 = await client.post( f"/api/v1/reports/{v2_id}/verify", json={"verification_reason": "Imaging addendum authorized"}, headers=radiologist_auth, ) assert verify_v2.status_code == 200 release_v2 = await client.post( f"/api/v1/reports/{v2_id}/release", json={}, headers=radiologist_auth, ) assert release_v2.status_code == 200 # 15% of the 2600.00 order = 390.00 NPR accrued on release ledger_res = await client.get("/api/v1/radiologist-commissions", headers=admin_auth) assert ledger_res.status_code == 200, ledger_res.text entries = ledger_res.json()["data"] assert len(entries) == 1 assert entries[0]["report_id"] == v2_id assert Decimal(str(entries[0]["commission_amount"])) == Decimal("390.00") # ───────────────────────────────────────────────────────────────────────── # Step 10: Client Link Resolution (72h Redis Access) # ───────────────────────────────────────────────────────────────────────── link_service = ClientLinkService() token, _, _ = await link_service.create_report_token( report_id=report.id, patient_id=patient_data["id"], lab_id=e2e_lab.id, ) # Public unauthenticated view client_view_res = await client.get(f"/api/v1/client/reports/{token}") assert client_view_res.status_code == 200 assert client_view_res.json()["data"]["patient"]["name"] == "Dinesh Sharma" # Public unauthenticated PDF download client_pdf_res = await client.get(f"/api/v1/client/reports/{token}/pdf") assert client_pdf_res.status_code == 200 assert client_pdf_res.content.startswith(b"%PDF-") # ───────────────────────────────────────────────────────────────────────── # Step 11: Billing & Split Tender Payment Settlement # ───────────────────────────────────────────────────────────────────────── # Active Bill: Subtotal 2600.00, Discount 100.00, Payable 2500.00 bill = Bill( lab_id=e2e_lab.id, patient_id=patient_data["id"], order_id=order_id, bill_number="INV/26/000001", status=BillStatus.UNPAID, subtotal=Decimal("2600.00"), discount_amount=Decimal("100.00"), total_amount=Decimal("2500.00"), paid_amount=Decimal("0.00"), due_amount=Decimal("2500.00"), ) db_session.add(bill) await db_session.commit() await db_session.refresh(bill) # Tender 1: NPR 1,500.00 Cash pay_cash = Payment( lab_id=e2e_lab.id, bill_id=bill.id, amount=Decimal("1500.00"), payment_mode=PaymentMode.CASH, paid_at=datetime.now(UTC), ) # Tender 2: NPR 1,000.00 Card (completes total settlement) pay_card = Payment( lab_id=e2e_lab.id, bill_id=bill.id, amount=Decimal("1000.00"), payment_mode=PaymentMode.CARD, paid_at=datetime.now(UTC), ) db_session.add_all([pay_cash, pay_card]) # Update bill settlement state bill.paid_amount = Decimal("2500.00") bill.due_amount = Decimal("0.00") bill.status = BillStatus.PAID await db_session.commit() # Phase A proof: the invoice carries both specimen codes bill_detail_res = await client.get( f"/api/v1/bills/order/{order_id}", headers=accountant_auth, ) assert bill_detail_res.status_code == 200, bill_detail_res.text samples_on_bill = bill_detail_res.json()["data"]["samples"] assert {s["sample_type"] for s in samples_on_bill} == {"EDTA Whole Blood", "urine"} # ───────────────────────────────────────────────────────────────────────── # Step 12: Referral Commission Accounting & Settlement # ───────────────────────────────────────────────────────────────────────── # 10% of 2500 = 250.00 NPR comm = CommissionEntry( lab_id=e2e_lab.id, referral_id=doctor.id, order_id=order_id, order_amount=Decimal("2500.00"), commission_rate=Decimal("10.00"), commission_amount=Decimal("250.00"), is_settled=True, settled_at=datetime.now(UTC), ) db_session.add(comm) await db_session.commit() # ───────────────────────────────────────────────────────────────────────── # Step 13: Executive Dashboard & Analytics Reflection # ───────────────────────────────────────────────────────────────────────── dashboard_res = await client.get( "/api/v1/dashboard/overview", params={"refresh": "true"}, headers=admin_auth, ) assert dashboard_res.status_code == 200 dash_data = dashboard_res.json()["data"] # Operational metrics op = dash_data["operational"] assert op["total_patients"] >= 1 assert op["total_orders"] >= 1 # Financial metrics fin = dash_data["financial"] assert Decimal(fin["total_invoiced"]) >= Decimal("2500.00") assert Decimal(fin["total_collected"]) >= Decimal("2500.00") assert Decimal(fin["total_outstanding"]) == Decimal("0.00") assert Decimal(fin["revenue_by_tender"]["cash"]) >= Decimal("1500.00") assert Decimal(fin["revenue_by_tender"]["card"]) >= Decimal("1000.00") assert Decimal(fin["accrued_commissions"]) >= Decimal("250.00") # Accountant access verification acct_res = await client.get("/api/v1/dashboard/financial", headers=accountant_auth) assert acct_res.status_code == 200 # Top performers top_tests = dash_data["top_tests"] assert any(t["code"] == "CBC" for t in top_tests) top_packages = dash_data["top_packages"] assert any(p["code"] == "EHS01" for p in top_packages) top_doctors = dash_data["top_referrers"] assert any(d["name"] == "Dr. Sandeep Giri" for d in top_doctors) # Time series trends trends = dash_data["trend"] today_str = date.today().strftime("%Y-%m-%d") today_point = next((pt for pt in trends if pt["date"] == today_str), None) assert today_point is not None assert today_point["order_count"] >= 1 assert Decimal(today_point["revenue"]) >= Decimal("2500.00") # Housekeeping: remove the test signature PNG written under ./storage import pathlib as _pathlib for _sig in _pathlib.Path("storage/signatures").glob("*.png"): _sig.unlink(missing_ok=True)