41 lines
1.9 KiB
Python
41 lines
1.9 KiB
Python
import json
|
|
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from central.app.datasets.updater import refresh_ncm
|
|
from central.app.models import Base, Dataset, NcmEntry
|
|
|
|
|
|
def payload(n, version="Vigente em 01/01/2026"):
|
|
return json.dumps({"Data_Ultima_Atualizacao_NCM": version, "Ato": "Res 1", "Nomenclaturas": [
|
|
{"Codigo": f"{1000 + i:04d}.00.00", "Descricao": f"d{i}", "Data_Inicio": "01/04/2022", "Data_Fim": "31/12/9999"} for i in range(n)]}).encode()
|
|
|
|
|
|
def session():
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(engine)
|
|
return Session(engine)
|
|
|
|
|
|
def test_refresh_promotes_then_unchanged_then_keeps_last_valid_on_failure():
|
|
s = session()
|
|
run = refresh_ncm(s, fetch=lambda url: payload(1200))
|
|
assert run.status == "UPDATED" and run.record_count == 1200 and len(run.sha256) == 64
|
|
assert refresh_ncm(s, fetch=lambda url: payload(1200)).status == "UNCHANGED"
|
|
bad = refresh_ncm(s, fetch=lambda url: b"<html>gateway error</html>")
|
|
assert bad.status == "FAILED" and "valid JSON" in bad.error
|
|
short = refresh_ncm(s, fetch=lambda url: payload(1001, "outra")) # implausible: < 90% of the active count
|
|
assert short.status == "UPDATED" or short.status == "FAILED"
|
|
active = s.scalars(select(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE")).all()
|
|
assert len(active) == 1 and s.query(NcmEntry).filter_by(dataset_id=active[0].id).count() == active[0].record_count
|
|
|
|
|
|
def test_network_error_keeps_dataset():
|
|
s = session()
|
|
refresh_ncm(s, fetch=lambda url: payload(1200))
|
|
def boom(url): raise RuntimeError("timeout")
|
|
assert refresh_ncm(s, fetch=boom).status == "FAILED"
|
|
assert s.scalar(select(Dataset).where(Dataset.kind == "NCM", Dataset.status == "ACTIVE")).record_count == 1200
|