173 lines
7.6 KiB
Python
173 lines
7.6 KiB
Python
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
CEST_HTML = b"""
|
|
<html><body>
|
|
<table><tr><td>Ignored</td></tr></table>
|
|
<h2>ANEXO II</h2>
|
|
<table>
|
|
<tr><th>ITEM</th><th>CEST</th><th>NCM/SH</th><th>DESCRICAO</th></tr>
|
|
<tr><td>1.0</td><td>01.001.00</td><td>0804.30.00</td><td>Abacaxis</td></tr>
|
|
<tr><td>2.0</td><td>01.002.00</td><td>0804.30.00<br>0207.14.00</td><td>Frutas e aves</td></tr>
|
|
</table>
|
|
<h2>ANEXO XXVII</h2>
|
|
<table><tr><th>CEST</th><th>NCM</th></tr><tr><td>99.999.99</td><td>9999.99.99</td></tr></table>
|
|
</body></html>
|
|
"""
|
|
|
|
|
|
def test_cest_parser_accepts_only_covered_annex_tables_and_hashes_raw_html():
|
|
from central.app.datasets.cest import parse_cest_html
|
|
|
|
parsed = parse_cest_html(CEST_HTML)
|
|
|
|
assert parsed.version == "CV142_18"
|
|
assert parsed.sha256
|
|
assert parsed.entries[0].annex == "II"
|
|
assert parsed.entries[1].ncm_codes == ("08043000", "02071400")
|
|
assert len(parsed.entries) == 2
|
|
|
|
|
|
def test_cest_parser_fails_closed_when_an_annex_table_has_no_required_columns():
|
|
from central.app.datasets.cest import parse_cest_html
|
|
|
|
with pytest.raises(ValueError, match="required columns"):
|
|
parse_cest_html(b"<h2>ANEXO II</h2><table><tr><th>CEST</th></tr><tr><td>01.001.00</td></tr></table>")
|
|
|
|
|
|
def test_cest_parser_accepts_official_four_digit_ncm_headings():
|
|
from central.app.datasets.cest import parse_cest_html
|
|
|
|
parsed = parse_cest_html(b"<h2>ANEXO III</h2><table><tr><th>CEST</th><th>NCM/SH</th></tr><tr><td>02.001.00</td><td>2205</td></tr></table>")
|
|
|
|
assert parsed.entries[0].ncm_codes == ("2205",)
|
|
|
|
|
|
def test_cest_parser_ignores_official_editorial_rows_without_cest_data():
|
|
from central.app.datasets.cest import parse_cest_html
|
|
|
|
parsed = parse_cest_html(b"<h2>ANEXO II</h2><table><tr><th>CEST</th><th>NCM/SH</th></tr><tr><td>Nova redacao dada ao item.</td></tr><tr><td>01.001.00</td><td>3815.12.10</td></tr></table>")
|
|
|
|
assert [entry.code for entry in parsed.entries] == ["01.001.00"]
|
|
|
|
|
|
def test_cest_parser_preserves_chapter_ncm_prefixes_and_empty_official_ncm_cells():
|
|
from central.app.datasets.cest import parse_cest_html
|
|
|
|
parsed = parse_cest_html(b"<h2>ANEXO XXVI</h2><table><tr><th>CEST</th><th>NCM/SH</th></tr><tr><td>28.055.00</td><td>Capitulo 33</td></tr><tr><td>28.056.00</td><td></td></tr></table>")
|
|
|
|
assert parsed.entries[0].ncm_codes == ("33",)
|
|
assert parsed.entries[1].ncm_codes == ()
|
|
|
|
|
|
def test_classifier_distinguishes_exact_legacy_and_invalid_ncm_and_uses_cest_matches():
|
|
from central.app.classification import Classifier
|
|
|
|
class Repository:
|
|
def active_entries(self):
|
|
return [type("Entry", (), {"code": "08043000", "description": "Abacaxis"})()]
|
|
|
|
def cest_for_ncm(self, code):
|
|
return [type("Cest", (), {"code": "01.001.00", "description": "Abacaxis"})()] if code == "08043000" else []
|
|
|
|
def legacy_ncm(self, code):
|
|
return {"code": code, "source": "Siscomex", "source_date": "2026-09-17", "reason": "NCM not present in current official nomenclature; no official successor."} if code == "02071400" else None
|
|
|
|
def dataset_versions(self):
|
|
return {"ncm": "ncm-version", "cest": "cest-version"}
|
|
|
|
classifier = Classifier(Repository())
|
|
|
|
exact = classifier.classify({"current_ncm": "0804.30.00", "description": "abacaxi"})
|
|
legacy = classifier.classify({"current_ncm": "0207.14.00"})
|
|
invalid = classifier.classify({"current_ncm": "0000.00.00"})
|
|
|
|
assert exact["status"] == "EXACT_NCM"
|
|
assert exact["classification"]["cest"] == ["01.001.00"]
|
|
assert legacy["status"] == "LEGACY_NCM"
|
|
assert legacy["classification"]["ncm"] == "02071400"
|
|
assert legacy["legacy_ncm"]["reason"].endswith("no official successor.")
|
|
assert invalid["status"] == "INVALID_NCM"
|
|
|
|
|
|
def test_classifier_reports_confirmed_cest_only_for_one_distinct_official_candidate():
|
|
from central.app.classification import Classifier
|
|
|
|
class Repository:
|
|
def active_entries(self):
|
|
return [type("Entry", (), {"code": "08043000", "description": "Abacaxi"})()]
|
|
def cest_for_ncm(self, code):
|
|
return [type("Cest", (), {"code": "01.001.00", "description": "Abacaxi"})()]
|
|
def legacy_ncm(self, code): return None
|
|
def dataset_versions(self): return {"ncm": "n", "cest": "c"}
|
|
|
|
result = Classifier(Repository()).classify({"current_ncm": "08043000", "description": "Abacaxi"})
|
|
assert result["cest_status"] == "CEST_CONFIRMED"
|
|
|
|
|
|
def test_classifier_reports_candidates_for_multiple_distinct_cest_codes():
|
|
from central.app.classification import Classifier
|
|
|
|
class Repository:
|
|
def active_entries(self):
|
|
return [type("Entry", (), {"code": "08043000", "description": "Abacaxi"})()]
|
|
def cest_for_ncm(self, code):
|
|
return [type("Cest", (), {"code": "01.001.00", "description": "Abacaxi"})(), type("Cest", (), {"code": "01.002.00", "description": "Frutas"})()]
|
|
def legacy_ncm(self, code): return None
|
|
def dataset_versions(self): return {"ncm": "n", "cest": "c"}
|
|
|
|
result = Classifier(Repository()).classify({"current_ncm": "08043000", "description": "Abacaxi"})
|
|
assert result["cest_status"] == "CEST_CANDIDATES"
|
|
assert result["classification"]["cest"] == ["01.001.00", "01.002.00"]
|
|
|
|
|
|
def test_cest_import_creates_many_to_many_ncm_links_and_stores_html_metadata():
|
|
from central.app.datasets.cest import parse_cest_html
|
|
from central.app.datasets.service import DatasetService
|
|
from central.app.models import Base, CestEntry, CestNcmEntry, Dataset
|
|
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
dataset = DatasetService(session).import_cest_payload(
|
|
parse_cest_html(CEST_HTML), "https://www.confaz.fazenda.gov.br/legislacao/convenios/2018/CV142_18", datetime.now(timezone.utc)
|
|
)
|
|
|
|
assert dataset.kind == "CEST"
|
|
assert dataset.raw_sha256 == dataset.sha256
|
|
assert session.query(CestEntry).filter_by(dataset_id=dataset.id).count() == 2
|
|
assert session.query(CestNcmEntry).count() == 3
|
|
|
|
|
|
def test_cest_lookup_matches_official_ncm_prefix_links():
|
|
from central.app.datasets.cest import parse_cest_html
|
|
from central.app.datasets.service import DatasetService
|
|
from central.app.models import Base
|
|
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
service = DatasetService(session)
|
|
service.import_cest_payload(parse_cest_html(b"<h2>ANEXO III</h2><table><tr><th>CEST</th><th>NCM</th></tr><tr><td>02.001.00</td><td>2205</td></tr></table>"), "https://source", datetime.now(timezone.utc))
|
|
|
|
assert [entry.code for entry in service.cest_for_ncm("22051000")] == ["02.001.00"]
|
|
|
|
|
|
def test_cest_import_unions_repeated_official_cest_codes_into_one_link_set():
|
|
from central.app.datasets.cest import parse_cest_html
|
|
from central.app.datasets.service import DatasetService
|
|
from central.app.models import Base, CestEntry, CestNcmEntry
|
|
|
|
payload = parse_cest_html(b"<h2>ANEXO II</h2><table><tr><th>CEST</th><th>NCM</th></tr><tr><td>01.042.00</td><td>8421.32.00</td></tr><tr><td>01.042.00</td><td>8421.39.20</td></tr></table>")
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
dataset = DatasetService(session).import_cest_payload(payload, "https://source", datetime.now(timezone.utc))
|
|
|
|
assert session.query(CestEntry).filter_by(dataset_id=dataset.id).count() == 1
|
|
assert {link.ncm_code for link in session.query(CestNcmEntry).all()} == {"84213200", "84213920"}
|