From d248a5f2824fda3e00c404f91aee072c6dce743d Mon Sep 17 00:00:00 2001 From: Nora Date: Mon, 29 Jun 2026 10:31:40 -0600 Subject: [PATCH] feat(sp27): extract handle_999 from scheduler.py into handlers/ --- backend/src/cyclone/handlers/handle_999.py | 129 +++++++++++++++++++ backend/src/cyclone/scheduler.py | 137 +++------------------ backend/tests/test_handlers_999.py | 123 ++++++++++++++++++ 3 files changed, 268 insertions(+), 121 deletions(-) create mode 100644 backend/src/cyclone/handlers/handle_999.py create mode 100644 backend/tests/test_handlers_999.py diff --git a/backend/src/cyclone/handlers/handle_999.py b/backend/src/cyclone/handlers/handle_999.py new file mode 100644 index 0000000..9bb807c --- /dev/null +++ b/backend/src/cyclone/handlers/handle_999.py @@ -0,0 +1,129 @@ +"""Handle a 999 Implementation Acknowledgment file (SP27 Task 2). + +Lifted verbatim from ``scheduler.py:_handle_999``. The handler owns +its own DB session, dispatches to ``parse_999_text``, applies 999 +rejections to any matched claims via ``inbox_state.apply_999_rejections``, +persists the ack row, and returns a ``(parser_used, claim_count)`` +tuple. + +The actor tag (``"999-parser-scheduler"``) is preserved so the audit +log keeps tracing back to the same source after extraction. Both +the FastAPI endpoint and the scheduler path (re)use this module — +see Task 6 for the API migration that drops the inline copy. + +``claim_count`` mirrors ``parsed.received`` — the count of AK2 +(set-level) responses in the 999, which is the authoritative +per-claim accept/reject signal (see ``_ack_id`` for why AK9 is +ignored). +""" +from __future__ import annotations + +import json +import logging +from typing import Optional + +from cyclone import db +from cyclone.audit_log import AuditEvent, append_event +from cyclone.handlers._ack_id import ( + ack_count_summary, + ack_synthetic_source_batch_id, +) +from cyclone.inbox_state import apply_999_rejections +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.parse_999 import parse_999_text +from cyclone.store import store as cycl_store + +log = logging.getLogger(__name__) + + +def handle( + text: str, + source_file: str, + *, + event_bus: Optional[object] = None, +) -> tuple[str, int]: + """Parse a 999, apply rejections, persist ack row. + + Args: + text: Raw 999 document bytes (decoded). + source_file: Filename the 999 came from. Used to derive a + unique synthetic ``batches.id`` (see ``_ack_id``). + event_bus: Optional pubsub handle. When provided, an + ``ack_received`` event is best-effort published. The + scheduler passes ``None``; the FastAPI endpoint will + be migrated to pass ``app.state.event_bus`` in Task 6. + + Returns: + ``(parser_used, claim_count)`` tuple. Matches the scheduler + ``_download_and_parse`` destructure; the HandleResult + migration happens in Task 7. + + Raises: + ValueError: on parser-level failure (wraps CycloneParseError). + """ + try: + result = parse_999_text(text, input_file=source_file) + except CycloneParseError as exc: + raise ValueError(f"999 parse error: {exc}") from exc + + received, accepted, rejected, ack_code = ack_count_summary(result) + icn = result.envelope.control_number + pcn = ( + result.set_responses[0].set_control_number + if result.set_responses else None + ) + synthetic_id = ack_synthetic_source_batch_id( + icn, pcn=pcn, source_filename=source_file, + ) + + with db.SessionLocal()() as session: + def _lookup(pcn: str): + return ( + session.query(db.Claim) + .filter_by(patient_control_number=pcn) + .first() + ) + rejection_result = apply_999_rejections( + session, result, claim_lookup=_lookup, + ) + if rejection_result.matched: + for cid in rejection_result.matched: + append_event(session, AuditEvent( + event_type="claim.rejected", + entity_type="claim", + entity_id=cid, + payload={"source_batch_id": synthetic_id}, + actor="999-parser-scheduler", + )) + row = cycl_store.add_ack( + source_batch_id=synthetic_id, + accepted_count=accepted, + rejected_count=rejected, + received_count=received, + ack_code=ack_code, + raw_json=json.loads(result.model_dump_json()), + ) + session.commit() + + # Best-effort pubsub publish. The scheduler doesn't pass an + # event_bus; the API migration in Task 6 will pass the FastAPI + # EventBus (which has an async ``publish``). This handler is + # sync, so an async-real EventBus returns an unawaited coroutine + # here — that's a known gap until Task 6 wires the publish + # through ``asyncio.run_coroutine_threadsafe`` or makes + # ``handle`` async. See TODO(sp27-task-6) below. + if event_bus is not None: + publish = getattr(event_bus, "publish", None) + if callable(publish): + # TODO(sp27-task-6): bridge async EventBus.publish → sync + # caller (see comment above). + try: + publish("ack_received", { + "source_batch_id": synthetic_id, + "ack_code": ack_code, + "kind": "999", + }) + except Exception as exc: # noqa: BLE001 + log.warning("event_bus publish failed: %s", exc) + + return ("parse_999", received) diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index 1f5724b..0b26802 100644 --- a/backend/src/cyclone/scheduler.py +++ b/backend/src/cyclone/scheduler.py @@ -55,7 +55,10 @@ from cyclone.audit_log import AuditEvent, append_event from cyclone.clearhouse import InboundFile, SftpClient from cyclone.db import ProcessedInboundFile from cyclone.edi.filenames import parse_inbound_filename -from cyclone.inbox_state import apply_999_rejections +from cyclone.handlers import handle_999 as _handle_999 +from cyclone.handlers._ack_id import ( + two77ca_synthetic_source_batch_id as _277ca_synthetic_source_batch_id, +) from cyclone.inbox_state_277ca import apply_277ca_rejections from cyclone.providers import SftpBlock @@ -141,60 +144,11 @@ class SchedulerStatus: # Per-file-type handlers. Each returns (parser_name, claim_count) and # persists its own DB rows. The scheduler records the outcome. # --------------------------------------------------------------------------- - - -def _handle_999(text: str, source_file: str) -> tuple[str, int]: - """Parse a 999, apply rejections, persist ack row. Returns (parser, count).""" - from cyclone.parsers.parse_999 import parse_999_text - from cyclone.parsers.exceptions import CycloneParseError - - try: - result = parse_999_text(text, input_file=source_file) - except CycloneParseError as exc: - raise ValueError(f"999 parse error: {exc}") from exc - - received, accepted, rejected, ack_code = _ack_count_summary(result) - icn = result.envelope.control_number - # The natural unique key for a 999 is the AK2 set_control_number - # (= the original claim's patient_control_number). Each 999 ack - # covers exactly one claim, so the PCN is 1:1 with the 999 and - # far more useful for the operator than the ISA interchange - # control number (Gainwell's MFT ships every 999 with the same - # default ICN, which used to collapse all 385 daily acks onto - # ``999-000000001``). Fall back to ICN → ``unknown`` if the AK2 is - # missing. - pcn = result.set_responses[0].set_control_number if result.set_responses else None - synthetic_id = _ack_synthetic_source_batch_id(icn, pcn=pcn, source_filename=source_file) - - with db.SessionLocal()() as session: - def _lookup(pcn: str): - return ( - session.query(db.Claim) - .filter_by(patient_control_number=pcn) - .first() - ) - rejection_result = apply_999_rejections( - session, result, claim_lookup=_lookup, - ) - if rejection_result.matched: - for cid in rejection_result.matched: - append_event(session, AuditEvent( - event_type="claim.rejected", - entity_type="claim", - entity_id=cid, - payload={"source_batch_id": synthetic_id}, - actor="999-parser-scheduler", - )) - row = cycl_store.add_ack( - source_batch_id=synthetic_id, - accepted_count=accepted, - rejected_count=rejected, - received_count=received, - ack_code=ack_code, - raw_json=json.loads(result.model_dump_json()), - ) - session.commit() - return "parse_999", received +# +# The 999 handler now lives in ``cyclone.handlers.handle_999`` (SP27 +# Task 2); the 277CA / TA1 / 835 handlers still live inline here and +# are lifted in Tasks 3-5. The dict literal below references the +# alias imports so the wiring stays identical. def _handle_835(text: str, source_file: str) -> tuple[str, int]: @@ -341,73 +295,14 @@ HANDLERS: dict[str, Callable[[str, str], tuple[str, int]]] = { # --------------------------------------------------------------------------- -def _ack_count_summary(result: Any) -> tuple[int, int, int, str]: - """Return (received, accepted, rejected, ack_code) for a 999. - - Mirrors the logic in ``cyclone.api._ack_count_summary`` but lives - here so the scheduler can run without importing the API module. - - Counts are derived from the **set-level** ``IK5`` responses - (one per AK2 in the 999), not the functional-group ``AK9`` — - Gainwell's MFT ships AK9 segments that contradict the per-set - IK5 (e.g. ``AK9*A*1*1*1`` with ``IK5*A``), so trusting AK9's - rejected count would over-report rejections. The set-level - IK5 is the authoritative per-claim accept/reject signal. - """ - sets = result.set_responses - received = len(sets) - accepted = sum(1 for s in sets if s.set_accept_reject.code == "A") - rejected = received - accepted - if rejected == 0: - code = "A" - elif accepted == 0: - code = "R" - else: - code = "P" - return (received, accepted, rejected, code) - - -def _ack_synthetic_source_batch_id( - interchange_control_number: str, - *, - pcn: str | None = None, - source_filename: str | None = None, -) -> str: - """Synthetic batches.id for a received 999 with no source batch. - - Gainwell's MFT ships every 999 with the same default ISA - interchange control number (``000000001``), so the ICN alone - collapses all daily acks onto one row. The AK2 - ``set_control_number`` (= the original claim's - patient_control_number) is per-batch — Gainwell's 999 - acks are per-batch, not per-claim, so a daily pull of 385 - 999s typically has only ~4 distinct PCNs. To make every - acks row distinguishable in the UI, the source_batch_id - always includes an 8-char hash of the inbound filename. - - Precedence for the human-readable part of the id - (column is VARCHAR(32)): - - 1. ``999-{pcn}-{hash8}`` if AK2 set_control_number is - present (the common case). 4+9+1+8 = 22 chars max. - 2. ``999-{icn}-{hash8}`` if no AK2 (envelope-only 999). - 3. ``999-{hash12}`` if no filename either (shouldn't - happen in production). - """ - import hashlib - short_hash = "" - if source_filename: - short_hash = hashlib.sha1(source_filename.encode("utf-8")).hexdigest()[:8] - if pcn and pcn.strip(): - return f"999-{pcn.strip()}-{short_hash}" - icn = (interchange_control_number or "").strip() or "000000001" - if short_hash: - return f"999-{icn}-{short_hash}" - return f"999-{short_hash or icn}" - - def _277ca_synthetic_source_batch_id(interchange_control_number: str) -> str: - """Synthetic batches.id for a received 277CA with no source batch.""" + """Synthetic batches.id for a received 277CA with no source batch. + + Kept here as a scheduler-local copy because ``_handle_277ca`` is + still inline (SP27 Task 4 will move it next to ``handle_999`` in + ``cyclone/handlers/``). For the 999 side, ``_handle_999`` is + already imported from ``cyclone.handlers.handle_999`` (Task 2). + """ return f"277CA-{(interchange_control_number or '').strip() or '000000001'}" diff --git a/backend/tests/test_handlers_999.py b/backend/tests/test_handlers_999.py new file mode 100644 index 0000000..14a42a4 --- /dev/null +++ b/backend/tests/test_handlers_999.py @@ -0,0 +1,123 @@ +"""Direct tests for the ``handle_999`` handler (SP27 Task 2). + +Locks the handler's contract independent of the scheduler lifecycle +so a regression in the scheduler wiring doesn't hide a regression +in the handler. +""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from cyclone import db +from cyclone.handlers.handle_999 import handle +from cyclone.parsers.exceptions import CycloneParseError + +ACCEPTED = Path(__file__).parent / "fixtures" / "minimal_999.txt" +REJECTED = Path(__file__).parent / "fixtures" / "minimal_999_rejected.txt" + + +def test_handle_999_persists_ack_row_and_returns_count(): + text = ACCEPTED.read_text() + parser_used, claim_count = handle(text, source_file=ACCEPTED.name) + assert parser_used == "parse_999" + assert claim_count == 1 # one AK2 in the happy-path fixture + # Lock the persistence half of the contract: a row was actually + # added to ``acks`` for this filename (source_batch_id encodes it). + from cyclone.handlers._ack_id import ack_synthetic_source_batch_id + expected_bsid = ack_synthetic_source_batch_id( + interchange_control_number="000000001", + pcn="0001", + source_filename=ACCEPTED.name, + ) + with db.SessionLocal()() as session: + rows = session.query(db.Ack).filter_by(source_batch_id=expected_bsid).all() + assert len(rows) == 1 + assert rows[0].received_count == 1 + assert rows[0].ack_code == "A" + + +def test_handle_999_rejected_persists_with_rejected_count(): + text = REJECTED.read_text() + parser_used, claim_count = handle(text, source_file=REJECTED.name) + assert parser_used == "parse_999" + assert claim_count == 1 + # Lock the rejected-count half: IK5="R" → ack_code = "R". + from cyclone.handlers._ack_id import ack_synthetic_source_batch_id + expected_bsid = ack_synthetic_source_batch_id( + interchange_control_number="000000002", + pcn="0001", + source_filename=REJECTED.name, + ) + with db.SessionLocal()() as session: + rows = session.query(db.Ack).filter_by(source_batch_id=expected_bsid).all() + assert len(rows) == 1 + assert rows[0].rejected_count == 1 + assert rows[0].ack_code == "R" + + +def test_handle_999_raises_value_error_on_bad_x12(): + # Garbage that tokenize may accept but parse_999 will reject. + bad = "ISA*00*bad~ST*999*0001~SE*2*0001~IEA*0*0~" + with pytest.raises((CycloneParseError, ValueError)): + handle(bad, source_file="bad.999") + + +def test_handle_999_distinct_filenames_get_distinct_synthetic_ids(): + """Two calls with the same PCN but different inbound filenames + produce distinct synthetic ``batches.id``s (the 8-char hash suffix + differs). The scheduler's dedup-by-filename needs this so a re-poll + doesn't collapse onto the same row.""" + from cyclone.handlers._ack_id import ack_synthetic_source_batch_id + + text = ACCEPTED.read_text() + _, _ = handle(text, source_file="a-999.x12") + _, _ = handle(text, source_file="b-999.x12") + + id_a = ack_synthetic_source_batch_id( + interchange_control_number="000000001", + pcn="0001", + source_filename="a-999.x12", + ) + id_b = ack_synthetic_source_batch_id( + interchange_control_number="000000001", + pcn="0001", + source_filename="b-999.x12", + ) + assert id_a != id_b + assert id_a.startswith("999-0001-") + assert id_b.startswith("999-0001-") + # And both rows are independently persisted (no collision). + with db.SessionLocal()() as session: + rows_a = session.query(db.Ack).filter_by(source_batch_id=id_a).all() + rows_b = session.query(db.Ack).filter_by(source_batch_id=id_b).all() + assert len(rows_a) == 1 + assert len(rows_b) == 1 + + +def test_handle_999_event_bus_parameter_is_optional(): + """Scheduler passes event_bus=None; api.py passes app.state.event_bus. + Both must work without crashing on publish.""" + text = ACCEPTED.read_text() + # Without event_bus + parser_used, claim_count = handle(text, source_file=ACCEPTED.name) + assert parser_used == "parse_999" + # With a stub event_bus — must accept but not crash on publish. + class _Stub: + def __init__(self) -> None: + self.calls: list[tuple[str, dict]] = [] + + def publish(self, kind: str, payload: dict) -> None: + self.calls.append((kind, payload)) + + spy = _Stub() + parser_used, claim_count = handle( + text, source_file=ACCEPTED.name, event_bus=spy, + ) + assert parser_used == "parse_999" + # The stub pattern is sync. Real EventBus is async and the handler + # is sync — Task 6 bridges that. For now, just verify the sync + # stub fires (or doesn't) without crashing. + # We don't assert the call count here because the async-vs-sync + # gap is owned by Task 6 — see TODO(sp27-task-6) in handle_999.