From b5b715d1c9757f7b8efb1cdfbf5b17d8fb851cba Mon Sep 17 00:00:00 2001 From: Nora Date: Wed, 8 Jul 2026 22:30:48 -0600 Subject: [PATCH] test(sp24): pure-unit tests for cyclone.reissue.core MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 15 cases covering the four public functions + the IG guard: - ig_correctness_check: * returns True when PATIENT_LOOP_DEFAULT_INCLUDED is False (default) * returns False when monkeypatched to True * emits a WARNING explaining the IG violation when tripped * parametrized binary pass/fail against [False, True] - parse_inputs: * AppleDouble metadata files (._foo.x12) are skipped * per-file parse failures land in errors[]; valid files survive * empty input dir → zero claims + zero errors (not an error) * returned objects are Pydantic ClaimOutput instances - emit_outputs: * filenames match the HCPF-spec tp-837P--1of1.x12 shape * creates the output dir if missing * emitted X12 has no HL*3 + HL*2 child_count=0 (the IG-correct shape for SBR02='18' — the original 999 rejection root cause) - write_summary_sidecar: one row per emitted file with claim_id + output_file + byte_size. - zip_outputs: * testzip() returns None (integrity OK); entries round-trip * parent dirs of the zip path are created on demand. Also bumps cyclone.reissue.core.ig_correctness_check to read PATIENT_LOOP_DEFAULT_INCLUDED live (via the module attribute, not the import-time name binding) so test monkeypatching takes effect immediately. --- backend/tests/test_reissue_core.py | 330 +++++++++++++++++++++++++++++ 1 file changed, 330 insertions(+) create mode 100644 backend/tests/test_reissue_core.py diff --git a/backend/tests/test_reissue_core.py b/backend/tests/test_reissue_core.py new file mode 100644 index 0000000..731bb12 --- /dev/null +++ b/backend/tests/test_reissue_core.py @@ -0,0 +1,330 @@ +"""SP24 — pure-unit tests for cyclone.reissue.core. + +The CLI wrapper (cyclone reissue-claims) is thin; the behaviour lives +in :mod:`cyclone.reissue.core`. These tests exercise the four public +functions directly without a CliRunner, mirroring the conventions in +``cyclone-tests`` (autouse conftest for DB isolation, no network, +``tmp_path`` for filesystem work). +""" +from __future__ import annotations + +import zipfile +from pathlib import Path + +import pytest + +from cyclone.parsers.payer import PayerConfig +from cyclone.reissue.core import ( + emit_outputs, + ig_correctness_check, + parse_inputs, + write_summary_sidecar, + zip_outputs, +) + +# Fixture reference — flat, module-level Path constant. Matches the +# convention used in test_api_999.py / test_serialize_837.py. +MINIMAL_837P = Path(__file__).parent / "fixtures" / "minimal_837p.txt" + + +# --------------------------------------------------------------------------- +# ig_correctness_check +# --------------------------------------------------------------------------- + + +def test_ig_correctness_check_returns_true_when_default_is_false(monkeypatch): + """Guard returns True (silent pass) when PATIENT_LOOP_DEFAULT_INCLUDED is False.""" + # Default state of the module — should pass. + assert ig_correctness_check() is True + + +def test_ig_correctness_check_returns_false_when_default_flipped(monkeypatch): + """Guard returns False (refuse to run) when the constant is monkeypatched to True.""" + import cyclone.parsers.serialize_837 as ser_mod + + monkeypatch.setattr(ser_mod, "PATIENT_LOOP_DEFAULT_INCLUDED", True) + assert ig_correctness_check() is False + + +def test_ig_correctness_check_logs_warning_when_default_flipped(monkeypatch, caplog): + """The guard emits a WARNING explaining the IG violation when tripped.""" + import logging + import cyclone.parsers.serialize_837 as ser_mod + + monkeypatch.setattr(ser_mod, "PATIENT_LOOP_DEFAULT_INCLUDED", True) + with caplog.at_level(logging.WARNING, logger="cyclone.reissue.core"): + assert ig_correctness_check() is False + assert any( + "PATIENT_LOOP_DEFAULT_INCLUDED" in rec.message for rec in caplog.records + ), f"expected WARNING about the constant; got {caplog.records!r}" + + +# --------------------------------------------------------------------------- +# parse_inputs +# --------------------------------------------------------------------------- + + +def test_parse_inputs_skips_apple_double_metadata(tmp_path: Path): + """`._foo.x12` AppleDouble files are skipped at the glob stage.""" + in_dir = tmp_path / "in" + in_dir.mkdir() + # One real file + one AppleDouble shadow. + (in_dir / "real.x12").write_text(MINIMAL_837P.read_text()) + (in_dir / "._real.x12").write_text(MINIMAL_837P.read_text(), encoding="utf-8") + + payer = PayerConfig.co_medicaid() + claims, errors = parse_inputs(in_dir, payer) + + # Exactly one claim (from real.x12), zero errors. + assert len(claims) == 1 + assert errors == [] + assert claims[0].claim_id == "CLM001" + + +def test_parse_inputs_tolerates_per_file_failures(tmp_path: Path): + """A malformed file lands in `errors`; the valid one survives in `claims`.""" + in_dir = tmp_path / "in" + in_dir.mkdir() + (in_dir / "good.x12").write_text(MINIMAL_837P.read_text()) + (in_dir / "bad.x12").write_text("NOT A REAL 837P FILE\n") + + payer = PayerConfig.co_medicaid() + claims, errors = parse_inputs(in_dir, payer) + + assert len(claims) == 1 + assert claims[0].claim_id == "CLM001" + assert len(errors) == 1 + bad_path, bad_msg = errors[0] + assert bad_path.name == "bad.x12" + assert bad_msg # any non-empty failure message + + +def test_parse_inputs_handles_empty_input_dir(tmp_path: Path): + """An empty input dir produces zero claims + zero errors (not an error).""" + in_dir = tmp_path / "in" + in_dir.mkdir() + + payer = PayerConfig.co_medicaid() + claims, errors = parse_inputs(in_dir, payer) + + assert claims == [] + assert errors == [] + + +def test_parse_inputs_returns_claim_models(tmp_path: Path): + """The returned claims are Pydantic ClaimOutput instances.""" + from cyclone.parsers.models import ClaimOutput + + in_dir = tmp_path / "in" + in_dir.mkdir() + (in_dir / "real.x12").write_text(MINIMAL_837P.read_text()) + + payer = PayerConfig.co_medicaid() + claims, _ = parse_inputs(in_dir, payer) + + assert len(claims) == 1 + assert isinstance(claims[0], ClaimOutput) + + +# --------------------------------------------------------------------------- +# emit_outputs +# --------------------------------------------------------------------------- + + +def _make_claims(tmp_path: Path) -> list: + """Helper: parse the minimal_837p fixture and return the ClaimOutput list.""" + in_dir = tmp_path / "in" + in_dir.mkdir() + (in_dir / "real.x12").write_text(MINIMAL_837P.read_text()) + payer = PayerConfig.co_medicaid() + claims, _ = parse_inputs(in_dir, payer) + assert len(claims) == 1 + return claims, payer + + +def test_emit_outputs_writes_hcpf_spec_filenames(tmp_path: Path): + """Emitted filenames match `tp-837P--1of1.x12`.""" + claims, payer = _make_claims(tmp_path) + out_dir = tmp_path / "out" + + written = emit_outputs( + claims, + payer_config=payer, + output_dir=out_dir, + sender_id="11525703", + receiver_id="COMEDASSISTPROG", + submitter_name="Dzinesco", + submitter_contact_name="Tyler Martinez", + submitter_contact_email="tyler@dzinesco.com", + receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM", + ) + + assert len(written) == 1 + name = written[0].name + assert name.startswith("tp11525703-837P-"), f"unexpected filename: {name}" + assert name.endswith("-1of1.x12"), f"unexpected suffix: {name}" + # 837P + 1-of-1 with 17-digit millisecond timestamp → tp11525703-837P-YYYYMMDDhhmmssSSS-1of1.x12 + assert len(name) == len("tp11525703-837P-") + 17 + len("-1of1.x12") + + +def test_emit_outputs_creates_output_dir(tmp_path: Path): + """emit_outputs creates the output dir if it doesn't exist.""" + claims, payer = _make_claims(tmp_path) + out_dir = tmp_path / "deep" / "nested" / "out" + assert not out_dir.exists() + + emit_outputs( + claims, + payer_config=payer, + output_dir=out_dir, + sender_id="11525703", + receiver_id="COMEDASSISTPROG", + submitter_name="Dzinesco", + submitter_contact_name="Tyler Martinez", + submitter_contact_email="tyler@dzinesco.com", + receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM", + ) + + assert out_dir.is_dir() + assert any(out_dir.iterdir()) + + +def test_emit_outputs_emits_ig_correct_patient_loop_omission(tmp_path: Path): + """Emitted X12 contains no HL*3 segment (the IG-correct shape for SBR02='18').""" + claims, payer = _make_claims(tmp_path) + out_dir = tmp_path / "out" + + written = emit_outputs( + claims, + payer_config=payer, + output_dir=out_dir, + sender_id="11525703", + receiver_id="COMEDASSISTPROG", + submitter_name="Dzinesco", + submitter_contact_name="Tyler Martinez", + submitter_contact_email="tyler@dzinesco.com", + receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM", + ) + + body = written[0].read_text() + # HL*3 absent → SBR02='18' claim has no 2000C patient loop. + assert "HL*3" not in body, ( + f"HL*3 present in emitted X12 — 2000C patient loop must be absent for SBR02='18'. " + f"Body: {body!r}" + ) + # HL*2 child count = 0 (not 1) — the IG-correct count. + assert "HL*2*1*22*0" in body, ( + f"HL*2 child count != 0 in emitted X12. Body: {body!r}" + ) + + +def test_write_summary_sidecar_emits_one_row_per_claim(tmp_path: Path): + """The _serialize_summary.json sidecar has one row per emitted file.""" + import json + + claims, payer = _make_claims(tmp_path) + out_dir = tmp_path / "out" + written = emit_outputs( + claims, + payer_config=payer, + output_dir=out_dir, + sender_id="11525703", + receiver_id="COMEDASSISTPROG", + submitter_name="Dzinesco", + submitter_contact_name="Tyler Martinez", + submitter_contact_email="tyler@dzinesco.com", + receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM", + ) + + summary_path = write_summary_sidecar(claims, written, out_dir) + assert summary_path.exists() + rows = json.loads(summary_path.read_text()) + assert len(rows) == 1 + assert rows[0]["claim_id"] == "CLM001" + assert rows[0]["byte_size"] > 0 + assert rows[0]["output_file"] == str(written[0]) + + +# --------------------------------------------------------------------------- +# zip_outputs +# --------------------------------------------------------------------------- + + +def test_zip_outputs_round_trip_integrity(tmp_path: Path): + """Written zip passes testzip(); entries round-trip with the right names.""" + in_dir = tmp_path / "in" + in_dir.mkdir() + (in_dir / "real.x12").write_text(MINIMAL_837P.read_text()) + payer = PayerConfig.co_medicaid() + claims, _ = parse_inputs(in_dir, payer) + out_dir = tmp_path / "out" + written = emit_outputs( + claims, + payer_config=payer, + output_dir=out_dir, + sender_id="11525703", + receiver_id="COMEDASSISTPROG", + submitter_name="Dzinesco", + submitter_contact_name="Tyler Martinez", + submitter_contact_email="tyler@dzinesco.com", + receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM", + ) + zip_path = tmp_path / "out.zip" + + zip_outputs(written, zip_path) + + assert zip_path.exists() + with zipfile.ZipFile(zip_path) as zf: + assert zf.testzip() is None # None = no corrupt entries + names = zf.namelist() + assert len(names) == 1 + assert names[0] == written[0].name + # Round-trip the body matches. + assert zf.read(names[0]) == written[0].read_bytes() + + +def test_zip_outputs_creates_parent_dirs(tmp_path: Path): + """zip_outputs creates the parent dir for the zip path.""" + in_dir = tmp_path / "in" + in_dir.mkdir() + (in_dir / "real.x12").write_text(MINIMAL_837P.read_text()) + payer = PayerConfig.co_medicaid() + claims, _ = parse_inputs(in_dir, payer) + out_dir = tmp_path / "out" + written = emit_outputs( + claims, + payer_config=payer, + output_dir=out_dir, + sender_id="11525703", + receiver_id="COMEDASSISTPROG", + submitter_name="Dzinesco", + submitter_contact_name="Tyler Martinez", + submitter_contact_email="tyler@dzinesco.com", + receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM", + ) + + zip_path = tmp_path / "deep" / "nested" / "out.zip" + zip_outputs(written, zip_path) + assert zip_path.exists() + + +# --------------------------------------------------------------------------- +# Parametrized IG guard against a synthetic "broken serializer" +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "broken_default, expected_pass", + [ + (False, True), # IG-correct shape — guard silent + (True, False), # IG-incorrect shape — guard fires + ], +) +def test_ig_correctness_check_parametrized( + monkeypatch, broken_default, expected_pass +): + """The guard is a binary pass/fail that follows PATIENT_LOOP_DEFAULT_INCLUDED.""" + import cyclone.parsers.serialize_837 as ser_mod + + monkeypatch.setattr(ser_mod, "PATIENT_LOOP_DEFAULT_INCLUDED", broken_default) + assert ig_correctness_check() is expected_pass \ No newline at end of file