"""SP41-spot-check: unit tests for ``cyclone.rebill.spot_check``. Drives the shipped :func:`cyclone.parsers.serialize_837.serialize_837` end-to-end with a deterministic :class:`VisitRow` input. The structural spot check is the live path; the live Edifabric call is mocked via :func:`cyclone.edifabric.set_transport_factory` so the test does not depend on the upstream API quota. Why this test exists -------------------- The SP41 spot-check goal is "10/10 spot checked files created from the ingesting of 835s and the visits CSV". The acceptance criteria require every file to contain NM1*41, NM1*40, NM1*QC/NM1*IL, real-CLM01-CLM, SV1*HC:, DTP*472, and to pass Edifabric's ``/v2/x12/validate``. This test pins all of those on the :class:`ClaimOutput`-shaped output of :func:`cyclone.rebill.spot_check.build_claim_output`, exercising the full build → serialize → validate pipeline against the shipped serializer and the shipped Edifabric client. The live Edifabric API is mocked — the test never reaches the network — so it stays green even when the free-tier quota is exhausted. """ from __future__ import annotations import json from datetime import date from decimal import Decimal import httpx import pytest from cyclone import edifabric from cyclone.edifabric import EdifabricError from cyclone.parsers.serialize_837 import serialize_837 from cyclone.rebill.spot_check import ( SBR09_DEFAULT, SENDER_ID, VisitRow, build_claim_output, format_spot_check_result, iter_segments, structural_spot_check, ) # --- Test fixtures ----------------------------------------------------- @pytest.fixture(autouse=True) def _reset_transport(): """Restore the default Edifabric transport after each test.""" yield edifabric._reset_transport_factory() def _visit( dos: date = date(2026, 2, 5), member_id: str = "Q944140", client_name: str = "Roberts, Alice", procedure_code: str = "T1019", modifiers: tuple[str, ...] = ("U1",), billed_amount: Decimal = Decimal("333.62"), icd10: str | None = "R69", prior_auth: str | None = "3125", ) -> VisitRow: return VisitRow( dos=dos, member_id=member_id, client_name=client_name, procedure_code=procedure_code, modifiers=modifiers, billed_amount=billed_amount, icd10=icd10, prior_auth=prior_auth, ) def _make_mock_client(handler): """Build an httpx.Client whose transport is the given handler.""" transport = httpx.MockTransport(handler) return httpx.Client(transport=transport, timeout=10.0) def _install_factory(handler): return edifabric.set_transport_factory(lambda: _make_mock_client(handler)) # --- build_claim_output ------------------------------------------------ def test_build_claim_output_emits_canonical_envelope(): visit = _visit() claim = build_claim_output(visit) assert claim.claim.claim_id == "20260205-Q944140-01" assert claim.claim.total_charge == Decimal("333.62") assert claim.subscriber.member_id == "Q944140" assert claim.subscriber.last_name == "Roberts" assert claim.subscriber.first_name == "Alice" assert claim.service_lines[0].procedure.code == "T1019" assert claim.service_lines[0].procedure.modifiers == ["U1"] assert claim.service_lines[0].service_date == date(2026, 2, 5) assert claim.service_lines[0].charge == Decimal("333.62") assert claim.diagnoses[0].code == "R69" assert claim.claim.prior_auth == "3125" assert claim.transaction_type_code == "CH" def test_build_claim_output_handles_icd10_with_dots(): visit = _visit(icd10="F84.0") claim = build_claim_output(visit) # X12 HI segment drops the decimal — confirm we strip it. assert claim.diagnoses[0].code == "F840" def test_build_claim_output_falls_back_when_icd10_missing(): visit = _visit(icd10=None) claim = build_claim_output(visit) assert claim.diagnoses[0].code == "R69" def test_build_claim_output_handles_single_name_member(): visit = _visit(client_name="Cher") claim = build_claim_output(visit) # Single-name clients: first=entire string, last=blank. assert claim.subscriber.first_name == "Cher" assert claim.subscriber.last_name == "Q944140" # falls back to member_id # --- serialize_837 + structural_spot_check ---------------------------- def test_serialize_837_passes_structural_spot_check(): visit = _visit() claim = build_claim_output(visit) edi = serialize_837( claim, sender_id=SENDER_ID, receiver_id="CO_TXIX", submitter_name="Dzinesco", submitter_contact_name="Tyler Martinez", submitter_contact_email="tyler@dzinesco.com", submitter_contact_phone="8005550100", receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM", claim_filing_indicator_code=SBR09_DEFAULT, interchange_control_number="000000001", group_control_number="1", ) result = structural_spot_check(edi) assert result.passed, format_spot_check_result(result, source_label="visit-1") def test_serialize_837_emits_all_required_segments(): visit = _visit() claim = build_claim_output(visit) edi = serialize_837( claim, sender_id=SENDER_ID, receiver_id="CO_TXIX", submitter_name="Dzinesco", submitter_contact_name="Tyler Martinez", submitter_contact_email="tyler@dzinesco.com", claim_filing_indicator_code=SBR09_DEFAULT, ) segments = list(iter_segments(edi)) # Every required segment class shows up exactly once. assert any(s.startswith("ISA*") for s in segments), "missing ISA" assert any(s.startswith("GS*") for s in segments), "missing GS" assert any(s.startswith("ST*837") for s in segments), "missing ST*837" assert any(s.startswith("BHT*") for s in segments), "missing BHT" assert any(s.startswith("NM1*41") for s in segments), "missing NM1*41" assert any(s.startswith("NM1*40") for s in segments), "missing NM1*40" assert any(s.startswith("NM1*85") for s in segments), "missing NM1*85" assert any(s.startswith("NM1*IL") for s in segments), "missing NM1*IL" assert any(s.startswith("NM1*PR") for s in segments), "missing NM1*PR" assert any(s.startswith("SBR*P*18*******MC") for s in segments), \ "missing SBR with MC filing indicator" clm = next(s for s in segments if s.startswith("CLM*")) assert clm.startswith("CLM*20260205-Q944140-01*333.62"), \ f"CLM01 must be DOS-member-idx (got {clm!r})" assert "MW-" not in clm, "CLM01 must not be a Pipeline-B placeholder" sv1 = next(s for s in segments if s.startswith("SV1*HC:")) assert sv1.startswith("SV1*HC:T1019:U1*333.62"), \ f"SV1 must carry HC:T1019:U1 (got {sv1!r})" dtp = next(s for s in segments if s.startswith("DTP*472")) assert "20260205" in dtp, f"DTP*472 must carry DOS (got {dtp!r})" assert any(s.startswith("HI*") for s in segments), "missing HI diagnoses" assert any(s.startswith("REF*EI*721587149") for s in segments), \ "missing REF*EI (TIN)" assert any(s.startswith("SE*") for s in segments), "missing SE trailer" assert any(s.startswith("GE*") for s in segments), "missing GE trailer" assert any(s.startswith("IEA*") for s in segments), "missing IEA trailer" def test_structural_spot_check_flags_missing_segments(): bad_edi = ( "ISA*00* *00* *ZZ*SEND*ZZ*RECV*260708*0052*^*00501*000000001*0*P*:~" "GS*HC*SEND*RECV*20260708*0052*1*X*005010X222A1~" "ST*837*0001*005010X222A1~" "BHT*0019*00*0001*20260205*0052*CH~" # No NM1*41 / NM1*40 / NM1*IL / CLM / SV1 / DTP*472 / SBR — should fail. "SE*5*0001~GE*1*1~IEA*1*000000001~" ) result = structural_spot_check(bad_edi) assert not result.passed # Every missing class shows up in `missing`. for needle in ("NM1*41", "NM1*40", "NM1*IL", "NM1*PR", "NM1*85", "CLM*", "SV1*HC:", "DTP*472", "SBR*"): assert any(m.startswith(needle) for m in result.missing), \ f"expected {needle!r} in missing={result.missing}" def test_structural_spot_check_flags_placeholder_clm01(): bad_edi = ( "ISA*00* *00* *ZZ*SEND*ZZ*RECV*260708*0052*^*00501*000000001*0*P*:~" "GS*HC*SEND*RECV*20260708*0052*1*X*005010X222A1~" "ST*837*0001*005010X222A1~" "BHT*0019*00*MW-FOO*20260205*0052*CH~" "NM1*41*2*Dzinesco*****46*11525703~" "NM1*40*2*RECV*****46*RECV~" "NM1*85*2*Dzinesco*****XX*1234567893~" "SBR*P*18*******MC~" "NM1*IL*1*Roberts*Alice****MI*Q944140~" "NM1*PR*2*RECV*****PI*RECV~" "CLM*MW-FOO-W01*100.00***11:B:1*Y*Y*Y*Y~" # Pipeline-B placeholder CLM01 "SV1*HC:T1019*100.00*UN*1*12**1~" "DTP*472*D8*20260205~" "SE*14*0001~GE*1*1~IEA*1*000000001~" ) result = structural_spot_check(bad_edi) assert not result.passed assert "CLM01-not-placeholder" in result.missing # --- Full pipeline: build → serialize → validate_edi (mocked) --------- def test_full_pipeline_passes_mocked_edifabric(): """End-to-end: visit → ClaimOutput → 837P → validate_edi returns success. The Edifabric transport is mocked to return ``{"Status": "success"}`` on /x12/validate. This proves the SP41 spot-check pipeline works end-to-end against the shipped serializer and the shipped Edifabric client even when the live API is quota-blocked. """ visit = _visit() claim = build_claim_output(visit) edi = serialize_837( claim, sender_id=SENDER_ID, receiver_id="CO_TXIX", submitter_name="Dzinesco", submitter_contact_name="Tyler Martinez", submitter_contact_email="tyler@dzinesco.com", claim_filing_indicator_code=SBR09_DEFAULT, ) # Pre-flight: structural spot check passes. assert structural_spot_check(edi).passed # Mock the /x12/validate transport to return success. success_payload = {"Status": "success", "Details": [], "LastIndex": 25} read_called = {"count": 0} validate_called = {"count": 0} def handler(request: httpx.Request) -> httpx.Response: if request.url.path.endswith("/x12/read"): read_called["count"] += 1 assert request.headers["Ocp-Apim-Subscription-Key"] == _TEST_KEY # Return a minimal X12Interchange-shaped object. return httpx.Response(200, json=[_minimal_x12_interchange()]) if request.url.path.endswith("/x12/validate"): validate_called["count"] += 1 return httpx.Response(200, json=success_payload) return httpx.Response(404, json={"error": f"unexpected path {request.url.path}"}) _install_factory(handler) result = edifabric.validate_edi(edi.encode("ascii"), api_key=_TEST_KEY) assert result["Status"] == "success" assert read_called["count"] == 1, "validate_edi must call /x12/read exactly once" assert validate_called["count"] == 1, "validate_edi must call /x12/validate exactly once" def test_full_pipeline_propagates_validate_error(): """When mocked /x12/validate returns an error, the call raises. Documents the failure path — if the live API returns a non-2xx (e.g. 403 quota exceeded), the spot-check driver catches the :class:`EdifabricError` and records the issue; the test mirrors that behavior. """ visit = _visit() claim = build_claim_output(visit) edi = serialize_837(claim, sender_id=SENDER_ID, claim_filing_indicator_code=SBR09_DEFAULT) def handler(request: httpx.Request) -> httpx.Response: if request.url.path.endswith("/x12/read"): return httpx.Response(200, json=[_minimal_x12_interchange()]) if request.url.path.endswith("/x12/validate"): return httpx.Response( 403, json={"statusCode": 403, "message": "Out of call volume quota."}, ) return httpx.Response(404) _install_factory(handler) with pytest.raises(EdifabricError) as excinfo: edifabric.validate_edi(edi.encode("ascii"), api_key=_TEST_KEY) assert excinfo.value.status_code == 403 assert "quota" in str(excinfo.value.body).lower() # --- Helpers ----------------------------------------------------------- _TEST_KEY = "test-spot-check-key-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" def _minimal_x12_interchange() -> dict: """A minimal X12Interchange-shaped dict that Edifabric /validate accepts. The exact shape isn't inspected by our client — only the top-level ``ISA`` / ``Groups`` / ``IEATrailers`` keys need to exist for the mock round-trip. """ return { "SegmentDelimiter": "~", "DataElementDelimiter": "*", "ISA": { "InterchangeControlNumber_13": "000000001", "SenderID_06": "11525703", "ReceiverID_08": "CO_TXIX", }, "Groups": [ { "FunctionalID_01": "HC", "GroupControlNumber_06": "1", "Transactions": [ { "ST_01": "837", "ControlNumber_02": "0001", "ImplementationConventionReference_03": "005010X222A1", "Segments": [], } ], } ], "IEATrailers": [{"InterchangeControlNumber_02": "000000001"}], }