cyclone/backend/tests/test_handlers_999.py

"""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