"""Claim-id dedup at the SFTP pre-flight stage.""" from datetime import datetime, timezone from pathlib import Path from unittest.mock import MagicMock import pytest from cyclone.submission.core import submit_file from cyclone.store.exceptions import DuplicateClaimError from cyclone.store.submission_dedup import ( record_submission, check_duplicate, DEFAULT_WINDOW_DAYS, ) # Fixture file with a single CLM01 = "CLM001" claim (the minimal_837p # fixture is the source of truth; duplicated into submit-batch/single-claim.x12 # but both byte-equal so we pick the canonical minimal path here). _FIXTURE = Path(__file__).parent / "fixtures" / "minimal_837p.txt" _FIXTURE_CLAIM_ID = "CLM001" def test_check_duplicate_returns_none_for_first_submission(tmp_path): db = tmp_path / "test.db" assert check_duplicate("ORIG-001", db_url=f"sqlite:///{db}") is None def test_check_duplicate_raises_within_window(tmp_path): db = tmp_path / "test.db" record_submission("ORIG-002", submitted_at=datetime(2026, 6, 1), db_url=f"sqlite:///{db}") try: check_duplicate("ORIG-002", db_url=f"sqlite:///{db}", now=datetime(2026, 6, 15)) except DuplicateClaimError as e: assert e.claim_id == "ORIG-002" assert e.original_submission_at == datetime(2026, 6, 1) else: raise AssertionError("expected DuplicateClaimError") def test_check_duplicate_passes_after_window(tmp_path): db = tmp_path / "test.db" record_submission("ORIG-003", submitted_at=datetime(2026, 1, 1), db_url=f"sqlite:///{db}") # 100 days later — past the 30-day default window assert check_duplicate("ORIG-003", db_url=f"sqlite:///{db}", now=datetime(2026, 4, 11)) is None def test_default_window_is_30_days(): assert DEFAULT_WINDOW_DAYS == 30 # --------------------------------------------------------------------------- # # SP41 Task 9 integration: dedup guard wired into submit_file pre-flight. # --------------------------------------------------------------------------- # def test_submit_file_raises_on_duplicate_claim_id(tmp_path): """``submit_file`` raises ``DuplicateClaimError`` if any CLM01 in the file was submitted within the 30-day dedup window, and the DB write + SFTP factory are both skipped. Mirrors the DB-first, upload-second invariant from ``submit_file``: the dedup check sits between the payer-mismatch guard and the BatchRecord837 DB write, so a duplicate blocks both the DB write AND the SFTP upload. This test pins both invariants. The conftest autouse fixture wires a fresh per-test SQLite DB at ``tmp_path/test.db`` via the ``CYCLONE_DB_URL`` env var (so ``cycl_db.SessionLocal()`` points there) — same engine the Batch write would use, same engine the dedup helper reads. """ # 1. Pre-populate the dedup table with the claim_id from the # fixture at a timestamp within the 30-day window. ``db_url`` # is ignored inside the helper (signature-preserving shim) but # the argument is required by the current Task 8 signature, so # we pass the per-test DB URL for clarity. record_submission( _FIXTURE_CLAIM_ID, submitted_at=datetime.now(timezone.utc), db_url=f"sqlite:///{tmp_path}/test.db", ) # 2. Sentinel: if submit_file reaches the SFTP factory call, # fail loudly. ``submit_file`` raises DuplicateClaimError # BEFORE the factory is invoked, so the AssertionError here # only fires if the dedup guard is bypassed (regression). def _sftp_factory(_block): # pragma: no cover - regression only raise AssertionError( "SFTP factory must not be called when dedup blocks submit_file" ) sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"}) # 3. Expect the 409-class domain exception with structured # ``claim_id`` / ``original_submission_at`` attributes so the # operator can see which CLM01 tripped the guard. with pytest.raises(DuplicateClaimError) as exc_info: submit_file( _FIXTURE, sftp_block=sftp_block, actor="test", validate=True, sftp_client_factory=_sftp_factory, ) assert exc_info.value.claim_id == _FIXTURE_CLAIM_ID assert exc_info.value.original_submission_at is not None # 4. DB invariant: no Batch row was written (dedup blocks the # DB write, not just the SFTP upload). The autouse fixture # points cycl_db at tmp_path/test.db, so a fresh # ``db_mod.SessionLocal()()`` re-opens it. from cyclone import db as db_mod with db_mod.SessionLocal()() as session: batch_count = session.query(db_mod.Batch).count() assert batch_count == 0, ( "dedup guard must block the BatchRecord837 DB write; " f"found {batch_count} Batch row(s)" ) # --------------------------------------------------------------------------- # # SP41 Task 9 — router-level coverage: api_routers/submission.submit_batch # catches DuplicateClaimError BEFORE the generic Exception handler and # surfaces it as a per-file ``outcome="unexpected_error"`` row, with the # duplicated claim_id baked into ``error`` and ``batch_id=None`` (no DB # row written). The HTTP status code stays 200 — per-file failures live # in the JSON body, not in the status (see the top-of-module contract # in api_routers/submission.py). # --------------------------------------------------------------------------- # def test_submit_batch_per_file_duplicate_claim_classified_as_unexpected( tmp_path, monkeypatch, ): """Pre-record CLM001 (the fixture's only CLM01) then POST a batch containing one copy of the fixture. Asserts the router: - keeps the HTTP status at 200 (per-file failures don't change status) - classifies the row as ``outcome == "unexpected_error"`` (the special-case handler in api_routers/submission.py) - includes the duplicated ``claim_id`` ("CLM001") in ``error`` - sets ``batch_id is None`` (no BatchRecord837 row written) Does NOT monkeypatch ``submit_file`` — the test exercises the real pre-flight raise path so the router's special-case catch is what produces the unexpected_error classification. submit_file raises DuplicateClaimError BEFORE the SFTP factory is invoked, so the paramiko factory is never reached even though the client fixture flips ``sftp_block.stub`` to ``False`` (no live MFT needed). """ from fastapi.testclient import TestClient from cyclone import db as db_mod from cyclone.api import app from cyclone.store import store as cycl_store # 1. Per-test DB + clearhouse (mirror test_api_submit_batch.py's # ``client`` fixture inline so this test is self-contained). # The autouse conftest already wired CYCLONE_DB_URL + init_db; # we just need to seed the clearhouse row and flip stub=False. db_mod._reset_for_tests() db_mod.init_db() cycl_store.ensure_clearhouse_seeded() ch = cycl_store.get_clearhouse() cycl_store.update_clearhouse( ch.model_copy(update={ "sftp_block": ch.sftp_block.model_copy(update={"stub": False}), }), ) # 2. Pre-record the duplicate. CLM001 is the fixture's only CLM01; # ``record_submission`` is the same helper the dedup unit test # uses, so the window + exception shape match. record_submission( _FIXTURE_CLAIM_ID, submitted_at=datetime.now(timezone.utc), db_url=f"sqlite:///{tmp_path}/test.db", ) # 3. Stage a single batch-* dir containing one copy of the fixture. # Mirrors _stage_batch from test_api_submit_batch.py. batch_dir = tmp_path / "batch-dedup-claims" batch_dir.mkdir() (batch_dir / "dup-claim.x12").write_bytes(_FIXTURE.read_bytes()) # 4. Sentinel on submit_file's sftp_client_factory — submit_file # raises BEFORE the factory is invoked, so this AssertionError # only fires if the dedup guard is bypassed end-to-end. We # don't actually call submit_file directly here; the sentinel # is wired via monkeypatch on the function the router uses, so # if anything in the chain accidentally reaches the factory # call we'd see it. In practice submit_file raises and we never # get there — this is a regression tripwire only. def _sentinel_factory(_block): # pragma: no cover - regression only raise AssertionError( "SFTP factory must not be called: dedup blocks submit_file " "before the factory runs" ) monkeypatch.setattr( "cyclone.submission.core._default_sftp_factory", _sentinel_factory, ) # 5. Hit the endpoint. Per-file failures don't change the status # code → 200 even though one file was rejected. client = TestClient(app) resp = client.post( "/api/submit-batch", json={"ingest_dir": str(tmp_path), "validate": True}, ) assert resp.status_code == 200, resp.text body = resp.json() # 6. Counts: 1 file total, classified as failed. (UNEXPECTED_ERROR # rolls into the ``failed`` counter in api_routers/submission.py # — see the counter branch at the bottom of the per-file loop.) assert body["submitted"] == 0 assert body["skipped"] == 0 assert body["failed"] == 1 assert len(body["results"]) == 1 # 7. The single result row carries the special-case classification # (NOT a generic "runtime error: kaboom" — that's a different # test in test_api_submit_batch.py). The router's # DuplicateClaimError handler sets UNEXPECTED_ERROR + bakes the # claim_id into ``error``. row = body["results"][0] assert row["outcome"] == "unexpected_error" assert _FIXTURE_CLAIM_ID in row["error"], ( f"expected claim_id {_FIXTURE_CLAIM_ID!r} in error string; " f"got {row['error']!r}" ) # The router's handler does NOT construct a batch_id — the dedup # guard fires BEFORE the DB write, so no BatchRecord837 row exists. assert row["batch_id"] is None # 8. DB invariant (defense in depth): even though we trust the # router-level assertion above, pin that no Batch row leaked. with db_mod.SessionLocal()() as session: batch_count = session.query(db_mod.Batch).count() assert batch_count == 0, ( "dedup guard must block the BatchRecord837 DB write end-to-end; " f"found {batch_count} Batch row(s)" )