From 30e1add8a2cbc9428797737832df2517a5cc3961 Mon Sep 17 00:00:00 2001 From: Nora Date: Mon, 29 Jun 2026 10:38:32 -0600 Subject: [PATCH] feat(sp27): extract handle_ta1 from scheduler.py into handlers/ --- backend/src/cyclone/handlers/handle_ta1.py | 89 +++++++++++++++++++ backend/src/cyclone/scheduler.py | 40 ++------- backend/tests/test_handlers_ta1.py | 99 ++++++++++++++++++++++ 3 files changed, 197 insertions(+), 31 deletions(-) create mode 100644 backend/src/cyclone/handlers/handle_ta1.py create mode 100644 backend/tests/test_handlers_ta1.py diff --git a/backend/src/cyclone/handlers/handle_ta1.py b/backend/src/cyclone/handlers/handle_ta1.py new file mode 100644 index 0000000..cbadcd4 --- /dev/null +++ b/backend/src/cyclone/handlers/handle_ta1.py @@ -0,0 +1,89 @@ +"""Handle a TA1 Interchange Acknowledgment file (SP27 Task 3). + +Lifted verbatim from ``scheduler.py:_handle_ta1``. The handler owns +its own DB session, dispatches to ``parse_ta1_text``, persists the +interchange ack row in ``ta1_acks``, and returns a +``(parser_used, claim_count)`` tuple. + +Unlike the 999 handler, TA1 is envelope-only — there is one TA1 per +ISA/IEA interchange, no set-level (AK2) or claim-level matching. +The claim_count is always 1 (one TA1 ack row per file). + +The actor tag is implicit (no audit event here — TA1 doesn't tag +anything in the activity log; it's an infrastructure-level ack). +""" +from __future__ import annotations + +import json +import logging +from typing import Optional + +from cyclone import db +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.parse_ta1 import parse_ta1_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 TA1, persist the interchange ack row. + + Args: + text: Raw TA1 document bytes (decoded). + source_file: Filename the TA1 came from. Used for audit + attribution; the ``batches.id`` for TA1 rows is derived + internally from the parsed envelope's control number + (``TA1-{ICN}``). + 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. TA1 always returns + ``claim_count=1`` (one TA1 ack row per interchange file). + + Raises: + ValueError: on parser-level failure (wraps CycloneParseError). + """ + try: + result = parse_ta1_text(text, input_file=source_file) + except CycloneParseError as exc: + raise ValueError(f"TA1 parse error: {exc}") from exc + + with db.SessionLocal()() as session: + cycl_store.add_ta1_ack( + source_batch_id=result.source_batch_id, + control_number=result.ta1.control_number, + interchange_date=result.ta1.interchange_date, + interchange_time=result.ta1.interchange_time, + ack_code=result.ta1.ack_code, + note_code=result.ta1.note_code, + ack_generated_date=result.ta1.ack_generated_date, + sender_id=result.envelope.sender_id, + receiver_id=result.envelope.receiver_id, + raw_json=json.loads(result.model_dump_json()), + ) + session.commit() + + # TODO(sp27-task-6): bridge async EventBus.publish → sync caller + # (same gap as handle_999; both fixed when the FastAPI endpoints + # migrate to call these handlers instead of inline code). + if event_bus is not None: + publish = getattr(event_bus, "publish", None) + if callable(publish): + try: + publish("ack_received", { + "source_batch_id": result.source_batch_id, + "ack_code": result.ta1.ack_code, + "kind": "TA1", + }) + except Exception as exc: # noqa: BLE001 + log.warning("event_bus publish failed: %s", exc) + + return ("parse_ta1", 1) diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index 0b26802..140c40d 100644 --- a/backend/src/cyclone/scheduler.py +++ b/backend/src/cyclone/scheduler.py @@ -56,6 +56,7 @@ from cyclone.clearhouse import InboundFile, SftpClient from cyclone.db import ProcessedInboundFile from cyclone.edi.filenames import parse_inbound_filename 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, ) @@ -145,10 +146,10 @@ class SchedulerStatus: # persists its own DB rows. The scheduler records the outcome. # --------------------------------------------------------------------------- # -# 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. +# 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. def _handle_835(text: str, source_file: str) -> tuple[str, int]: @@ -252,33 +253,6 @@ def _handle_277ca(text: str, source_file: str) -> tuple[str, int]: return "parse_277ca", len(result.claim_statuses) -def _handle_ta1(text: str, source_file: str) -> tuple[str, int]: - """Parse a TA1, persist the interchange ack row.""" - from cyclone.parsers.parse_ta1 import parse_ta1_text - from cyclone.parsers.exceptions import CycloneParseError - - try: - result = parse_ta1_text(text, input_file=source_file) - except CycloneParseError as exc: - raise ValueError(f"TA1 parse error: {exc}") from exc - - with db.SessionLocal()() as session: - cycl_store.add_ta1_ack( - source_batch_id=result.source_batch_id, - control_number=result.ta1.control_number, - interchange_date=result.ta1.interchange_date, - interchange_time=result.ta1.interchange_time, - ack_code=result.ta1.ack_code, - note_code=result.ta1.note_code, - ack_generated_date=result.ta1.ack_generated_date, - sender_id=result.envelope.sender_id, - receiver_id=result.envelope.receiver_id, - raw_json=json.loads(result.model_dump_json()), - ) - session.commit() - return "parse_ta1", 1 - - # Map file_type → handler. Mirrors ROUTED_FILE_TYPES. HANDLERS: dict[str, Callable[[str, str], tuple[str, int]]] = { "999": _handle_999, @@ -302,6 +276,10 @@ def _277ca_synthetic_source_batch_id(interchange_control_number: str) -> str: 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'}" diff --git a/backend/tests/test_handlers_ta1.py b/backend/tests/test_handlers_ta1.py new file mode 100644 index 0000000..8804a78 --- /dev/null +++ b/backend/tests/test_handlers_ta1.py @@ -0,0 +1,99 @@ +"""Direct tests for the ``handle_ta1`` handler (SP27 Task 3). + +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 + +import pytest + +from cyclone import db +from cyclone.handlers.handle_ta1 import handle +from cyclone.parsers.exceptions import CycloneParseError + + +ACCEPTED = ( + "ISA*00* *00* *ZZ*SENDER *ZZ*RECEIVER " + "*260520*1750*^*00501*000000001*0*P*:~" + "TA1*000000001*20260520*1750*A*000*20260520~" + "IEA*1*000000001~" +) + +REJECTED = ( + "ISA*00* *00* *ZZ*COMEDASSISTPROG*ZZ*TP11525703 " + "*260520*1750*^*00501*000000001*0*P*:~" + "TA1*320293557*260520*2338*R*006~" + "IEA*0*000000001~" +) + + +def test_handle_ta1_persists_row_and_returns_count(): + """Accepted TA1 → parse_ta1, claim_count=1, persists ack row with ack_code=A.""" + parser_used, claim_count = handle(ACCEPTED, source_file="accepted.ta1") + assert parser_used == "parse_ta1" + assert claim_count == 1 + + # Lock the persistence half: a row was added to ta1_acks. + with db.SessionLocal()() as session: + rows = ( + session.query(db.Ta1Ack) + .filter_by(source_batch_id="TA1-000000001") + .all() + ) + assert len(rows) == 1 + assert rows[0].ack_code == "A" + assert rows[0].note_code == "000" + assert rows[0].control_number == "000000001" + + +def test_handle_ta1_rejected_persists_with_rejected_ack_code(): + """Rejected TA1 (R ack_code, non-zero note_code) → persists row, R ack_code. + + Note: source_batch_id is derived from the ISA control number + (``TA1-``), not the TA1 segment's internal ICN. Both + fixtures share ISA13=000000001, so source_batch_id is the same + string — but ack_code + note_code differentiate the two rows. + """ + parser_used, claim_count = handle(REJECTED, source_file="rejected.ta1") + assert parser_used == "parse_ta1" + assert claim_count == 1 + + # Lock the rejection half: ack_code="R" + note_code="006" persisted. + # (DB is reset per-test, so this fixture owns exactly one row.) + with db.SessionLocal()() as session: + rows = ( + session.query(db.Ta1Ack) + .filter_by(source_batch_id="TA1-000000001") + .all() + ) + assert len(rows) == 1 + assert rows[0].ack_code == "R" + assert rows[0].note_code == "006" + + +def test_handle_ta1_missing_segment_raises(): + """No TA1 segment → handler raises ValueError (wraps CycloneParseError).""" + text = ( + "ISA*00* *00* *ZZ*SENDER *ZZ*RECEIVER " + "*260520*1750*^*00501*000000001*0*P*:~" + "IEA*0*000000001~" + ) + with pytest.raises((CycloneParseError, ValueError)): + handle(text, source_file="empty.ta1") + + +def test_handle_ta1_does_not_create_batches_row(): + """TA1 is an envelope-only ack; no ``batches`` row should be created + (only ``ta1_acks``).""" + parser_used, claim_count = handle(ACCEPTED, source_file="accepted.ta1") + with db.SessionLocal()() as session: + # Confirm no batch was created. Source_batch_id starts with + # "TA1-" — search the batches table (which uses UUIDs) for any + # row that starts with "TA1-". + from sqlalchemy import select + stmt = select(db.Batch.__table__.c.id).where( + db.Batch.__table__.c.id.like("TA1-%") + ) + rows = session.execute(stmt).all() + assert rows == []