Nora
92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
"""SP39: Resubmission model + idempotency.
|
|
|
|
Schema:
|
|
- 7 columns: id, claim_id, batch_id, resubmitted_at, source_corrected_path,
|
|
interchange_control_number, group_control_number
|
|
- Unique index on (claim_id, interchange_control_number) -> re-insert is rejected.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from cyclone import db as cycl_db
|
|
from cyclone.db import Resubmission
|
|
|
|
|
|
def test_resubmission_model_columns():
|
|
cols = {c.name for c in Resubmission.__table__.columns}
|
|
assert cols == {
|
|
"id", "claim_id", "batch_id", "resubmitted_at",
|
|
"source_corrected_path",
|
|
"interchange_control_number", "group_control_number",
|
|
}
|
|
|
|
|
|
def test_unique_index_on_claim_id_and_icn():
|
|
idx_names = {i.name for i in Resubmission.__table__.indexes}
|
|
assert "ux_resubmissions_claim_icn" in idx_names
|
|
ux = next(
|
|
i for i in Resubmission.__table__.indexes if i.name == "ux_resubmissions_claim_icn"
|
|
)
|
|
assert ux.unique is True
|
|
# The unique index covers both columns.
|
|
cols = {c.name for c in ux.columns}
|
|
assert cols == {"claim_id", "interchange_control_number"}
|
|
|
|
|
|
def test_duplicate_insert_raises_integrity_error():
|
|
"""Idempotency contract: a second insert with the same
|
|
(claim_id, interchange_control_number) is rejected at the DB layer."""
|
|
cycl_db.init_db()
|
|
now = datetime.now(timezone.utc)
|
|
with cycl_db.SessionLocal()() as s:
|
|
s.add(Resubmission(
|
|
claim_id="claim-x", batch_id="batch-y",
|
|
resubmitted_at=now,
|
|
source_corrected_path="/path/one",
|
|
interchange_control_number="000000001",
|
|
group_control_number="1",
|
|
))
|
|
s.commit()
|
|
with cycl_db.SessionLocal()() as s:
|
|
s.add(Resubmission(
|
|
claim_id="claim-x", batch_id="batch-y",
|
|
resubmitted_at=now,
|
|
source_corrected_path="/path/two",
|
|
interchange_control_number="000000001", # same ICN
|
|
group_control_number="1",
|
|
))
|
|
with pytest.raises(IntegrityError):
|
|
s.commit()
|
|
s.rollback()
|
|
|
|
|
|
def test_different_icn_on_same_claim_is_allowed():
|
|
"""Distinct ICNs (i.e. distinct pushes) are independent rows."""
|
|
cycl_db.init_db()
|
|
now = datetime.now(timezone.utc)
|
|
with cycl_db.SessionLocal()() as s:
|
|
s.add(Resubmission(
|
|
claim_id="claim-y", batch_id="batch-y",
|
|
resubmitted_at=now,
|
|
source_corrected_path="/path/one",
|
|
interchange_control_number="000000001",
|
|
group_control_number="1",
|
|
))
|
|
s.add(Resubmission(
|
|
claim_id="claim-y", batch_id="batch-y",
|
|
resubmitted_at=now,
|
|
source_corrected_path="/path/two",
|
|
interchange_control_number="000000002", # different ICN
|
|
group_control_number="2",
|
|
))
|
|
s.commit()
|
|
with cycl_db.SessionLocal()() as s:
|
|
from sqlalchemy import select
|
|
rows = s.execute(select(Resubmission).where(
|
|
Resubmission.claim_id == "claim-y"
|
|
)).scalars().all()
|
|
assert len(rows) == 2 |