From 35730fcf14ce4cfcd75e08ec62ed3481219d86cd Mon Sep 17 00:00:00 2001 From: Nora Date: Mon, 29 Jun 2026 10:46:29 -0600 Subject: [PATCH] feat(sp27): extract handle_277ca from scheduler.py into handlers/ --- backend/src/cyclone/handlers/handle_277ca.py | 122 ++++++++++++++++ backend/src/cyclone/scheduler.py | 98 +++---------- backend/tests/test_handlers_277ca.py | 141 +++++++++++++++++++ 3 files changed, 282 insertions(+), 79 deletions(-) create mode 100644 backend/src/cyclone/handlers/handle_277ca.py create mode 100644 backend/tests/test_handlers_277ca.py diff --git a/backend/src/cyclone/handlers/handle_277ca.py b/backend/src/cyclone/handlers/handle_277ca.py new file mode 100644 index 0000000..34b6666 --- /dev/null +++ b/backend/src/cyclone/handlers/handle_277ca.py @@ -0,0 +1,122 @@ +"""Handle a 277CA Claim Acknowledgment file (SP27 Task 4). + +Lifted verbatim from ``scheduler.py:_handle_277ca``. The handler owns +its own DB session, dispatches to ``parse_277ca_text``, persists the +277CA ack row, applies 277CA rejections to matched claims via +``inbox_state.apply_277ca_rejections``, and emits +``claim.payer_rejected`` audit events for each newly-stamped claim. + +The actor tag (``"277ca-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 +— the API migration drops the inline copy in Task 6. + +``claim.rejected_after_remit`` audit emission (when a 277CA rejects +a claim that already has ``matched_remittance_id`` set) is deferred +to SP27 Task 13. Today the handler only emits ``claim.payer_rejected``. +""" +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 two77ca_synthetic_source_batch_id +from cyclone.inbox_state_277ca import apply_277ca_rejections +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.parse_277ca import parse_277ca_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 277CA, persist ack + stamp payer-rejected claims. + + Args: + text: Raw 277CA document bytes (decoded). + source_file: Filename the 277CA came from. + event_bus: Optional pubsub handle. Reserved for the + FastAPI-endpoint migration in Task 6 — the scheduler + doesn't pass one. + + Returns: + ``(parser_used, claim_count)`` tuple where ``claim_count`` is + the number of STC statuses in the file (one per claim). + + Raises: + ValueError: on parser-level failure (wraps CycloneParseError). + """ + try: + result = parse_277ca_text(text, input_file=source_file) + except CycloneParseError as exc: + raise ValueError(f"277CA parse error: {exc}") from exc + + icn = result.envelope.control_number + synthetic_id = two77ca_synthetic_source_batch_id(icn) + accepted = sum( + 1 for s in result.claim_statuses if s.classification == "accepted" + ) + paid = sum( + 1 for s in result.claim_statuses if s.classification == "paid" + ) + rejected = sum( + 1 for s in result.claim_statuses if s.classification == "rejected" + ) + pended = sum( + 1 for s in result.claim_statuses if s.classification == "pended" + ) + + with db.SessionLocal()() as session: + row = cycl_store.add_277ca_ack( + source_batch_id=synthetic_id, + control_number=icn, + accepted_count=accepted, + rejected_count=rejected, + paid_count=paid, + pended_count=pended, + raw_json=json.loads(result.model_dump_json()), + ) + def _lookup(pcn: str): + return ( + session.query(db.Claim) + .filter(db.Claim.patient_control_number == pcn) + .first() + ) + apply_result = apply_277ca_rejections( + session, result, claim_lookup=_lookup, two77ca_id=row.id, + ) + if apply_result.matched: + for cid in apply_result.matched: + append_event(session, AuditEvent( + event_type="claim.payer_rejected", + entity_type="claim", + entity_id=cid, + payload={"source_batch_id": synthetic_id, "277ca_id": row.id}, + actor="277ca-parser-scheduler", + )) + session.commit() + + # TODO(sp27-task-6): bridge async EventBus.publish → sync caller + # (same gap as handle_999 + handle_ta1; Task 6 will fix when the + # FastAPI endpoints migrate to call these handlers). + if event_bus is not None: + publish = getattr(event_bus, "publish", None) + if callable(publish): + try: + publish("ack_received", { + "source_batch_id": synthetic_id, + "ack_code": "277CA", + "kind": "277CA", + }) + except Exception as exc: # noqa: BLE001 + log.warning("event_bus publish failed: %s", exc) + + return ("parse_277ca", len(result.claim_statuses)) diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index 140c40d..4b885bf 100644 --- a/backend/src/cyclone/scheduler.py +++ b/backend/src/cyclone/scheduler.py @@ -55,11 +55,9 @@ 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.handlers import handle_277ca as _handle_277ca from cyclone.handlers import handle_999 as _handle_999 from cyclone.handlers import handle_ta1 as _handle_ta1 -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 @@ -146,10 +144,10 @@ class SchedulerStatus: # persists its own DB rows. The scheduler records the outcome. # --------------------------------------------------------------------------- # -# 999 (SP27 Task 2) and TA1 (Task 3) now live in ``cyclone.handlers``; -# the 277CA and 835 handlers stay inline and are lifted in Tasks 4-5. -# The HANDLERS dict literal below references the alias imports so the -# wiring stays identical. +# 999 (Task 2), TA1 (Task 3), and 277CA (Task 4) now live in +# ``cyclone.handlers``; only the 835 handler stays inline and is +# lifted in Task 5. The HANDLERS dict literal below references the +# alias imports so the wiring stays identical. def _handle_835(text: str, source_file: str) -> tuple[str, int]: @@ -196,61 +194,11 @@ def _handle_835(text: str, source_file: str) -> tuple[str, int]: return "parse_835", len(result.claims) -def _handle_277ca(text: str, source_file: str) -> tuple[str, int]: - """Parse a 277CA, persist ack + stamp payer-rejected claims.""" - from cyclone.parsers.parse_277ca import parse_277ca_text - from cyclone.parsers.exceptions import CycloneParseError - - try: - result = parse_277ca_text(text, input_file=source_file) - except CycloneParseError as exc: - raise ValueError(f"277CA parse error: {exc}") from exc - - icn = result.envelope.control_number - synthetic_id = _277ca_synthetic_source_batch_id(icn) - accepted = sum( - 1 for s in result.claim_statuses if s.classification == "accepted" - ) - paid = sum( - 1 for s in result.claim_statuses if s.classification == "paid" - ) - rejected = sum( - 1 for s in result.claim_statuses if s.classification == "rejected" - ) - pended = sum( - 1 for s in result.claim_statuses if s.classification == "pended" - ) - - with db.SessionLocal()() as session: - row = cycl_store.add_277ca_ack( - source_batch_id=synthetic_id, - control_number=icn, - accepted_count=accepted, - rejected_count=rejected, - paid_count=paid, - pended_count=pended, - raw_json=json.loads(result.model_dump_json()), - ) - def _lookup(pcn: str): - return ( - session.query(db.Claim) - .filter(db.Claim.patient_control_number == pcn) - .first() - ) - apply_result = apply_277ca_rejections( - session, result, claim_lookup=_lookup, two77ca_id=row.id, - ) - if apply_result.matched: - for cid in apply_result.matched: - append_event(session, AuditEvent( - event_type="claim.payer_rejected", - entity_type="claim", - entity_id=cid, - payload={"source_batch_id": synthetic_id, "277ca_id": row.id}, - actor="277ca-parser-scheduler", - )) - session.commit() - return "parse_277ca", len(result.claim_statuses) +# The 277CA handler has moved to ``cyclone.handlers.handle_277ca`` +# (SP27 Task 4); the inline def was deleted. The HANDLERS dict literal +# below still references ``_handle_277ca`` because the import-alias +# line at the top of this module binds that name to the new function. +# Only the 835 handler stays inline (Task 5). # Map file_type → handler. Mirrors ROUTED_FILE_TYPES. @@ -265,23 +213,15 @@ HANDLERS: dict[str, Callable[[str, str], tuple[str, int]]] = { # --------------------------------------------------------------------------- # Light copies of helpers the API endpoints use, so the scheduler can -# run without depending on the FastAPI module. -# --------------------------------------------------------------------------- - - -def _277ca_synthetic_source_batch_id(interchange_control_number: str) -> str: - """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). - - TODO(sp27-task-4): delete this duplicate when ``_handle_277ca`` - moves. The canonical copy now lives in - ``cyclone.handlers._ack_id.two77ca_synthetic_source_batch_id``. - """ - return f"277CA-{(interchange_control_number or '').strip() or '000000001'}" +# Run without depending on the FastAPI module. +# +# Note: ``_277ca_synthetic_source_batch_id`` lived here as a +# scheduler-local duplicate of +# ``cyclone.handlers._ack_id.two77ca_synthetic_source_batch_id`` until +# SP27 Task 4 landed; the inline def has been deleted because +# ``handle_277ca`` now imports the canonical copy. The historical +# helper was the only remaining inline def in this section — the +# surviving inline handler is ``_handle_835`` (lifts in Task 5). # --------------------------------------------------------------------------- diff --git a/backend/tests/test_handlers_277ca.py b/backend/tests/test_handlers_277ca.py new file mode 100644 index 0000000..c24c92f --- /dev/null +++ b/backend/tests/test_handlers_277ca.py @@ -0,0 +1,141 @@ +"""Direct tests for the ``handle_277ca`` handler (SP27 Task 4). + +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. + +Note: ``claim.rejected_after_remit`` audit emission (when a 277CA +rejects a claim already matched to a remit) is deferred to SP27 +Task 13. For now we lock the existing claim.payer_rejected behavior +only. +""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from cyclone import db +from cyclone.audit_log import AuditEvent +from cyclone.handlers.handle_277ca import handle +from cyclone.parsers.exceptions import CycloneParseError + +ACCEPTED = Path(__file__).parent / "fixtures" / "minimal_277ca.txt" +REJECTED = Path(__file__).parent / "fixtures" / "minimal_277ca_rejected_only.txt" + + +def _seed_claim(claim_id: str, pcn: str) -> None: + with db.SessionLocal()() as s: + from cyclone.db import Claim + s.add(Claim( + id=claim_id, batch_id="BATCH-1", + patient_control_number=pcn, charge_amount=100.00, + )) + s.commit() + + +def test_handle_277ca_persists_ack_with_classification_counts(): + """Happy path: persist a 277CA ack row with the right counts. + + The minimal_277ca.txt fixture has 3 claim statuses: + - CLAIM001 → A3:19 → accepted + - CLAIM002 → A6:19 → rejected + - CLAIM003 → A8:19 → pended + + So accepted=1, rejected=1, pended=1, paid=0. + """ + text = ACCEPTED.read_text() + parser_used, claim_count = handle(text, source_file=ACCEPTED.name) + assert parser_used == "parse_277ca" + assert claim_count == 3 + + # Lock persistence: a row was added to two77ca_acks. + with db.SessionLocal()() as session: + from cyclone.db import Two77caAck + rows = ( + session.query(Two77caAck) + .filter_by(source_batch_id="277CA-000000123") + .all() + ) + assert len(rows) == 1 + row = rows[0] + assert row.control_number == "000000123" + assert row.accepted_count == 1 + assert row.rejected_count == 1 + assert row.pended_count == 1 + assert row.paid_count == 0 + + +def test_handle_277ca_accepted_only_no_inbox_change(): + """A 277CA with only accepted statuses (no rejections) leaves the + inbox unchanged — no claim.payer_rejected audit row, no + payer_rejected_at stamp on any claim. + """ + text = ACCEPTED.read_text() + # Seed all 3 claims so the 277CA can match them by PCN. + _seed_claim("c1", "CLAIM001") + _seed_claim("c2", "CLAIM002") + _seed_claim("c3", "CLAIM003") + + handle(text, source_file=ACCEPTED.name) + + with db.SessionLocal()() as session: + from cyclone.db import Claim + # CLAIM001 (accepted A3) — no payer_rejected_at + c1 = session.get(Claim, "c1") + assert c1.payer_rejected_at is None + # CLAIM002 (rejected A6) — stamped + c2 = session.get(Claim, "c2") + assert c2.payer_rejected_at is not None + assert c2.payer_rejected_status_code == "A6" + # CLAIM003 (pended A8) — no payer_rejected_at (A8 is pended, not rejected) + c3 = session.get(Claim, "c3") + assert c3.payer_rejected_at is None + # Activity-event audit row exists for the rejected one (c2) + from sqlalchemy import select + events = session.execute( + select(db.AuditLog.__table__.c.event_type).where( + db.AuditLog.__table__.c.entity_id == "c2" + ) + ).all() + event_types = [e[0] for e in events] + assert "claim.payer_rejected" in event_types + + +def test_handle_277ca_rejects_only_one_claim(): + """The rejected_only fixture has one A7:19 for CLAIM099. + Seed a matching claim and verify only it gets stamped. + """ + _seed_claim("c_rejected", "CLAIM099") + _seed_claim("c_unrelated", "OTHERPCN") + + text = REJECTED.read_text() + parser_used, claim_count = handle(text, source_file=REJECTED.name) + assert parser_used == "parse_277ca" + assert claim_count == 1 # one claim_status in the fixture + + with db.SessionLocal()() as session: + from cyclone.db import Claim, Two77caAck + # The matching claim is stamped as rejected. + c_rej = session.get(Claim, "c_rejected") + assert c_rej.payer_rejected_at is not None + assert c_rej.payer_rejected_status_code == "A7" + # The unrelated claim is untouched. + c_unr = session.get(Claim, "c_unrelated") + assert c_unr.payer_rejected_at is None + # The two77ca_ack row was persisted with the rejected count. + rows = ( + session.query(Two77caAck) + .filter_by(source_batch_id="277CA-000000789") + .all() + ) + assert len(rows) == 1 + assert rows[0].rejected_count == 1 + assert rows[0].accepted_count == 0 + + +def test_handle_277ca_raises_value_error_on_bad_x12(): + """Garbage that tokenize may accept but parse_277ca will reject.""" + bad = "ISA*00*bad~ST*277CA*0001~SE*2*0001~IEA*0*0~" + with pytest.raises((CycloneParseError, ValueError)): + handle(bad, source_file="bad.277ca")