"""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_no_longer_accepts_event_bus_kwarg(): """SP25: the handler returns ``(parser_used, claim_count)`` only. Pre-SP25 the handler accepted a keyword-only ``event_bus=`` that it tried to publish to directly — a sync caller invoking the real async ``EventBus.publish`` produced an unawaited coroutine that was silently swallowed by the bare ``except``. The store now owns publish-from-store; the handler's surface is reduced to (text, source_file) so the sync/async gap is closed. """ text = ACCEPTED.read_text() parser_used, claim_count = handle(text, source_file=ACCEPTED.name) assert parser_used == "parse_999" assert claim_count >= 1 # The handler MUST reject the legacy ``event_bus=`` kwarg so any # stale caller (e.g. the inline copy in api.py that Task 6 was # supposed to migrate) fails loudly during refactors rather than # silently dropping events on the floor. with pytest.raises(TypeError, match="event_bus"): handle(text, source_file=ACCEPTED.name, event_bus=None) def test_handle_999_publishes_via_store(): """SP25: the row → event chain goes through the store, not the handler. This is the regression test for the original bug: the handler tried to publish via ``event_bus.publish(...)`` synchronously, which silently dropped events because the real bus is async. With the store owning publish, every write surfaces a real ``ack_received`` event. """ from cyclone.pubsub import EventBus from cyclone.store import store bus = EventBus(max_queue_size=64) bus.subscribe_raw(["ack_received"]) text = ACCEPTED.read_text() # The handler itself takes NO bus — the operator (scheduler or # endpoint) threads the bus to the store. We can't plumb a bus # through handle(), but we can verify that *any* path that uses # the store's add_ack fires the event. This is enough to lock # the regression: pre-SP25 the handler would have published via # its own path and the store's publish would be a no-op. Post-SP25 # only the store path exists. with db.SessionLocal()() as _s: row = store.add_ack( source_batch_id="999-HANDLE-PATH", accepted_count=1, rejected_count=0, received_count=1, ack_code="A", raw_json={"envelope": {"control_number": "000000001"}}, event_bus=bus, ) queue = bus._subscribers["ack_received"][0] assert not queue.empty() event = queue.get_nowait() assert event["_kind"] == "ack_received" assert event["id"] == row.id