445 lines
26 KiB
Python
445 lines
26 KiB
Python
import re
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from central.app import rtc_reference_data as ref
|
|
from central.app.cclasstrib_resolver import local_catalog_hash
|
|
from gti_fiscal_agent import rtc_auto_apply as apply_mod
|
|
from gti_fiscal_agent import unified_fiscal as uf
|
|
|
|
pytestmark = pytest.mark.real_unified_fiscal
|
|
|
|
TODAY = date(2026, 9, 24)
|
|
NORMAL = {"crt": "3"}
|
|
CLASS_COLS = ("CCLASSTRIB", "CSTIBSCBS", "INICIO_VIGENCIA", "FIM_VIGENCIA", "TIPO_ALIQUOTA", "PRED_IBS", "PRED_CBS", "IND_NFE", "IND_NFCE")
|
|
SCHEMA = ([("C000025", c) for c in ("CODIGO", "CCLASSTRIB", "CLASSIFICACAO_FISCAL", "PISCOFINS", "INCIDENCIA_PISCOFINS")] + [("RTC_CCLASSTRIB", c) for c in CLASS_COLS] + [("RTC_CST", "CSTIBSCBS"), ("PARAMETROS", "USAR_REFORMA_TRIBUTARIA")])
|
|
|
|
|
|
class FakeFirebird:
|
|
"""A tiny SGV: PARAMETROS, the RTC catalog (from the official table) and C000025. Every statement is recorded; only SELECT and the single allowed UPDATE are understood."""
|
|
|
|
def __init__(self, products, param=(1,), catalog_tweak=None, skip=()):
|
|
self.products = {c: dict(v) for c, v in products.items()} # code -> {"CCLASSTRIB":..., "CLASSIFICACAO_FISCAL":...}
|
|
self.param = list(param)
|
|
self.catalog = {r["code"]: {"cst": r["cst"], "start": r["start"], "end": r["end"], "tipo": r["tipo_aliquota"], "pred_ibs": r["p_red_ibs"], "pred_cbs": r["p_red_cbs"], "ind_nfe": r["ind_nfe"], "ind_nfce": r["ind_nfce"]}
|
|
for r in ref.OFFICIAL_CCLASSTRIB if r["code"] not in skip and not r["end"]}
|
|
for code, change in (catalog_tweak or {}).items():
|
|
self.catalog[code] = {**self.catalog[code], **change}
|
|
self.cst_table = {r["cst"] for r in ref.OFFICIAL_CCLASSTRIB} # RTC_CST is its own table: a class may point to a CST that is not in it
|
|
self.crt = ("3",) # C000004.CRT rows
|
|
self.statements, self.commits, self.rollbacks = [], 0, 0
|
|
self._pending = {}
|
|
|
|
def cursor(self):
|
|
return FakeCursor(self)
|
|
|
|
def commit(self):
|
|
for code, value in self._pending.items():
|
|
self.products[code]["CCLASSTRIB"] = value
|
|
self._pending = {}
|
|
self.commits += 1
|
|
|
|
def rollback(self):
|
|
self._pending = {}
|
|
self.rollbacks += 1
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self, db):
|
|
self.db, self._rows, self.rowcount = db, [], -1
|
|
|
|
def execute(self, sql, params=()):
|
|
db = self.db
|
|
db.statements.append((sql, tuple(params)))
|
|
text = " ".join(sql.split())
|
|
up = text.upper()
|
|
if up.startswith("SELECT TRIM(RF.RDB$RELATION_NAME)"):
|
|
self._rows = list(SCHEMA)
|
|
elif up.startswith("SELECT FIRST 20 USAR_REFORMA_TRIBUTARIA"):
|
|
self._rows = [(v,) for v in db.param]
|
|
elif "RDB$RELATION_CONSTRAINTS" in up:
|
|
self._rows = [(1,)]
|
|
elif up.startswith("SELECT TRIM(CCLASSTRIB)") and "FROM RTC_CCLASSTRIB" in up:
|
|
cols = [c.strip() for c in text[len("SELECT "):up.index(" FROM ")].split(",")]
|
|
rows = []
|
|
for code, r in db.catalog.items():
|
|
mapping = {"CSTIBSCBS": r["cst"], "INICIO_VIGENCIA": date.fromisoformat(r["start"]) if r["start"] else None, "FIM_VIGENCIA": None, "TIPO_ALIQUOTA": r["tipo"], "PRED_IBS": r["pred_ibs"],
|
|
"PRED_CBS": r["pred_cbs"], "IND_NFE": "1" if r["ind_nfe"] else "0", "IND_NFCE": "1" if r["ind_nfce"] else "0"}
|
|
rows.append(tuple(code if c.startswith("TRIM(CCLASSTRIB") else mapping[re.sub(r"TRIM\((\w+)\)", r"\1", c)] for c in cols))
|
|
self._rows = rows
|
|
elif up.startswith("SELECT TRIM(CSTIBSCBS) FROM RTC_CST"):
|
|
self._rows = [(c,) for c in sorted(db.cst_table)]
|
|
elif up.startswith("SELECT CODIGO, CCLASSTRIB, CLASSIFICACAO_FISCAL"):
|
|
self._rows = [(c, v["CCLASSTRIB"], v["CLASSIFICACAO_FISCAL"], "N", "NENHUMA") for c, v in sorted(db.products.items())]
|
|
elif up.startswith("SELECT DISTINCT TRIM(CRT) FROM C000004"):
|
|
self._rows = [(c,) for c in db.crt]
|
|
elif up.startswith("SELECT CCLASSTRIB, CLASSIFICACAO_FISCAL, PRODUTO FROM C000025"):
|
|
code = params[0]
|
|
self._rows = [(db.products[code]["CCLASSTRIB"], db.products[code]["CLASSIFICACAO_FISCAL"], db.products[code].get("PRODUTO", ""))] if code in db.products else []
|
|
if code in db.products and db.products[code].get("DUP"):
|
|
self._rows = self._rows * 2
|
|
elif up.startswith("SELECT CCLASSTRIB FROM C000025"):
|
|
code = params[0]
|
|
self._rows = [(db.products[code]["CCLASSTRIB"],)]
|
|
elif up.startswith("UPDATE C000025 SET CCLASSTRIB = NULL WHERE CODIGO = ? AND TRIM(CCLASSTRIB) = ?"):
|
|
code, written = params
|
|
product = db.products.get(code)
|
|
hit = product is not None and (product["CCLASSTRIB"] or "").strip() == written
|
|
self.rowcount = 1 if hit else 0
|
|
if hit:
|
|
db._pending[code] = None
|
|
elif up.startswith("UPDATE C000025 SET CCLASSTRIB = ? WHERE CODIGO = ?"):
|
|
value, code = params
|
|
product = db.products.get(code)
|
|
hit = product is not None and not (product["CCLASSTRIB"] or "").strip()
|
|
self.rowcount = (2 if product and product.get("DUP") else 1) if hit else 0
|
|
if hit and self.rowcount == 1:
|
|
db._pending[code] = value
|
|
else:
|
|
raise AssertionError(f"unexpected statement: {sql}")
|
|
|
|
def fetchall(self):
|
|
return self._rows
|
|
|
|
def fetchone(self):
|
|
return self._rows[0] if self._rows else None
|
|
|
|
|
|
def product(ncm, value=None, **extra):
|
|
return {"CCLASSTRIB": value, "CLASSIFICACAO_FISCAL": ncm, **extra}
|
|
|
|
|
|
def assess(db, ncm_ok=None, company=NORMAL):
|
|
return uf.assess_from_connection(db, company, today=TODAY, ncm_valid=ncm_ok if ncm_ok is not None else {c: True for c in db.products})
|
|
|
|
|
|
def candidates_of(db, limit=500):
|
|
out = assess(db)
|
|
products = [{"CODIGO": c, "CCLASSTRIB": v["CCLASSTRIB"], "CLASSIFICACAO_FISCAL": v["CLASSIFICACAO_FISCAL"]} for c, v in db.products.items()]
|
|
cands, total = apply_mod.select_candidates(out["findings"], products, limit=limit)
|
|
return out, cands, total
|
|
|
|
|
|
def run(db, cands, out, **kw):
|
|
return apply_mod.apply_candidates(db, cands, company=NORMAL, decision={"local_catalog_hash": out["context"]["local_catalog_hash"]}, now="2026-09-24T20:00:00+00:00", job_id=7, today=TODAY, **kw)
|
|
|
|
|
|
PRODUCTS = {"000001": product("10063021"), "000002": product("04090000"), "000003": product("15171000"), "000004": product("30049099"), "000005": product("39235000", "000001")}
|
|
|
|
|
|
def test_gate_is_driven_by_the_central_mode_only_and_defaults_to_closed():
|
|
assert apply_mod.write_gate(mode="ENABLED")["allowed"] is True
|
|
assert apply_mod.write_gate(mode="APPROVAL_REQUIRED", approved=True)["allowed"] is True
|
|
for kwargs in ({"mode": "DISABLED"}, {"mode": "DRY_RUN"}, {"mode": "APPROVAL_REQUIRED", "approved": False}, {"mode": "ENABLED", "capability": False}, {"mode": "bogus"}, {"mode": ""}):
|
|
g = apply_mod.write_gate(**kwargs)
|
|
assert g["allowed"] is False and g["missing"]
|
|
assert apply_mod.ALLOWED_RTC_WRITE_FIELDS == ("CCLASSTRIB",) and apply_mod.CAPABILITY == "RTC_CCLASSTRIB_WRITE_V1"
|
|
|
|
|
|
def test_only_proven_products_become_candidates():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out, cands, total = candidates_of(db)
|
|
assert sorted((c["product_code"], c["proposed"], c["rule_id"]) for c in cands) == [("000001", "200003", "RTC-R3"), ("000002", "200034", "RTC-R3"), ("000003", "200003", "RTC-R3")]
|
|
assert total == 3 and out["context"]["rtc_resolution"]["ready_to_apply"] == 3 and out["context"]["rtc_resolution"]["ambiguous"] == 1 # medicine chapter is ambiguous
|
|
assert out["context"]["rtc_resolution"]["valid_existing"] == 1 # 000005 already has the right class
|
|
assert "000004" not in {c["product_code"] for c in cands}
|
|
|
|
|
|
def test_apply_writes_only_cclasstrib_with_audit_and_second_run_writes_nothing():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out, cands, _ = candidates_of(db)
|
|
result = run(db, cands, out)
|
|
assert result["write_successes"] == 3 and result["write_attempts"] == 3 and result["aborted"] == []
|
|
assert {c: db.products[c]["CCLASSTRIB"] for c in ("000001", "000002", "000003")} == {"000001": "200003", "000002": "200034", "000003": "200003"}
|
|
updates = [s for s, _ in db.statements if s.upper().startswith("UPDATE")]
|
|
assert len(updates) == 3 and all(re.fullmatch(r"UPDATE C000025 SET CCLASSTRIB = \? WHERE CODIGO = \? AND \(CCLASSTRIB IS NULL OR TRIM\(CCLASSTRIB\) = ''\)", u) for u in updates)
|
|
assert not [s for s, _ in db.statements if re.match(r"(INSERT|DELETE|ALTER|DROP)", s.strip(), re.I)]
|
|
audit = result["applied"][0]
|
|
assert {"product_code", "field", "old_value", "new_value", "rule_id", "evidence", "official_source_version", "local_catalog_hash", "timestamp", "job_id", "outcome"} <= set(audit)
|
|
assert audit["old_value"] is None and audit["field"] == "CCLASSTRIB" and audit["official_source_version"]["cclasstrib_sha256"] == ref.OFFICIAL_CATALOG["sha256"] and audit["job_id"] == 7
|
|
out2, cands2, total2 = candidates_of(db) # idempotent: nothing left to write
|
|
assert cands2 == [] and total2 == 0 and out2["context"]["rtc_resolution"]["valid_existing"] == 4
|
|
before = len(db.statements)
|
|
assert run(db, cands2, out2)["write_attempts"] == 0 and len(db.statements) == before
|
|
|
|
|
|
def test_rtc_switched_off_between_decision_and_write_aborts_every_product():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out, cands, _ = candidates_of(db)
|
|
db.param = [0]
|
|
result = run(db, cands, out)
|
|
assert result["write_successes"] == 0 and result["aborted_reasons"] == {"RTC_NO_LONGER_ENABLED": 3} and all(p["CCLASSTRIB"] in (None, "000001") for p in db.products.values())
|
|
|
|
|
|
def test_conflicting_parameter_rows_abort():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out, cands, _ = candidates_of(db)
|
|
db.param = [1, 0]
|
|
assert run(db, cands, out)["aborted_reasons"] == {"RTC_NO_LONGER_ENABLED": 3}
|
|
|
|
|
|
def test_catalog_changed_or_invalid_between_decision_and_write_aborts():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out, cands, _ = candidates_of(db)
|
|
db.catalog["200003"]["pred_ibs"] = "60"
|
|
assert run(db, cands, out)["aborted_reasons"] == {"LOCAL_CATALOG_CHANGED_SINCE_DECISION": 3}
|
|
db2 = FakeFirebird(PRODUCTS)
|
|
out2, cands2, _ = candidates_of(db2)
|
|
db2.catalog["999999"] = {**db2.catalog["000001"], "cst": "555"} # class pointing to a CST that does not exist -> invalid catalog
|
|
invalid = apply_mod.apply_candidates(db2, cands2, company=NORMAL, decision={}, now="2026-09-24T20:00:00+00:00", job_id=7, today=TODAY) # no hash pinned: integrity check alone must stop it
|
|
assert invalid["aborted_reasons"] == {"RTC_CATALOG_INVALID": 3}
|
|
|
|
|
|
def test_product_changed_between_decision_and_write_is_left_alone():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out, cands, _ = candidates_of(db)
|
|
db.products["000001"]["CLASSIFICACAO_FISCAL"] = "30049099" # NCM edited in the ERP meanwhile
|
|
db.products["000002"]["CCLASSTRIB"] = "200034" # operator filled it meanwhile (same value)
|
|
db.products["000003"]["CCLASSTRIB"] = "200034" # ... or with a DIFFERENT value: never overwritten
|
|
result = run(db, cands, out)
|
|
assert result["aborted_reasons"] == {"PRODUCT_CHANGED_NCM": 1, "NO_WRITE_REQUIRED": 1, "VALUE_ALREADY_SET_NEVER_OVERWRITTEN": 1} and result["write_successes"] == 0
|
|
assert db.products["000003"]["CCLASSTRIB"] == "200034" and not [s for s, _ in db.statements if s.upper().startswith("UPDATE")]
|
|
|
|
|
|
def test_duplicate_code_rowcount_protection_and_isolation_of_failures():
|
|
products = {**PRODUCTS, "000002": product("04090000", DUP=True)}
|
|
db = FakeFirebird(products)
|
|
out, cands, _ = candidates_of(db)
|
|
result = run(db, cands, out)
|
|
assert result["aborted_reasons"] == {"PRODUCT_ROW_NOT_UNIQUE": 1} and result["write_successes"] == 2 # the ambiguous row is skipped, the rest still applied
|
|
assert db.products["000002"]["CCLASSTRIB"] is None
|
|
|
|
|
|
def test_existing_different_valid_class_is_a_conflict_blocker_never_a_write():
|
|
db = FakeFirebird({"000001": product("10063021", "000001")}) # rice already classified as fully taxed: differs from the proven Anexo I class
|
|
out = assess(db)
|
|
types = {b["type"] for b in out["findings"]["000001"]["blockers"]}
|
|
assert types == {"CCLASSTRIB_CONFLICT"} and out["context"]["rtc_resolution"]["conflicts"] == 1
|
|
_, cands, _ = candidates_of(db)
|
|
assert cands == []
|
|
|
|
|
|
def test_official_mismatch_of_an_existing_class_is_reported():
|
|
db = FakeFirebird({"000001": product("39235000", "000001")}, catalog_tweak={"000001": {"ind_nfe": False, "ind_nfce": False}})
|
|
types = {b["type"] for b in assess(db)["findings"]["000001"]["blockers"]}
|
|
assert types == {"CCLASSTRIB_OFFICIAL_MISMATCH"}
|
|
|
|
|
|
def test_batch_limit_bounds_the_blast_radius():
|
|
db = FakeFirebird({f"{i:06d}": product("04090000") for i in range(1, 8)})
|
|
out, cands, total = candidates_of(db, limit=3)
|
|
assert len(cands) == 3 and total == 7
|
|
|
|
|
|
def test_simples_never_produces_candidates_even_with_rtc_on():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out = uf.assess_from_connection(db, {"crt": "1"}, today=TODAY, ncm_valid={c: True for c in PRODUCTS})
|
|
products = [{"CODIGO": c, "CCLASSTRIB": v["CCLASSTRIB"], "CLASSIFICACAO_FISCAL": v["CLASSIFICACAO_FISCAL"]} for c, v in db.products.items()]
|
|
candidates, total = apply_mod.select_candidates(out["findings"], products)
|
|
|
|
assert out["context"]["rtc_resolution"]["ready_to_apply"] == 0
|
|
assert out["context"]["rtc_resolution"]["not_applicable"] == len(PRODUCTS)
|
|
assert candidates == [] and total == 0
|
|
assert not [b for f in out["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
|
|
assert all(sql.lstrip().upper().startswith("SELECT") for sql, _ in db.statements)
|
|
|
|
|
|
def test_rtc_off_has_no_resolution_and_no_candidates():
|
|
db = FakeFirebird(PRODUCTS, param=(0,))
|
|
out, cands, total = candidates_of(db)
|
|
assert cands == [] and total == 0 and out["context"]["rtc_resolution"] is None
|
|
|
|
|
|
def test_catalog_hash_is_recorded_for_audit():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out = assess(db)
|
|
schema, catalog, _ = uf.load_rtc_context(db.cursor())
|
|
assert out["context"]["local_catalog_hash"] == local_catalog_hash(catalog["rows"]) and re.fullmatch(r"[0-9a-f]{64}", out["context"]["local_catalog_hash"])
|
|
|
|
|
|
# ---------------------------------------------------------------- job integration (gates)
|
|
import pytest # noqa: E402
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _hub_echoes_the_job_authorization(monkeypatch):
|
|
"""By default the Hub still says what the job said; individual tests change that to prove the pre-write re-check."""
|
|
from test_run_agent_service import service
|
|
monkeypatch.setattr(service, "_fetch_authorizations", lambda config, job: dict(job.get("authorizations") or {}))
|
|
|
|
|
|
def _service():
|
|
from test_run_agent_service import _config, service
|
|
return service, _config
|
|
|
|
|
|
def _fiscal_for(db):
|
|
return assess(db)
|
|
|
|
|
|
def _products_of(db):
|
|
return [{"CODIGO": c, "CCLASSTRIB": v["CCLASSTRIB"], "CLASSIFICACAO_FISCAL": v["CLASSIFICACAO_FISCAL"], "PRODUTO": v.get("PRODUTO", "")} for c, v in db.products.items()]
|
|
|
|
|
|
def _no_write(service, monkeypatch):
|
|
monkeypatch.setattr(service, "open_write_connection", lambda *a, **k: (_ for _ in ()).throw(AssertionError("no write connection may be opened")))
|
|
|
|
|
|
def test_job_without_a_central_mode_or_with_disabled_or_dry_run_never_writes(monkeypatch):
|
|
service, _config = _service()
|
|
db = FakeFirebird(PRODUCTS)
|
|
fiscal = _fiscal_for(db)
|
|
_no_write(service, monkeypatch)
|
|
for auth in ({}, {"rtc_write_mode": "DISABLED"}, {"rtc_write_mode": "DRY_RUN"}, {"rtc_write_mode": "bogus"}, {"rtc_write_mode": "APPROVAL_REQUIRED", "rtc_approved_codes": []}, {"rtc_auto_apply": True}):
|
|
summary = service._rtc_auto_apply({}, _config(), {"job_id": 1, "authorizations": auth}, _products_of(db), fiscal, NORMAL)
|
|
assert summary["gate"]["allowed"] is False and summary["ready_total"] == 3 and summary["write_attempts"] == 0 and summary["applied"] == []
|
|
off = service._rtc_auto_apply({}, _config(), {"job_id": 1, "authorizations": {"rtc_write_mode": "DISABLED"}}, _products_of(db), fiscal, NORMAL)
|
|
dry = service._rtc_auto_apply({}, _config(), {"job_id": 1, "authorizations": {"rtc_write_mode": "DRY_RUN"}}, _products_of(db), fiscal, NORMAL)
|
|
assert off["preview"] == [] and off["mode"] == "REPORT_ONLY"
|
|
assert [c["product_code"] for c in dry["preview"]] == ["000001", "000002", "000003"] and all(c["rule_id"] == "RTC-R3" for c in dry["preview"]) and dry["capability"] == "RTC_CCLASSTRIB_WRITE_V1"
|
|
|
|
|
|
def test_r4_suggestions_are_never_in_the_preview_and_never_written(monkeypatch):
|
|
service, _config = _service()
|
|
db = FakeFirebird({"000001": product("39235000"), "000002": product("39235000")})
|
|
fiscal = _fiscal_for(db)
|
|
_no_write(service, monkeypatch)
|
|
summary = service._rtc_auto_apply({}, _config(), {"job_id": 1, "authorizations": {"rtc_write_mode": "ENABLED"}}, _products_of(db), fiscal, NORMAL)
|
|
assert summary["ready_total"] == 0 and summary["preview"] == [] and summary["write_attempts"] == 0
|
|
|
|
|
|
def test_enabled_mode_applies_without_any_local_switch(monkeypatch):
|
|
service, _config = _service()
|
|
db = FakeFirebird(PRODUCTS)
|
|
fiscal = _fiscal_for(db)
|
|
monkeypatch.setattr(service, "open_write_connection", lambda env, path: _Closeable(db))
|
|
summary = service._rtc_auto_apply({}, _config(), {"job_id": 9, "authorizations": {"rtc_write_mode": "ENABLED"}}, _products_of(db), fiscal, NORMAL) # default config: no firebird_writes, no env switch
|
|
assert summary["mode"] == "ENABLED" and summary["write_successes"] == 3 and summary["allowed_fields"] == ["CCLASSTRIB"] and all(a["job_id"] == 9 and a["authorization_origin"] == "HUB_MODE:ENABLED" for a in summary["applied"])
|
|
assert [db.products[c]["CCLASSTRIB"] for c in ("000001", "000002", "000003")] == ["200003", "200034", "200003"]
|
|
a = summary["applied"][0]
|
|
assert a["rule_version"] and a["company_crt"] == "3" and a["action"] == "APPLY" and a["rtc_status"] == "ENABLED" and a["local_catalog_state"] == "MATCH_SAFE" and a["old_value"] is None
|
|
|
|
|
|
def test_approval_required_writes_only_the_approved_products(monkeypatch):
|
|
service, _config = _service()
|
|
db = FakeFirebird(PRODUCTS)
|
|
fiscal = _fiscal_for(db)
|
|
monkeypatch.setattr(service, "open_write_connection", lambda env, path: _Closeable(db))
|
|
summary = service._rtc_auto_apply({}, _config(), {"job_id": 9, "authorizations": {"rtc_write_mode": "APPROVAL_REQUIRED", "rtc_approved_codes": ["000002"]}}, _products_of(db), fiscal, NORMAL)
|
|
assert summary["write_successes"] == 1 and [db.products[c]["CCLASSTRIB"] for c in ("000001", "000002", "000003")] == [None, "200034", None]
|
|
|
|
|
|
def test_job_never_writes_when_the_erp_says_rtc_is_off_even_in_enabled_mode(monkeypatch):
|
|
service, _config = _service()
|
|
db = FakeFirebird(PRODUCTS, param=(0,))
|
|
fiscal = _fiscal_for(db)
|
|
_no_write(service, monkeypatch)
|
|
products = [{"CODIGO": c, "CCLASSTRIB": None, "CLASSIFICACAO_FISCAL": "04090000"} for c in db.products]
|
|
summary = service._rtc_auto_apply({}, _config(), {"job_id": 1, "authorizations": {"rtc_write_mode": "ENABLED"}}, products, fiscal, NORMAL)
|
|
assert summary["write_attempts"] == 0 and summary["ready_total"] == 0
|
|
|
|
|
|
def test_company_regime_or_fingerprint_change_aborts_before_the_update():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out, cands, _ = candidates_of(db)
|
|
db.crt = ("1",) # the company was switched to Simples between decision and write
|
|
res = run(db, cands, out)
|
|
assert res["write_successes"] == 0 and res["aborted_reasons"] == {"COMPANY_CONTEXT_CHANGED": 3}
|
|
db.crt = ("3", "1") # ambiguous company context is also refused
|
|
assert run(db, cands, out)["aborted_reasons"] == {"COMPANY_CONTEXT_CHANGED": 3}
|
|
db.crt = ("3",)
|
|
db.products["000001"]["PRODUTO"] = "RENAMED AFTER THE DECISION"
|
|
res = run(db, cands, out)
|
|
assert res["aborted_reasons"] == {"PRODUCT_CHANGED_SINCE_DECISION": 1} and res["write_successes"] == 2 and db.products["000001"]["CCLASSTRIB"] is None
|
|
assert not any("UPDATE" in q.upper() and "C000004" in q.upper() for q, _ in db.statements)
|
|
|
|
|
|
def test_rollback_restores_only_when_the_value_still_equals_the_written_one_and_reapply_works():
|
|
db = FakeFirebird(PRODUCTS)
|
|
out, cands, _ = candidates_of(db)
|
|
first = run(db, cands, out)
|
|
assert first["write_successes"] == 3
|
|
items = [{"audit_id": i + 1, "product_code": a["product_code"], "written_value": a["new_value"], "rule_id": a["rule_id"]} for i, a in enumerate(first["applied"])]
|
|
db.products["000002"]["CCLASSTRIB"] = "200035" # an operator changed it afterwards
|
|
res = apply_mod.rollback_records(db, items, now="2026-09-24T21:00:00+00:00", job_id=8)
|
|
by = {r["product_code"]: r for r in res["rollbacks"]}
|
|
assert res["rolled_back"] == 2 and by["000002"]["outcome"] == "ABORTED" and by["000002"]["reason"] == "ROLLBACK_CONFLICT_VALUE_CHANGED" and db.products["000002"]["CCLASSTRIB"] == "200035"
|
|
assert by["000001"]["outcome"] == "ROLLED_BACK" and by["000001"]["rollback_of"] == 1 and by["000001"]["old_value"] == "200003" and by["000001"]["new_value"] is None and db.products["000001"]["CCLASSTRIB"] is None
|
|
again = apply_mod.rollback_records(db, items, now="2026-09-24T21:05:00+00:00", job_id=9)["rollbacks"]
|
|
assert [r["reason"] for r in again if r["product_code"] == "000001"] == ["ALREADY_EMPTY"]
|
|
out2, cands2, _ = candidates_of(db) # reapply: the emptied products are candidates again, the changed one is not
|
|
assert sorted(c["product_code"] for c in cands2) == ["000001", "000003"] and run(db, cands2, out2)["write_successes"] == 2
|
|
|
|
|
|
def test_rollback_uses_one_fixed_statement_and_rejects_malformed_requests():
|
|
db = FakeFirebird(PRODUCTS)
|
|
bad = apply_mod.rollback_records(db, [{"audit_id": 1, "product_code": "000001", "written_value": ""}, {"audit_id": 2, "product_code": "000001", "written_value": "200003", "restore_value": "999999"}], now="t")
|
|
assert [r["reason"] for r in bad["rollbacks"]] == ["ROLLBACK_REQUEST_INVALID"] * 2 and db.statements == []
|
|
db.products["000001"]["CCLASSTRIB"] = "200003"
|
|
apply_mod.rollback_records(db, [{"audit_id": 3, "product_code": "000001", "written_value": "200003"}], now="t")
|
|
writes = [q for q, _ in db.statements if q.strip().upper().startswith(("UPDATE", "INSERT", "DELETE"))]
|
|
assert writes == ["UPDATE C000025 SET CCLASSTRIB = NULL WHERE CODIGO = ? AND TRIM(CCLASSTRIB) = ?"]
|
|
|
|
|
|
class _Closeable:
|
|
def __init__(self, db):
|
|
self._db = db
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._db, name)
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
def test_firebird_unavailable_for_the_write_connection_fails_the_job_closed_with_zero_writes(monkeypatch):
|
|
service, _config = _service()
|
|
db = FakeFirebird(PRODUCTS)
|
|
fiscal = _fiscal_for(db)
|
|
monkeypatch.setattr(service, "open_write_connection", lambda *a, **k: (_ for _ in ()).throw(OSError("Firebird unavailable")))
|
|
import pytest as _pytest
|
|
with _pytest.raises(OSError):
|
|
service._rtc_auto_apply({}, _config(), {"job_id": 1, "authorizations": {"rtc_write_mode": "ENABLED"}}, _products_of(db), fiscal, NORMAL)
|
|
assert all(v["CCLASSTRIB"] is None for c, v in db.products.items() if c != "000005") and not [q for q, _ in db.statements if q.upper().startswith("UPDATE")]
|
|
|
|
|
|
def test_connection_is_closed_even_when_a_product_fails_mid_batch(monkeypatch):
|
|
service, _config = _service()
|
|
db = FakeFirebird({**PRODUCTS, "000002": product("04090000", DUP=True)})
|
|
fiscal = _fiscal_for(db)
|
|
closed = []
|
|
conn = _Closeable(db)
|
|
conn.close = lambda: closed.append(True)
|
|
monkeypatch.setattr(service, "open_write_connection", lambda env, path: conn)
|
|
summary = service._rtc_auto_apply({}, _config(), {"job_id": 1, "authorizations": {"rtc_write_mode": "ENABLED"}}, _products_of(db), fiscal, NORMAL)
|
|
assert closed == [True] and summary["write_successes"] == 2 and summary["aborted_reasons"]
|
|
|
|
|
|
def test_authorization_revoked_after_the_job_was_claimed_means_zero_writes(monkeypatch):
|
|
service, _config = _service()
|
|
db = FakeFirebird(PRODUCTS)
|
|
fiscal = _fiscal_for(db)
|
|
monkeypatch.setattr(service, "open_write_connection", lambda env, path: _Closeable(db))
|
|
monkeypatch.setattr(service, "_fetch_authorizations", lambda config, job: {"rtc_write_mode": "DISABLED"}) # operator switched it off while the job was running
|
|
summary = service._rtc_auto_apply({}, _config(), {"job_id": 5, "authorizations": {"rtc_write_mode": "ENABLED"}}, _products_of(db), fiscal, NORMAL)
|
|
assert summary["write_successes"] == 0 and summary["aborted_reasons"] == {"AUTHORIZATION_CHANGED": 3}
|
|
assert [db.products[c]["CCLASSTRIB"] for c in ("000001", "000002", "000003")] == [None, None, None]
|
|
|
|
|
|
def test_unreachable_hub_is_not_authorization_and_approval_removal_is_honoured(monkeypatch):
|
|
service, _config = _service()
|
|
db = FakeFirebird(PRODUCTS)
|
|
fiscal = _fiscal_for(db)
|
|
monkeypatch.setattr(service, "open_write_connection", lambda env, path: _Closeable(db))
|
|
|
|
def down(config, job):
|
|
raise ConnectionError("hub down")
|
|
|
|
monkeypatch.setattr(service, "_fetch_authorizations", down)
|
|
summary = service._rtc_auto_apply({}, _config(), {"job_id": 6, "authorizations": {"rtc_write_mode": "ENABLED"}}, _products_of(db), fiscal, NORMAL)
|
|
assert summary["write_successes"] == 0 and summary["aborted_reasons"] == {"AUTHORIZATION_UNVERIFIABLE": 3}
|
|
monkeypatch.setattr(service, "_fetch_authorizations", lambda config, job: {"rtc_write_mode": "APPROVAL_REQUIRED", "rtc_approved_codes": ["000001"]}) # 000002 was un-approved meanwhile
|
|
summary = service._rtc_auto_apply({}, _config(), {"job_id": 7, "authorizations": {"rtc_write_mode": "APPROVAL_REQUIRED", "rtc_approved_codes": ["000001", "000002"]}}, _products_of(db), fiscal, NORMAL)
|
|
assert summary["write_successes"] == 1 and db.products["000001"]["CCLASSTRIB"] == "200003" and db.products["000002"]["CCLASSTRIB"] is None
|