67 lines
3.1 KiB
Python
67 lines
3.1 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from central.app.datasets.ncm import parse_ncm_payload
|
|
|
|
|
|
PAYLOAD = b'{"Data_Ultima_Atualizacao_NCM":"2026-09-17","Ato":"Ato","Nomenclaturas":[{"Codigo":"08043000","Descricao":"Abacaxis frescos","Data_Inicio":"2022-01-01","Data_Fim":"9999-12-31"}]}'
|
|
|
|
|
|
def test_import_promotes_and_is_idempotent():
|
|
from central.app.datasets.service import DatasetService
|
|
from central.app.models import Base, Dataset, NcmEntry
|
|
|
|
engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
service = DatasetService(session)
|
|
first = service.import_payload(parse_ncm_payload(PAYLOAD), "https://source", datetime.now(timezone.utc), minimum_entries=1)
|
|
second = service.import_payload(parse_ncm_payload(PAYLOAD), "https://source", datetime.now(timezone.utc), minimum_entries=1)
|
|
assert first.id == second.id
|
|
assert session.get(Dataset, first.id).status == "ACTIVE"
|
|
assert session.query(NcmEntry).filter_by(dataset_id=first.id).count() == 1
|
|
|
|
|
|
def test_classifier_validates_ncm_and_never_proposes_fiscal_fields():
|
|
from central.app.classification import Classifier
|
|
|
|
class Repository:
|
|
def active_entries(self):
|
|
return [type("Entry", (), {"code": "08043000", "description": "Abacaxis frescos"})()]
|
|
|
|
def dataset_versions(self):
|
|
return {"ncm": "dataset-1"}
|
|
|
|
result = Classifier(Repository()).classify({"description": "abacaxi", "current_ncm": "0804.30.00"})
|
|
assert result["status"] == "EXACT_NCM"
|
|
assert result["classification"]["ncm"] == "08043000"
|
|
assert result["classification"]["cest"] is None
|
|
assert result["can_auto_apply"] is False
|
|
|
|
|
|
def test_api_auth_and_cache():
|
|
from central.app.main import create_app
|
|
from central.app.models import Base, Dataset, NcmEntry
|
|
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
dataset = Dataset(kind="NCM", provider="Siscomex", sha256="a" * 64, version_label="v1", act="act", source_url="url", status="ACTIVE", record_count=1)
|
|
session.add(dataset)
|
|
session.flush()
|
|
session.add(NcmEntry(dataset_id=dataset.id, code="08043000", description="Abacaxis frescos", valid_from="2022", valid_to="9999"))
|
|
session.commit()
|
|
app = create_app(session_factory=lambda: Session(engine), bootstrap_token="test-token")
|
|
client = TestClient(app)
|
|
payload = {"product": {"product_id": "1", "description": "abacaxi", "current_ncm": "08043000"}}
|
|
assert client.post("/v1/classify", json=payload).status_code == 401
|
|
headers = {"Authorization": "Bearer test-token"}
|
|
first = client.post("/v1/classify", json=payload, headers=headers).json()
|
|
second = client.post("/v1/classify", json=payload, headers=headers).json()
|
|
assert first["request_fingerprint"] == second["request_fingerprint"]
|
|
assert second["provenance"]["cache_hit"] is True
|