"""Tests that the store's ACK add methods publish on the event bus. Per SP25, the store is the single write surface; the handlers and parse endpoints no longer publish. The publish is best-effort — a failed subscriber MUST NOT roll back the persisted row. """ from __future__ import annotations import json from datetime import date, datetime, timezone from pathlib import Path import pytest from fastapi.testclient import TestClient from cyclone import db from cyclone.api import app from cyclone.pubsub import EventBus from cyclone.store import store @pytest.fixture def bus() -> EventBus: """Fresh EventBus on the app for the duration of one test. The autouse conftest already wires an EventBus onto ``app.state.event_bus`` per-test. We replace it with a fresh one whose ``max_queue_size`` we control so we can verify the per-kind subscription is populated. Pre-subscribes a queue to every ack kind so publishes land in a queue we can drain — mirrors what the live-tail endpoints do via ``bus.subscribe_raw(...)``. """ new_bus = EventBus(max_queue_size=64) new_bus.subscribe_raw(["ack_received"]) new_bus.subscribe_raw(["ta1_ack_received"]) new_bus.subscribe_raw(["two77ca_ack_received"]) app.state.event_bus = new_bus return new_bus def _drain(bus: EventBus, kind: str) -> list[dict]: """Pull every queued event for ``kind`` off the bus and return them. Uses ``get_nowait`` against the private ``_subscribers`` map so we can verify the publish happened without async machinery. """ events: list[dict] = [] for queue in bus._subscribers.get(kind, []): while not queue.empty(): events.append(queue.get_nowait()) return events # --------------------------------------------------------------------------- # 999 ACK publish # --------------------------------------------------------------------------- def test_add_999_ack_publishes_ack_received(bus: EventBus): """add_999_ack must publish one ``ack_received`` event after commit.""" with db.SessionLocal()() as s: row = store.add_ack( source_batch_id="999-SP25-1", accepted_count=2, rejected_count=0, received_count=2, ack_code="A", raw_json={ "envelope": {"control_number": "000000001"}, "set_responses": [{"set_control_number": "PCN-1"}], }, event_bus=bus, ) events = _drain(bus, "ack_received") assert len(events) == 1 payload = events[0] # The event payload MUST match the list-endpoint shape so a # live-tail consumer sees the same row shape as a JSON fetch. assert payload["_kind"] == "ack_received" assert payload["id"] == row.id assert payload["source_batch_id"] == "999-SP25-1" assert payload["accepted_count"] == 2 assert payload["rejected_count"] == 0 assert payload["received_count"] == 2 assert payload["ack_code"] == "A" # Patient control number surfaces through the UI serializer. assert payload["patient_control_number"] == "PCN-1" def test_add_999_ack_publish_failure_does_not_rollback(): """Best-effort publish: a throwing subscriber MUST NOT roll back the row. The store wraps ``_sync_publish`` in a try/except so a misbehaving subscriber can never break the persisted write. """ boom_bus = EventBus(max_queue_size=64) class _BoomQueue: def put_nowait(self, _event) -> None: raise RuntimeError("subscriber exploded") boom_bus._subscribers["ack_received"] = [_BoomQueue()] # type: ignore[list-item] with db.SessionLocal()() as s: row = store.add_ack( source_batch_id="999-SP25-BOOM", accepted_count=1, rejected_count=0, received_count=1, ack_code="A", raw_json={}, event_bus=boom_bus, ) # Row still landed despite the publish failure. assert store.get_ack(row.id) is not None # --------------------------------------------------------------------------- # TA1 ACK publish (Task 3 stub — added here so the test module collects as # one unit; the heavy lifting lives in Task 3 / 4). # --------------------------------------------------------------------------- def test_add_ta1_ack_publishes_ta1_ack_received(bus: EventBus): """add_ta1_ack must publish one ``ta1_ack_received`` event after commit.""" with db.SessionLocal()() as s: row = store.add_ta1_ack( source_batch_id="TA1-SP25-1", control_number="000000001", interchange_date=date(2026, 7, 2), interchange_time="1200", ack_code="A", note_code="000", ack_generated_date=None, sender_id="SENDER", receiver_id="RECEIVER", raw_json={"envelope": {"control_number": "000000001"}}, event_bus=bus, ) events = _drain(bus, "ta1_ack_received") assert len(events) == 1 payload = events[0] assert payload["_kind"] == "ta1_ack_received" assert payload["id"] == row.id assert payload["control_number"] == "000000001" assert payload["ack_code"] == "A" assert payload["sender_id"] == "SENDER" assert payload["receiver_id"] == "RECEIVER" assert payload["interchange_date"] == "2026-07-02" # --------------------------------------------------------------------------- # 277CA ACK publish (Task 4 stub) # --------------------------------------------------------------------------- def test_add_277ca_ack_publishes_two77ca_ack_received(bus: EventBus): """add_277ca_ack must publish one ``two77ca_ack_received`` event.""" with db.SessionLocal()() as s: row = store.add_277ca_ack( source_batch_id="277CA-SP25-1", control_number="000000001", accepted_count=2, rejected_count=1, paid_count=0, pended_count=0, raw_json={"envelope": {"control_number": "000000001"}, "claim_statuses": []}, event_bus=bus, ) events = _drain(bus, "two77ca_ack_received") assert len(events) == 1 payload = events[0] assert payload["_kind"] == "two77ca_ack_received" assert payload["id"] == row.id assert payload["control_number"] == "000000001" assert payload["accepted_count"] == 2 assert payload["rejected_count"] == 1 # --------------------------------------------------------------------------- # /api/parse-999 fires the event (Task 6 stub) # --------------------------------------------------------------------------- MINIMAL_999 = Path(__file__).parent / "fixtures" / "minimal_999.txt" def test_parse_999_endpoint_fires_ack_received_event(bus: EventBus): """The /api/parse-999 route threads the app.state event bus into add_999_ack. SP25: the parse endpoint no longer hand-rolls a publish. The store owns publish-from-store; the endpoint just threads the bus. """ client = TestClient(app) text = MINIMAL_999.read_text() resp = client.post( "/api/parse-999", files={"file": ("minimal_999.txt", text, "text/plain")}, ) assert resp.status_code == 200, resp.text events = _drain(bus, "ack_received") assert len(events) == 1 assert events[0]["_kind"] == "ack_received"