94 lines
3.6 KiB
Python
94 lines
3.6 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.nfe_moc_tables import parse_moc_tables
|
|
from central.app.datasets.service import DatasetService
|
|
from central.app.models import Base
|
|
|
|
SOURCE = Path(__file__).parents[2] / "sources" / "raw" / "bootstrap_v1" / "moc7-anexo-i-leiaute-e-rv.pdf"
|
|
pytestmark = pytest.mark.skipif(not SOURCE.exists(), reason="official MOC PDF not present locally")
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _payload():
|
|
# Parsing this 153-page PDF is expensive (~40s); every test in this module
|
|
# shares one parse instead of re-running it per test.
|
|
return parse_moc_tables(SOURCE.read_bytes())
|
|
|
|
|
|
def test_parses_all_five_domains_with_expected_minimum_coverage():
|
|
payload = _payload()
|
|
from collections import Counter
|
|
counts = Counter(e.domain for e in payload.entries)
|
|
assert counts["ORIGEM_MERCADORIA"] == 9
|
|
assert counts["CST_ICMS"] >= 11
|
|
assert counts["CSOSN"] >= 10
|
|
assert counts["CST_PIS"] >= 29
|
|
assert counts["CST_COFINS"] >= 29
|
|
|
|
|
|
def test_every_entry_has_provenance_and_no_invented_text():
|
|
payload = _payload()
|
|
for entry in payload.entries:
|
|
assert entry.source_section.startswith("MOC7 p.")
|
|
assert entry.description.strip()
|
|
assert entry.code.strip()
|
|
|
|
|
|
def test_origem_codes_are_0_through_8():
|
|
payload = _payload()
|
|
codes = sorted(e.code for e in payload.entries if e.domain == "ORIGEM_MERCADORIA")
|
|
assert codes == [str(n) for n in range(9)]
|
|
|
|
|
|
def test_cst_icms_includes_split_group_40_41_50():
|
|
payload = _payload()
|
|
by_code = {e.code: e for e in payload.entries if e.domain == "CST_ICMS"}
|
|
assert "isenta" in by_code["40"].description.lower()
|
|
assert "tributada" in by_code["41"].description.lower()
|
|
assert "suspens" in by_code["50"].description.lower()
|
|
|
|
|
|
def test_reimport_is_idempotent():
|
|
payload = _payload()
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
service = DatasetService(session)
|
|
first = service.import_moc_tables(payload, "file://moc7.pdf", datetime.now(timezone.utc))
|
|
second = service.import_moc_tables(payload, "file://moc7.pdf", datetime.now(timezone.utc))
|
|
assert first.id == second.id
|
|
assert second.status == "ACTIVE"
|
|
|
|
|
|
def test_lookup_smoke_for_each_domain():
|
|
payload = _payload()
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
service = DatasetService(session)
|
|
service.import_moc_tables(payload, "file://moc7.pdf", datetime.now(timezone.utc))
|
|
|
|
assert len(service.fiscal_codes("ORIGEM_MERCADORIA")) == 9
|
|
assert service.fiscal_codes("ORIGEM_MERCADORIA", code="0")[0].code == "0"
|
|
assert service.fiscal_codes("CST_ICMS", code="00")
|
|
assert service.fiscal_codes("CSOSN", code="101")
|
|
assert service.fiscal_codes("CST_PIS", code="01")
|
|
assert service.fiscal_codes("CST_COFINS", code="01")
|
|
assert service.fiscal_codes("CST_ICMS", code="99") == []
|
|
|
|
|
|
def test_dataset_versions_includes_nfe_moc_tables_and_changes_hash_on_reimport():
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
service = DatasetService(session)
|
|
assert service.dataset_versions()["nfe_moc_tables"] is None
|
|
service.import_moc_tables(_payload(), "file://moc7.pdf", datetime.now(timezone.utc))
|
|
assert service.dataset_versions()["nfe_moc_tables"] is not None
|