cyclone/backend/tests/test_ack_orphan_summary.py

"""SP38: store-helper tests for ``find_ack_orphan_st02_summary``.

The summary is the per-ST02 breakdown the ``cyclone ack-orphans``
CLI commands consume. It enumerates every distinct orphan 999 ST02
+ ack count + whether a ``batches`` row already covers that ST02.

Tests live here (not in ``test_apply_claim_ack_links.py``) because
the helper is a sp38 surface, not an sp28 invariant. Keeping the
tests in a sibling file matches the ``cyclone-tests`` convention.

The autouse ``_auto_init_db`` fixture in ``conftest.py`` provides
a fresh ``tmp_path/test.db`` for every test — no manual DB setup
needed here.
"""
from __future__ import annotations

import json
from datetime import datetime, timezone

import pytest

from cyclone import db
from cyclone.db import Ack, Batch, ClaimAck
from cyclone.store import CycloneStore


def _now():
    """A single shared 'now' for deterministic test timestamps."""
    return datetime.now(timezone.utc)


def _ingest_999(s, st02: str, *, batch_id: str | None = None) -> int:
    """Insert a 999 ack row whose set_responses[0].set_control_number == st02.

    Returns the new ack id. The 999 is an orphan (no claim_acks row
    is inserted) so it surfaces in the summary.
    """
    raw = {
        "envelope": {
            "sender_id": "SUBMITTERID",
            "receiver_id": "RECEIVERID",
            "control_number": "000000099",
            "transaction_date": "2024-01-01",
            "implementation_guide": "005010X231A1",
        },
        "functional_group_acks": [],
        "set_responses": [
            {"set_control_number": st02, "transaction_set_identifier": "837",
             "ak2": {"functional_id_code": "837"}, "segment_errors": [],
             "set_accept_reject": {"code": "A"}}
        ],
        "summary": {"accepted_count": 1, "rejected_count": 0},
    }
    ack = Ack(
        source_batch_id=batch_id or f"999-{st02}-test",
        accepted_count=1,
        rejected_count=0,
        received_count=1,
        ack_code="A",
        parsed_at=_now(),
        raw_json=json.dumps(raw),
    )
    s.add(ack)
    s.flush()
    return ack.id


def _seed_batch(s, *, st02: str, batch_id: str | None = None) -> str:
    """Insert a batches row with the given ST02. Returns the batch id."""
    import uuid
    bid = batch_id or uuid.uuid4().hex
    s.add(Batch(
        id=bid,
        kind="837p",
        input_filename="test.837p",
        transaction_set_control_number=st02,
        parsed_at=_now(),
    ))
    s.flush()
    return bid


def test_summary_returns_per_st02_counts():
    """Two distinct orphan ST02s surface as two separate summary rows."""
    with db.SessionLocal()() as s:
        _ingest_999(s, "991102989")
        _ingest_999(s, "991102989")  # same ST02 twice
        _ingest_999(s, "991102988")
        s.commit()

    summary = CycloneStore().find_ack_orphan_st02_summary()
    by_st02 = {row["st02"]: row for row in summary}
    assert by_st02["991102989"]["ack_count"] == 2
    assert by_st02["991102988"]["ack_count"] == 1
    assert by_st02["991102989"]["has_batch"] is False
    assert by_st02["991102988"]["has_batch"] is False


def test_summary_sets_has_batch_true_when_batches_row_exists():
    """When a batches row has the orphan ST02, ``has_batch`` is True
    and ``batch_id`` matches the batches row's id."""
    with db.SessionLocal()() as s:
        bid = _seed_batch(s, st02="991102977")
        _ingest_999(s, "991102977", batch_id=bid)
        s.commit()

    summary = CycloneStore().find_ack_orphan_st02_summary()
    assert len(summary) == 1
    row = summary[0]
    assert row["st02"] == "991102977"
    assert row["ack_count"] == 1
    assert row["has_batch"] is True
    assert row["batch_id"] == bid


def test_summary_sorted_by_ack_count_descending():
    """The summary is sorted so the heaviest orphans surface first
    in the CLI output (matches the spec's intent that operators
    triage the biggest backlog first)."""
    with db.SessionLocal()() as s:
        for _ in range(3):
            _ingest_999(s, "991102988")
        _ingest_999(s, "991102987")
        for _ in range(5):
            _ingest_999(s, "991102989")
        s.commit()

    summary = CycloneStore().find_ack_orphan_st02_summary()
    counts = [row["ack_count"] for row in summary]
    assert counts == sorted(counts, reverse=True)
    assert counts[0] == 5  # 991102989 has the most


def test_summary_excludes_999s_that_have_a_claim_acks_link():
    """A 999 with a claim_acks row is NOT an orphan and must not
    appear in the summary."""
    with db.SessionLocal()() as s:
        linked_id = _ingest_999(s, "991102977")
        _ingest_999(s, "991102988")  # orphan, no link
        # Manually insert a claim_acks link for the first 999.
        s.add(ClaimAck(
            claim_id="CLM-1",
            batch_id="b1",
            ack_id=linked_id,
            ack_kind="999",
            ak2_index=0,
            set_control_number="991102977",
            set_accept_reject_code="A",
            linked_at=_now(),
            linked_by="auto",
        ))
        s.commit()

    summary = CycloneStore().find_ack_orphan_st02_summary()
    by_st02 = {row["st02"]: row for row in summary}
    assert "991102977" not in by_st02  # linked → not orphan
    assert by_st02["991102988"]["ack_count"] == 1


def test_summary_skips_999s_with_malformed_raw_json():
    """999s whose raw_json is missing set_responses or is malformed
    JSON are SKIPPED — they can't be reconciled, so they don't
    contribute to the summary."""
    with db.SessionLocal()() as s:
        # Valid orphan
        _ingest_999(s, "991102988")
        # Malformed JSON — raw_json is not parseable
        s.add(Ack(
            source_batch_id="999-malformed",
            accepted_count=1, rejected_count=0, received_count=1,
            ack_code="A",
            parsed_at=_now(),
            raw_json="{not valid json",
        ))
        # Empty set_responses
        s.add(Ack(
            source_batch_id="999-empty",
            accepted_count=1, rejected_count=0, received_count=1,
            ack_code="A",
            parsed_at=_now(),
            raw_json=json.dumps({"envelope": {}, "set_responses": [], "summary": {}}),
        ))
        s.commit()

    summary = CycloneStore().find_ack_orphan_st02_summary()
    by_st02 = {row["st02"]: row for row in summary}
    assert set(by_st02.keys()) == {"991102988"}
    assert by_st02["991102988"]["ack_count"] == 1


def test_summary_empty_when_no_orphans():
    """With zero orphans, the summary is an empty list."""
    summary = CycloneStore().find_ack_orphan_st02_summary()
    assert summary == []


# ---- Task 3: reconcile_orphan_st02s --------------------------------------- #


def test_reconcile_creates_synthetic_batches_for_missing_st02s():
    """``reconcile_orphan_st02s`` inserts one synthetic batch row per
    orphan ST02 that doesn't already have a batches row.

    Each synthetic row uses the sentinel ``input_filename`` and
    ``kind = '837p'`` so they're trivially distinguishable in
    queries (the spec: grep for ``<synthetic:orphan-reconcile>``
    to find every row this SP38 created).
    """
    with db.SessionLocal()() as s:
        _ingest_999(s, "991102989")
        _ingest_999(s, "991102988")
        s.commit()

    plan = CycloneStore().reconcile_orphan_st02s(dry_run=False)

    assert plan["created"] == 2
    assert plan["skipped"] == 0
    assert len(plan["synthetic_batch_ids"]) == 2

    with db.SessionLocal()() as s:
        rows = (
            s.query(Batch)
            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
            .all()
        )
        assert len(rows) == 2
        assert all(r.kind == "837p" for r in rows)
        st02s = {r.transaction_set_control_number for r in rows}
        assert st02s == {"991102989", "991102988"}


def test_reconcile_skips_st02s_that_already_have_a_batch():
    """``reconcile`` does NOT create a synthetic row for an ST02
    that already has a batches row — the operator's intent is to
    fill gaps, not duplicate coverage."""
    with db.SessionLocal()() as s:
        existing = _seed_batch(s, st02="991102977")
        _ingest_999(s, "991102977", batch_id=existing)
        _ingest_999(s, "991102988")  # orphan, no batch
        s.commit()

    plan = CycloneStore().reconcile_orphan_st02s(dry_run=False)

    assert plan["created"] == 1
    assert plan["skipped"] == 1

    with db.SessionLocal()() as s:
        synthetic = (
            s.query(Batch)
            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
            .all()
        )
        assert len(synthetic) == 1
        assert synthetic[0].transaction_set_control_number == "991102988"


def test_reconcile_is_idempotent():
    """Re-running reconcile after a successful pass is a no-op:
    ``created=0, skipped=N``."""
    with db.SessionLocal()() as s:
        _ingest_999(s, "991102989")
        s.commit()

    CycloneStore().reconcile_orphan_st02s(dry_run=False)
    plan2 = CycloneStore().reconcile_orphan_st02s(dry_run=False)

    assert plan2["created"] == 0
    assert plan2["skipped"] == 1

    with db.SessionLocal()() as s:
        n = (
            s.query(Batch)
            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
            .count()
        )
        assert n == 1


def test_reconcile_dry_run_does_not_write():
    """``dry_run=True`` returns the same plan shape but does not
    insert any rows. Operators use this to preview the reconcile
    before committing."""
    with db.SessionLocal()() as s:
        _ingest_999(s, "991102989")
        s.commit()

    plan = CycloneStore().reconcile_orphan_st02s(dry_run=True)

    assert plan["created"] == 1
    assert plan["skipped"] == 0

    with db.SessionLocal()() as s:
        n = (
            s.query(Batch)
            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
            .count()
        )
        assert n == 0  # no row inserted


def test_reconcile_records_ack_count_in_totals_json():
    """The synthetic row's ``totals_json`` includes ``ack_count`` so
    future operators can see the orphan weight without re-querying."""
    import json
    with db.SessionLocal()() as s:
        for _ in range(4):
            _ingest_999(s, "991102989")
        s.commit()

    CycloneStore().reconcile_orphan_st02s(dry_run=False)

    with db.SessionLocal()() as s:
        row = (
            s.query(Batch)
            .filter(Batch.transaction_set_control_number == "991102989")
            .one()
        )
        totals = json.loads(row.totals_json or "{}")
        assert totals.get("orphan_reconcile") is True
        assert totals.get("ack_count") == 4


def test_reconcile_sentinel_is_grep_discoverable():
    """``input_filename = '<synthetic:orphan-reconcile>'`` must be
    discoverable via a plain ``LIKE '<synthetic:%>'`` query so
    operators can find every sp38-created row without knowing the
    exact sentinel string. Pins the spec's D2 grep-discoverability
    invariant."""
    with db.SessionLocal()() as s:
        _ingest_999(s, "991102988")
        s.commit()

    CycloneStore().reconcile_orphan_st02s(dry_run=False)

    with db.SessionLocal()() as s:
        n = (
            s.query(Batch)
            .filter(Batch.input_filename.like("<synthetic:%>"))
            .count()
        )
        assert n == 1


def test_summary_skips_999s_with_non_dict_raw_json():
    """``_extract_999_st02`` must tolerate the un-dict shapes a JSON
    column can take (None, list) and never raise. Rows with
    non-dict raw_json are skipped — they have no ST02 to contribute
    to the summary regardless.

    Note: the SQLAlchemy JSON column rejects bytes at insert time
    (it must be JSON-serializable), so we don't exercise the
    ``bytes`` branch here — that's a defensive-programming guard for
    raw sqlite3 reads, not a normal ORM codepath.
    """
    with db.SessionLocal()() as s:
        # Valid orphan — contributes 1 row
        _ingest_999(s, "991102988")
        # None — should be skipped, not raise
        s.add(Ack(
            source_batch_id="999-none-raw",
            accepted_count=1, rejected_count=0, received_count=1,
            ack_code="A", parsed_at=_now(),
            raw_json=None,
        ))
        # list — not a dict; should be skipped (not raise)
        s.add(Ack(
            source_batch_id="999-list-raw",
            accepted_count=1, rejected_count=0, received_count=1,
            ack_code="A", parsed_at=_now(),
            raw_json=[1, 2, 3],
        ))
        s.commit()

    summary = CycloneStore().find_ack_orphan_st02_summary()
    by_st02 = {row["st02"]: row for row in summary}
    assert by_st02 == {"991102988": {"st02": "991102988", "ack_count": 1,
                                     "has_batch": False, "batch_id": None}}