gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
from datetime import datetime, timezone
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from central.app.datasets import bootstrap_v1_extra as parsers
|
|
from central.app.datasets.service import DatasetService
|
|
from central.app.models import Base
|
|
|
|
ROOT = Path(__file__).parents[2] / "sources" / "raw" / "bootstrap_v1"
|
|
FILES = {
|
|
"cfop": ROOT / "IT_2023_002_v2_10_Tabela_CFOP_Descrição.xlsx",
|
|
"fcp": ROOT / "Tabela_Alíquotas_FCP_por_UF_2025-02_07.xlsx",
|
|
"anp": ROOT / "Tabela_Codigo_Produto_ANP-20241118.xlsx",
|
|
"ncm_utrib": ROOT / "IT2024.001v2.40 - Tabela NCM 2022 com Utrib_Comércio Exterior_vigência 01.10.26.xlsx",
|
|
"biofuel": ROOT / "IT 2025.004 v.1.20 - Tabela de Índice.xlsx",
|
|
"credpres": ROOT / "cCredPres_2026-06-22.xlsx",
|
|
}
|
|
pytestmark = pytest.mark.skipif(not all(p.exists() for p in FILES.values()), reason="bootstrap_v1 extra source files not present locally")
|
|
|
|
|
|
@lru_cache(maxsize=None)
|
|
def _payload(name):
|
|
parse_fn = {"cfop": parsers.parse_cfop, "fcp": parsers.parse_fcp, "anp": parsers.parse_anp,
|
|
"ncm_utrib": parsers.parse_ncm_utrib, "biofuel": parsers.parse_rtc_biofuel_index,
|
|
"credpres": parsers.parse_rtc_cred_pres}[name]
|
|
return parse_fn(FILES[name].read_bytes())
|
|
|
|
|
|
def _engine():
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
return engine
|
|
|
|
|
|
def test_cfop_parses_full_table_with_official_fields():
|
|
payload = _payload("cfop")
|
|
assert len(payload.entries) >= 500
|
|
sample = payload.entries[0]
|
|
assert set(sample) >= {"code", "title", "description", "effective_from", "ind_nfe", "ind_comb"}
|
|
|
|
|
|
def test_fcp_covers_all_states_and_df():
|
|
payload = _payload("fcp")
|
|
ufs = {e["uf"] for e in payload.entries}
|
|
assert len(ufs) == 27 # 26 states + DF
|
|
assert "BA" in ufs
|
|
|
|
|
|
def test_anp_has_no_invented_descriptions():
|
|
payload = _payload("anp")
|
|
assert len(payload.entries) >= 900
|
|
assert all(set(e) == {"cprod_anp"} for e in payload.entries) # source truly has no description column
|
|
|
|
|
|
def test_ncm_utrib_covers_thousands_of_ncms():
|
|
payload = _payload("ncm_utrib")
|
|
assert len(payload.entries) >= 10000
|
|
assert all(len(e["ncm"]) == 8 for e in payload.entries[:50])
|
|
|
|
|
|
def test_biofuel_index_separates_current_from_historical():
|
|
payload = _payload("biofuel")
|
|
current = [e for e in payload.entries if not e["is_historical"]]
|
|
historical = [e for e in payload.entries if e["is_historical"]]
|
|
assert current and historical
|
|
assert all(e["p_bio_percent"] is not None for e in current)
|
|
|
|
|
|
def test_credpres_preserves_rich_official_fields_not_just_code_description():
|
|
payload = _payload("credpres")
|
|
sample = payload.entries[0]
|
|
assert sample["legal_basis"]
|
|
assert "d_ini_vig_cbs" in sample and "aliquota_cbs_text" in sample
|
|
|
|
|
|
@pytest.mark.parametrize("domain,method,lookup", [
|
|
("cfop", "import_cfop", lambda s: s.cfop_lookup(code="1101")),
|
|
("fcp", "import_fcp", lambda s: [s.fcp_for_uf("BA")]),
|
|
("anp", "import_anp", lambda s: [s.anp_code_exists("110203073")]),
|
|
("ncm_utrib", "import_ncm_utrib", lambda s: [s.ncm_utrib_for_ncm("01012100")]),
|
|
("biofuel", "import_rtc_biofuel_index", lambda s: s.rtc_biofuel_for_anp("320102001")),
|
|
("credpres", "import_rtc_cred_pres", lambda s: s.rtc_cred_pres(code="1")),
|
|
])
|
|
def test_import_idempotent_and_lookup_smoke(domain, method, lookup):
|
|
engine = _engine()
|
|
with Session(engine) as session:
|
|
service = DatasetService(session)
|
|
payload = _payload(domain)
|
|
first = getattr(service, method)(payload, f"file://{domain}", datetime.now(timezone.utc))
|
|
second = getattr(service, method)(payload, f"file://{domain}", datetime.now(timezone.utc))
|
|
assert first.id == second.id
|
|
assert second.status == "ACTIVE"
|
|
assert second.record_count == len(payload.entries)
|
|
|
|
result = lookup(service)
|
|
assert result
|
|
assert all(item is not None and item is not False for item in result) or result == [True]
|
|
|
|
|
|
def test_dataset_versions_includes_all_new_kinds():
|
|
engine = _engine()
|
|
with Session(engine) as session:
|
|
service = DatasetService(session)
|
|
versions = service.dataset_versions()
|
|
for key in ("cfop", "fcp", "anp", "ncm_utrib", "rtc_biofuel_index", "rtc_ccredpres"):
|
|
assert key in versions
|
|
assert versions[key] is None
|
|
service.import_cfop(_payload("cfop"), "file://cfop", datetime.now(timezone.utc))
|
|
assert service.dataset_versions()["cfop"] is not None
|