gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
92 lines
4.9 KiB
Python
92 lines
4.9 KiB
Python
"""Missão 5 / Fase 1: Reference Data Registry - DRAFT-first ingestion, admin approval, diff summary.
|
|
Never auto-activates (unlike the 9 legacy dataset kinds in central/app/datasets/service.py, which
|
|
this module never touches)."""
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from central.app.models import Base, Dataset
|
|
from central.app.reference_registry import RegistryError, approve_draft, diff_against_vigente, list_registry, register_draft, reject_draft
|
|
|
|
|
|
def engine():
|
|
e = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(e)
|
|
return e
|
|
|
|
|
|
def test_register_draft_never_auto_activates():
|
|
with Session(engine()) as s:
|
|
d = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=15156)
|
|
assert d.status == "DRAFT"
|
|
assert s.scalar(select(Dataset).where(Dataset.kind == "NCM", Dataset.status == "VIGENTE")) is None
|
|
|
|
|
|
def test_resubmitting_the_same_sha256_returns_the_existing_row_not_a_duplicate():
|
|
with Session(engine()) as s:
|
|
d1 = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=15156)
|
|
d2 = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1-again", record_count=99999)
|
|
assert d1.id == d2.id and d2.version_label == "v1" # untouched, not overwritten by the resubmission
|
|
assert len(list(s.scalars(select(Dataset)))) == 1
|
|
|
|
|
|
def test_approve_promotes_to_vigente_and_supersedes_the_previous_one():
|
|
with Session(engine()) as s:
|
|
first = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=100)
|
|
approve_draft(s, first.id, admin_name="admin1")
|
|
assert s.get(Dataset, first.id).status == "VIGENTE"
|
|
second = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="b" * 64, version_label="v2", record_count=101)
|
|
diff = diff_against_vigente(s, second)
|
|
assert diff["current_vigente"]["version_label"] == "v1" and diff["record_count_delta"] == 1 and not diff["is_first_version"]
|
|
approve_draft(s, second.id, admin_name="admin2")
|
|
assert s.get(Dataset, first.id).status == "OBSOLETO"
|
|
assert s.get(Dataset, second.id).status == "VIGENTE"
|
|
assert s.get(Dataset, second.id).approved_by == "admin2"
|
|
|
|
|
|
def test_approve_refuses_a_dataset_with_parse_errors():
|
|
with Session(engine()) as s:
|
|
d = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=1, parse_errors={"MINIMUM_ENTRIES": "too few"})
|
|
try:
|
|
approve_draft(s, d.id, admin_name="admin1")
|
|
assert False, "expected RegistryError"
|
|
except RegistryError as e:
|
|
assert "parse errors" in str(e)
|
|
assert s.get(Dataset, d.id).status == "DRAFT"
|
|
|
|
|
|
def test_approve_refuses_a_non_draft_dataset():
|
|
with Session(engine()) as s:
|
|
d = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=1)
|
|
approve_draft(s, d.id, admin_name="admin1")
|
|
try:
|
|
approve_draft(s, d.id, admin_name="admin2")
|
|
assert False, "expected RegistryError"
|
|
except RegistryError as e:
|
|
assert "only a DRAFT" in str(e)
|
|
|
|
|
|
def test_reject_never_touches_the_current_vigente():
|
|
with Session(engine()) as s:
|
|
first = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=1)
|
|
approve_draft(s, first.id, admin_name="admin1")
|
|
bad = register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="b" * 64, version_label="v2-bad", record_count=1)
|
|
reject_draft(s, bad.id, admin_name="admin2")
|
|
assert s.get(Dataset, bad.id).status == "REJECTED"
|
|
assert s.get(Dataset, first.id).status == "VIGENTE"
|
|
|
|
|
|
def test_diff_against_vigente_for_the_first_ever_version_of_a_kind():
|
|
with Session(engine()) as s:
|
|
d = register_draft(s, kind="CEST", provider="Confaz", source_url="https://x", sha256="c" * 64, version_label="v1", record_count=1051)
|
|
diff = diff_against_vigente(s, d)
|
|
assert diff["is_first_version"] is True and diff["current_vigente"] is None
|
|
|
|
|
|
def test_list_registry_orders_by_kind_then_newest_first():
|
|
with Session(engine()) as s:
|
|
register_draft(s, kind="NCM", provider="Siscomex", source_url="https://x", sha256="a" * 64, version_label="v1", record_count=1)
|
|
register_draft(s, kind="CEST", provider="Confaz", source_url="https://x", sha256="c" * 64, version_label="v1", record_count=1)
|
|
rows = list_registry(s)
|
|
assert {r.kind for r in rows} == {"NCM", "CEST"}
|