110 lines
4.4 KiB
Python
110 lines
4.4 KiB
Python
"""Missão 6 — Bloco A: Testes de Higiene (Migrations e Inventário dos 9 Datasets)."""
|
|
import subprocess
|
|
from pathlib import Path
|
|
from alembic.config import Config
|
|
from alembic.script import ScriptDirectory
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def test_alembic_master_head_is_single():
|
|
"""Garante que na branch master o alembic possui uma única head: 0034_reference_data_registry."""
|
|
cfg = Config(str(ROOT / "central" / "alembic.ini"))
|
|
script = ScriptDirectory.from_config(cfg)
|
|
heads = script.get_heads()
|
|
assert heads == ["0034_reference_data_registry"], f"Esperado head única 0034, obtido: {heads}"
|
|
|
|
|
|
def test_feature_branch_migration_chains_to_0034():
|
|
"""Garante que na branch feature/cfop-ippt-write a migração 0035 encadeia exatamente em 0034."""
|
|
# Obter conteúdo de 0035 na branch feature/cfop-ippt-write via git show
|
|
result = subprocess.run(
|
|
["git", "show", "feature/cfop-ippt-write:central/alembic/versions/0035_cfop_ippt_write_mode.py"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(ROOT),
|
|
)
|
|
assert result.returncode == 0, f"Arquivo 0035_cfop_ippt_write_mode.py deve existir na branch feature/cfop-ippt-write: {result.stderr}"
|
|
content = result.stdout
|
|
assert 'revision = "0035_cfop_ippt_write_mode"' in content
|
|
assert 'down_revision = "0034_reference_data_registry"' in content
|
|
|
|
# Verifica que 0033 não existe mais na branch
|
|
res_old = subprocess.run(
|
|
["git", "show", "feature/cfop-ippt-write:central/alembic/versions/0033_cfop_ippt_write_mode.py"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(ROOT),
|
|
)
|
|
assert res_old.returncode != 0, "Migração 0033 não deve mais existir na branch feature"
|
|
|
|
|
|
def test_inventory_9_datasets_coverage():
|
|
"""Garante que os 9 datasets existentes no Hub estão integralmente catalogados."""
|
|
from central.app.datasets.inventory import DATASET_INVENTORY_9
|
|
kinds = [item.kind for item in DATASET_INVENTORY_9]
|
|
expected_kinds = ["NCM", "TIPI_IPI", "CEST", "BA_ST_MVA", "NFE_MOC_TABLES", "CFOP", "FCP", "ANP", "NCM_UTRIB"]
|
|
assert kinds == expected_kinds
|
|
for item in DATASET_INVENTORY_9:
|
|
assert item.name
|
|
assert item.provider
|
|
assert item.source_url.startswith("http")
|
|
assert item.version_label
|
|
assert item.expected_rows_approx > 0
|
|
assert item.act
|
|
assert item.validity_info
|
|
assert item.coverage
|
|
assert item.licence_terms
|
|
|
|
|
|
def test_sync_inventory_to_registry_and_listing():
|
|
"""Garante que datasets ACTIVE são reconhecidos pelo Reference Data Registry."""
|
|
from central.app.datasets.inventory import sync_inventory_to_registry
|
|
from central.app.reference_registry import list_registry, diff_against_vigente
|
|
from central.app.models import Dataset
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from central.app.models import Base
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
|
|
with Session(engine) as session:
|
|
# Cadastrar dataset com status antigo ACTIVE (sem licence_terms)
|
|
ds = Dataset(
|
|
kind="NCM", provider="Siscomex", sha256="abc123hash",
|
|
version_label="2026-09-28", act="", source_url="https://portalunico.siscomex.gov.br",
|
|
status="ACTIVE", record_count=15156,
|
|
)
|
|
session.add(ds)
|
|
session.commit()
|
|
|
|
# Antes do sync, campos novos eram vazios
|
|
assert ds.licence_terms is None
|
|
|
|
# Executar sincronização
|
|
stats = sync_inventory_to_registry(session)
|
|
assert stats["active_found"] == 1
|
|
assert stats["updated"] == 1
|
|
assert ds.licence_terms is not None
|
|
assert ds.act == "Resolução Gecex / MDIC"
|
|
|
|
# Listagem do registry deve listar datasets ACTIVE
|
|
listed = list_registry(session)
|
|
assert len(listed) == 1
|
|
assert listed[0].kind == "NCM"
|
|
|
|
# Diff contra DRAFT deve encontrar o ACTIVE como vigente
|
|
draft = Dataset(
|
|
kind="NCM", provider="Siscomex", sha256="diff123hash",
|
|
version_label="2026-10-01", act="Nova Resolução", source_url="url",
|
|
status="DRAFT", record_count=15200,
|
|
)
|
|
session.add(draft)
|
|
session.commit()
|
|
diff = diff_against_vigente(session, draft)
|
|
assert diff["current_vigente"] is not None
|
|
assert diff["current_vigente"]["sha256"] == "abc123hash"
|
|
assert diff["record_count_delta"] == 15200 - 15156
|
|
|