Nora
b094231995
The persistence layer for the SP28 join table lands in
cyclone/store/claim_acks.py:
- add_claim_ack — insert + publish 'claim_ack_written' on the bus
(mirrors the publish-from-store pattern used by the ACKs paths)
- list_acks_for_claim / list_claims_for_ack — read helpers for the
two list endpoints
- find_ack_orphans — Inbox ack-orphans lane source: every ack row
whose ack_id has no ClaimAck row tied to it
- remove_claim_ack — unlink + publish 'claim_ack_dropped'
- batch_envelope_index — D10 in-memory map of
{envelope.control_number: batch.id}, called once per ingest (cost
is ~16 lookups today; trivially cheap)
Plus the to_ui_claim_ack serializer in store/ui.py mirroring
to_ui_ack shape — single source of truth for the wire format so the
live-tail event payload matches the list endpoint byte-for-byte.
Includes a single SELECT against claims for the claim_state field
(TA1 batch-level rows return 'n/a').
The CycloneStore facade in store/__init__.py re-exports all six
methods (add_claim_ack, list_acks_for_claim, list_claims_for_ack,
find_ack_orphans, remove_claim_ack, batch_envelope_index) plus
to_ui_claim_ack from the ui module.
Migration-version assertions in test_acks.py and test_db_migrate.py
bumped 17 → 18 to match the new migration head.
Steps 3.1/3.2/3.3/3.4 of the SP28 implementation plan.
285 lines
10 KiB
Python
285 lines
10 KiB
Python
"""Tests for the 999 ACK migration + ORM + store helpers (SP3 P3 T13)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
|
|
from cyclone import db, db_migrate
|
|
from cyclone.db import Ack
|
|
from cyclone.store import store
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _setup(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
|
|
db._reset_for_tests()
|
|
db.init_db()
|
|
yield
|
|
db._reset_for_tests()
|
|
|
|
|
|
def _make_batch(batch_id: str = "b-1") -> None:
|
|
"""Insert a stub batches row so the acks.source_batch_id FK resolves."""
|
|
from cyclone.db import Batch
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id=batch_id, kind="837p", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 20, 12, 0, tzinfo=timezone.utc),
|
|
))
|
|
s.commit()
|
|
|
|
|
|
def test_migration_0002_creates_acks_table():
|
|
"""On a fresh DB the migration runner must create the `acks` table."""
|
|
inspector = sa.inspect(db.engine())
|
|
assert "acks" in inspector.get_table_names()
|
|
# The index declared in 0002_acks.sql must also exist.
|
|
indexes = {ix["name"] for ix in inspector.get_indexes("acks")}
|
|
assert "ix_acks_source_batch_id" in indexes
|
|
# And the column list matches the spec.
|
|
cols = {c["name"] for c in inspector.get_columns("acks")}
|
|
expected = {
|
|
"id", "source_batch_id", "accepted_count", "rejected_count",
|
|
"received_count", "ack_code", "parsed_at", "raw_json",
|
|
}
|
|
assert expected <= cols # spec columns are all present
|
|
|
|
|
|
def test_migration_latest_idempotent_on_fresh_db():
|
|
"""Re-running the migration on the same DB must be a no-op (PRAGMA
|
|
user_version already at the latest version — currently 18 after
|
|
0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
|
|
providers/payers/clearhouse, SP10's 0008 payer_rejected,
|
|
SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
|
|
SP16's 0011 processed_inbound_files, SP17's 0012 db_backups,
|
|
SP-auth's 0013 users + sessions, SP-audit's 0014 audit_log.user_id,
|
|
SP22's 0015 drop_claims_unique_constraint, SP27-Task 11's 0016
|
|
claims.matched_remittance_id index, SP27-Task 17's 0017
|
|
claim.patient_control_number backfill UPDATE, SP28's 0018
|
|
claim_acks join table)."""
|
|
with db.engine().begin() as c:
|
|
v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
|
assert v1 == 18
|
|
# A second run should not raise and should not bump the version.
|
|
db_migrate.run(db.engine())
|
|
with db.engine().begin() as c:
|
|
v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
|
assert v2 == 18
|
|
|
|
|
|
def test_add_ack_persists_row():
|
|
_make_batch("b-1")
|
|
row = store.add_ack(
|
|
source_batch_id="b-1",
|
|
accepted_count=3,
|
|
rejected_count=1,
|
|
received_count=4,
|
|
ack_code="P",
|
|
raw_json={"envelope": {"control_number": "0001"}, "set_count": 1},
|
|
)
|
|
assert row.id is not None
|
|
assert row.source_batch_id == "b-1"
|
|
assert row.accepted_count == 3
|
|
assert row.rejected_count == 1
|
|
assert row.received_count == 4
|
|
assert row.ack_code == "P"
|
|
assert row.raw_json == {"envelope": {"control_number": "0001"}, "set_count": 1}
|
|
# Round-trip: fetch from the same session.
|
|
with db.SessionLocal()() as s:
|
|
loaded = s.get(Ack, row.id)
|
|
assert loaded is not None
|
|
assert loaded.ack_code == "P"
|
|
assert loaded.raw_json == {"envelope": {"control_number": "0001"}, "set_count": 1}
|
|
|
|
|
|
def test_list_acks_newest_first():
|
|
_make_batch("b-1")
|
|
# Insert two rows in a known order; the second must come first.
|
|
a1 = store.add_ack(
|
|
source_batch_id="b-1", accepted_count=1, rejected_count=0,
|
|
received_count=1, ack_code="A", raw_json={"order": 1},
|
|
)
|
|
a2 = store.add_ack(
|
|
source_batch_id="b-1", accepted_count=0, rejected_count=1,
|
|
received_count=1, ack_code="R", raw_json={"order": 2},
|
|
)
|
|
rows = store.list_acks()
|
|
assert len(rows) == 2
|
|
# Newest first means a2 should be at index 0 (auto-increment id desc).
|
|
assert rows[0].id == a2.id
|
|
assert rows[1].id == a1.id
|
|
assert rows[0].ack_code == "R"
|
|
assert rows[1].ack_code == "A"
|
|
|
|
|
|
def test_get_ack_returns_none_for_missing():
|
|
assert store.get_ack(99999) is None
|
|
|
|
|
|
def test_get_ack_returns_row_when_present():
|
|
_make_batch("b-1")
|
|
row = store.add_ack(
|
|
source_batch_id="b-1", accepted_count=1, rejected_count=0,
|
|
received_count=1, ack_code="A", raw_json={"hello": "world"},
|
|
)
|
|
fetched = store.get_ack(row.id)
|
|
assert fetched is not None
|
|
assert fetched.id == row.id
|
|
assert fetched.source_batch_id == "b-1"
|
|
assert fetched.raw_json == {"hello": "world"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hotfix 2026-06-29: acks list endpoint silently capped at limit=100
|
|
# without exposing the true total or full-set aggregates, so the Acks page
|
|
# rendered "100 on file" and KPI totals summed from the page only when the
|
|
# row count exceeded 100. These tests pin the new offset param + the
|
|
# server-side `aggregates` field so the silent-failure mode can't regress.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _seed_acks(n: int) -> list[int]:
|
|
"""Insert n ack rows with deterministic per-row counts. Returns the ids."""
|
|
_make_batch("b-hotfix")
|
|
ids: list[int] = []
|
|
for i in range(n):
|
|
row = store.add_ack(
|
|
source_batch_id="b-hotfix",
|
|
accepted_count=i + 1,
|
|
rejected_count=i,
|
|
received_count=i + 1,
|
|
ack_code="A",
|
|
raw_json={"order": i},
|
|
)
|
|
ids.append(row.id)
|
|
return ids
|
|
|
|
|
|
def test_list_acks_endpoint_pagination_offsets_correctly():
|
|
"""`offset` walks the full set, `has_more` flips at the boundary."""
|
|
from fastapi.testclient import TestClient
|
|
from cyclone.api import app
|
|
from cyclone.auth.users import create
|
|
from cyclone.db import SessionLocal
|
|
|
|
with SessionLocal()() as s:
|
|
create(s, username="u", password="p", role="admin")
|
|
s.commit()
|
|
client = TestClient(app)
|
|
client.post("/api/auth/login", json={"username": "u", "password": "p"})
|
|
|
|
_seed_acks(5)
|
|
|
|
r = client.get("/api/acks?limit=2&offset=0")
|
|
d = r.json()
|
|
assert d["total"] == 5
|
|
assert d["returned"] == 2
|
|
assert d["has_more"] is True
|
|
assert len(d["items"]) == 2
|
|
|
|
r = client.get("/api/acks?limit=2&offset=4")
|
|
d = r.json()
|
|
assert d["total"] == 5
|
|
assert d["returned"] == 1
|
|
assert d["has_more"] is False
|
|
assert len(d["items"]) == 1
|
|
|
|
|
|
def test_list_acks_endpoint_aggregates_reflect_full_set():
|
|
"""`aggregates` sums over the full row set, not the page.
|
|
|
|
Without this the Acks KPI strip silently under-reports once the
|
|
row count exceeds the page size — the silent-failure the operator
|
|
flagged on 2026-06-29.
|
|
"""
|
|
from fastapi.testclient import TestClient
|
|
from cyclone.api import app
|
|
from cyclone.auth.users import create
|
|
from cyclone.db import SessionLocal
|
|
|
|
with SessionLocal()() as s:
|
|
create(s, username="u", password="p", role="admin")
|
|
s.commit()
|
|
client = TestClient(app)
|
|
client.post("/api/auth/login", json={"username": "u", "password": "p"})
|
|
|
|
# 5 rows: accepted = 1+2+3+4+5 = 15, rejected = 0+1+2+3+4 = 10,
|
|
# received = same as accepted.
|
|
_seed_acks(5)
|
|
|
|
# Page 1 of 2 — full set is 5 rows, page only shows 2, but aggregates
|
|
# must reflect all 5.
|
|
r = client.get("/api/acks?limit=2&offset=0")
|
|
d = r.json()
|
|
assert d["total"] == 5
|
|
assert d["returned"] == 2
|
|
assert d["aggregates"]["accepted_count"] == 15
|
|
assert d["aggregates"]["rejected_count"] == 10
|
|
assert d["aggregates"]["received_count"] == 15
|
|
|
|
# Page 2 must return identical aggregates — page slice mustn't shift them.
|
|
r = client.get("/api/acks?limit=2&offset=2")
|
|
d = r.json()
|
|
assert d["aggregates"]["accepted_count"] == 15
|
|
assert d["aggregates"]["rejected_count"] == 10
|
|
assert d["aggregates"]["received_count"] == 15
|
|
|
|
|
|
def test_list_acks_endpoint_limit_cap_is_5000():
|
|
"""The validator still enforces an upper bound so a client can't
|
|
request 1,000,000 rows and OOM the SQLite-backed list call."""
|
|
from fastapi.testclient import TestClient
|
|
from cyclone.api import app
|
|
from cyclone.auth.users import create
|
|
from cyclone.db import SessionLocal
|
|
|
|
with SessionLocal()() as s:
|
|
create(s, username="u", password="p", role="admin")
|
|
s.commit()
|
|
client = TestClient(app)
|
|
client.post("/api/auth/login", json={"username": "u", "password": "p"})
|
|
|
|
r = client.get("/api/acks?limit=10000")
|
|
assert r.status_code == 422 # FastAPI validation error
|
|
|
|
r = client.get("/api/acks?limit=5000")
|
|
assert r.status_code == 200
|
|
|
|
|
|
def test_list_acks_endpoint_offset_past_end_returns_empty_page():
|
|
"""`offset` past the row count must yield an empty page, not 500.
|
|
|
|
Pinning this so a future refactor that introduces streaming or
|
|
cursor-based pagination can't accidentally error or 500 when the
|
|
UI holds stale page state across a row count change.
|
|
"""
|
|
from fastapi.testclient import TestClient
|
|
from cyclone.api import app
|
|
from cyclone.auth.users import create
|
|
from cyclone.db import SessionLocal
|
|
|
|
with SessionLocal()() as s:
|
|
create(s, username="u", password="p", role="admin")
|
|
s.commit()
|
|
client = TestClient(app)
|
|
client.post("/api/auth/login", json={"username": "u", "password": "p"})
|
|
|
|
_seed_acks(3)
|
|
|
|
r = client.get("/api/acks?limit=2&offset=99")
|
|
assert r.status_code == 200
|
|
d = r.json()
|
|
assert d["total"] == 3
|
|
assert d["returned"] == 0
|
|
assert d["has_more"] is False
|
|
assert d["items"] == []
|
|
# Aggregates must still reflect the full 3-row set on the empty page —
|
|
# otherwise a stale UI page state would silently zero out the KPI strip.
|
|
assert d["aggregates"]["accepted_count"] == 1 + 2 + 3
|
|
assert d["aggregates"]["rejected_count"] == 0 + 1 + 2
|
|
assert d["aggregates"]["received_count"] == 1 + 2 + 3
|