133 lines
8.1 KiB
Python
133 lines
8.1 KiB
Python
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gti_fiscal_agent.company_context import resolve
|
|
from gti_fiscal_agent.preflight import major_minor, run_preflight
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
spec = importlib.util.spec_from_file_location("run_agent_service", ROOT / "agent" / "scripts" / "run_agent_service.py")
|
|
service = importlib.util.module_from_spec(spec); sys.modules[spec.name] = service; spec.loader.exec_module(service)
|
|
|
|
CN = "11111111000111"
|
|
|
|
|
|
def env(tmp_path, **over):
|
|
base = {"GTI_HUB_URL": "http://hub.test", "AGENT_TOKEN": "SECRETTOKENVALUE9", "FIREBIRD_DATABASE_PATH": "x.fdb", "FB_USER": "u", "FB_PASSWORD": "SECRETPASSVALUE9", "FISCAL_MODE": "DRY_RUN", "FIREBIRD_WRITES": "0",
|
|
"AGENT_STATE_DB": str(tmp_path / "state.sqlite"), "AGENT_LOG_PATH": str(tmp_path / "logs" / "a.log"), "AGENT_VERSION": "1.0.0-rc1"}
|
|
base.update(over)
|
|
return base
|
|
|
|
|
|
def http_ok(url, token=None, timeout=10):
|
|
if url.endswith("/health"):
|
|
return 200, {"status": "ok"}
|
|
if url.endswith("/version"):
|
|
return 200, {"version": "1.0.0-rc1"}
|
|
return (200 if token == "SECRETTOKENVALUE9" else 401), {}
|
|
|
|
|
|
def fake_probe(company_over=None, caps_missing=()):
|
|
def probe(env):
|
|
cols = {"CLASSIFICACAO_FISCAL": (37, 20), "CFOP_ENTRADA": (14, 4), "CST_ENTRADA": (14, 4), "CFOP": (37, 4), "CST": (37, 10), "ALIQUOTA": (16, 8), "CEST": (37, 7), "SITUACAO_TRIBUTARIA": (37, 2),
|
|
"COD_ENTRADA_COFINS": (37, 3), "COD_SAIDA_COFINS": (37, 3), "ALIQ_COFINS": (37, 8), "COD_ENTRADA_PIS": (37, 3), "COD_SAIDA_PIS": (37, 3), "ALIQ_PIS": (37, 8), "CSOSN": (37, 60), "IPI": (27, 8),
|
|
"CST_IPI": (14, 3), "IAT": (37, 1), "INCIDENCIA_PISCOFINS": (37, 30), "PISCOFINS": (37, 1), "IPPT": (37, 1), "EX_TIPI": (37, 3), "COD_ORIGEM_MERCADORIA": (8, 4)}
|
|
for c in caps_missing:
|
|
cols.pop(c)
|
|
ev = {"c000004": [{"code": "000001", "cnpj": CN, "uf": "BA", "crt": "1", "optante": "S"}], "emissor": [{"cnpj": CN, "uf": "BA", "crt": "1"}], "nfce_keys": [{"cuf": "29", "cnpj": CN, "n": 5}], "nf_by_filial": {}}
|
|
ev.update(company_over or {})
|
|
return {"server_version": "WI-V6.3.9 Firebird 2.5", "ods": "11.2", "tables": {"C000004": {}, "C000025": cols}, "products": 3712, "company": resolve(ev, {})}
|
|
return probe
|
|
|
|
|
|
def test_preflight_passes_for_a_clean_dry_run_installation(tmp_path):
|
|
r = run_preflight(env(tmp_path), http_get=http_ok, probe=fake_probe())
|
|
assert r["PREFLIGHT"] == "PASS" and r["blockers"] == [] and r["effective_firebird_writes"] is False
|
|
names = {c["name"] for c in r["checks"]}
|
|
assert {"HUB_REACHABLE", "AUTH", "HUB_VERSION_COMPATIBLE", "FIREBIRD_REACHABLE", "C000004", "C000025", "CAPABILITIES", "COMPANY_CONTEXT", "UF", "CRT", "SCHEMA_TYPE_GUARDS", "SQLITE", "LOG_DIRECTORY", "FISCAL_MODE", "FIREBIRD_WRITES"} <= names
|
|
assert "SECRETTOKENVALUE9" not in str(r) and "SECRETPASSVALUE9" not in str(r)
|
|
|
|
|
|
@pytest.mark.parametrize("over,blocker", [
|
|
({"FISCAL_MODE": "DRY_RUN", "FIREBIRD_WRITES": "1"}, "FIREBIRD_WRITES"),
|
|
({"FISCAL_MODE": ""}, "FISCAL_MODE"),
|
|
({"AGENT_TOKEN": ""}, "CONFIG"),
|
|
])
|
|
def test_preflight_blocks_bad_config(tmp_path, over, blocker):
|
|
r = run_preflight(env(tmp_path, **over), http_get=http_ok, probe=fake_probe())
|
|
assert r["PREFLIGHT"] == "FAIL" and any(b.startswith(blocker) for b in r["blockers"])
|
|
|
|
|
|
def test_preflight_fails_on_hub_auth_version_and_missing_capability(tmp_path):
|
|
def bad_auth(url, token=None, timeout=10):
|
|
return http_ok(url, "wrong" if token else None)
|
|
assert any(b.startswith("AUTH") for b in run_preflight(env(tmp_path), http_get=bad_auth, probe=fake_probe())["blockers"])
|
|
|
|
def old_hub(url, token=None, timeout=10):
|
|
return (200, {"version": "0.9.0"}) if url.endswith("/version") else http_ok(url, token)
|
|
assert any(b.startswith("HUB_VERSION_COMPATIBLE") for b in run_preflight(env(tmp_path), http_get=old_hub, probe=fake_probe())["blockers"])
|
|
assert any(b.startswith("CAPABILITIES") for b in run_preflight(env(tmp_path), http_get=http_ok, probe=fake_probe(caps_missing=("CEST",)))["blockers"])
|
|
|
|
|
|
def test_preflight_fails_when_company_context_is_ambiguous_or_crt_conflicts(tmp_path):
|
|
two = {"c000004": [{"code": "1", "cnpj": CN, "uf": "BA", "crt": "1"}, {"code": "2", "cnpj": "22222222000122", "uf": "SP", "crt": "1"}], "emissor": []}
|
|
r = run_preflight(env(tmp_path), http_get=http_ok, probe=fake_probe(two))
|
|
assert r["PREFLIGHT"] == "FAIL" and any(b.startswith("COMPANY_CONTEXT") for b in r["blockers"])
|
|
conflict = {"emissor": [{"cnpj": CN, "uf": "BA", "crt": "3"}]}
|
|
assert any("CRT_SOURCE_CONFLICT" in b for b in run_preflight(env(tmp_path), http_get=http_ok, probe=fake_probe(conflict))["blockers"])
|
|
|
|
|
|
def test_hub_unreachable_is_a_clear_blocker(tmp_path):
|
|
def down(url, token=None, timeout=10):
|
|
raise ConnectionError("nope")
|
|
r = run_preflight(env(tmp_path), http_get=down, probe=fake_probe())
|
|
assert r["PREFLIGHT"] == "FAIL" and any(b.startswith("HUB_REACHABLE") for b in r["blockers"])
|
|
|
|
|
|
def test_major_minor():
|
|
assert major_minor("1.0.0-rc1") == "1.0" and major_minor("1.0.3") == "1.0" and major_minor("0.9.0") != major_minor("1.0.0-rc1")
|
|
|
|
|
|
# ---------------- zero-write guarantee ----------------
|
|
def test_dry_run_makes_writes_technically_impossible_whatever_firebird_writes_says():
|
|
for extra in ({"FIREBIRD_WRITES": "1", "FISCAL_MODE": "DRY_RUN"}, {"FIREBIRD_WRITES": "0", "FISCAL_MODE": "AUTO_APPLY"}, {"FIREBIRD_WRITES": "1", "FISCAL_MODE": "dry_run"}):
|
|
cfg = service.Config({"FIREBIRD_DATABASE_PATH": "x.fdb", **extra})
|
|
assert cfg.firebird_writes is False
|
|
assert service.Config({"FIREBIRD_DATABASE_PATH": "x.fdb", "FIREBIRD_WRITES": "1", "FISCAL_MODE": "AUTO_APPLY"}).firebird_writes is True
|
|
assert service.Config({"FIREBIRD_DATABASE_PATH": "x.fdb", "FIREBIRD_WRITES": "1"}).firebird_writes is True # legacy: unset FISCAL_MODE follows the flag
|
|
assert service.Config({"FIREBIRD_DATABASE_PATH": "x.fdb"}).fiscal_mode == "DRY_RUN"
|
|
|
|
|
|
def test_write_connection_cannot_be_opened_when_writes_are_disabled(monkeypatch):
|
|
opened = []
|
|
monkeypatch.setitem(sys.modules, "fdb", type("F", (), {"connect": staticmethod(lambda **kw: opened.append(kw))}))
|
|
with pytest.raises(RuntimeError):
|
|
service.open_write_connection({"FB_USER": "u", "FB_PASSWORD": "SECRETPASSVALUE9"}, "x.fdb", firebird_writes=False)
|
|
assert opened == []
|
|
|
|
|
|
def test_dry_run_job_never_opens_a_write_connection_nor_calls_the_executor(monkeypatch):
|
|
class Boom:
|
|
def __getattr__(self, name):
|
|
if name in ("execute_guarded_update", "execute_guarded_restore"):
|
|
raise AssertionError("executor write called in DRY_RUN")
|
|
return getattr(__import__("scripts.firebird_write_executor", fromlist=["x"]) if False else object(), name, None)
|
|
cfg = service.Config({"FIREBIRD_DATABASE_PATH": "x.fdb", "FIREBIRD_WRITES": "1", "FISCAL_MODE": "DRY_RUN"})
|
|
monkeypatch.setattr(service, "read_products", lambda env, codigos=None: [{"CODIGO": "000001", "PRODUTO": "X", "CLASSIFICACAO_FISCAL": "22030000", "IPI": None}])
|
|
monkeypatch.setattr(service, "HubClient", lambda url, token: type("H", (), {"classify_batch": lambda self, p: [{"status": "EXACT_NCM", "ipi_status": "IPI_RESOLVED", "classification": {"ipi": {"rate_kind": "NUMERIC", "rate_numeric": 3.9}},
|
|
"dataset_versions": {}}]})())
|
|
monkeypatch.setattr(service, "build_job_coverage", lambda *a, **k: {"TOTAL_PRODUCTS": 1})
|
|
monkeypatch.setattr(service, "open_write_connection", lambda *a, **k: (_ for _ in ()).throw(AssertionError("write connection opened in DRY_RUN")))
|
|
|
|
class State:
|
|
def log_ipi_write(self, *a): pass
|
|
|
|
class Ex:
|
|
WriteRefused = Exception
|
|
def execute_guarded_update(self, *a, **k): raise AssertionError("executor write called in DRY_RUN")
|
|
|
|
result, error = service.execute_job({"kind": "REPROCESS_ALL", "agent_id": 1, "payload": {}, "job_id": 1}, cfg, Ex(), State(), {})
|
|
assert error is None and result["writes_attempted"] == 0 and result["firebird_writes_enabled"] is False
|