gtiandClaude Sonnet 5
e935ed61d5
Missao 2 (fechamento de lacunas fiscais, read-only + UI/engine gated): - field_contract: pis_cofins_profile now also accepts the literal sentinels NOT_APPLICABLE/OPERATION_TIME, resolving to NA only from an audited FiscalContextEntry - never inferred from emission statistics. CFOP_ENTRADA under an already-decided entry_scope=MIXED_OR_DYNAMIC now resolves to NA instead of MANUAL, so it stops blocking product completeness and stops re-asking an already-answered question. Regression tests prove: no ADMIN decision => unchanged old behavior; decision present => gated bucket, no write. - context_keys: validate_entry accepts the two sentinels for pis_cofins_profile. - Contexto Fiscal UI: PIS/COFINS and CFOP_ENTRADA cards show the Bloco 2 cross-database evidence (5 installations, CRT 1 and 3: emission never reads the PIS/COFINS cadastral fields anywhere) and a link to an ADMIN-facing, anonymized dossie-for-the-accountant; entry_scope is now a dropdown. - release/tests: decoupled the Hub's GTI_FISCAL_VERSION from the Agent's own version (test_versions_are_coherent_across_hub_and_packages vs the new test_agent_version_is_internally_coherent) - a Hub-only release must not require touching, rebuilding or republishing the Agent. - New read-only evidence: docs/runs/GTI_dryrun_universe_summary.json (527 products, Agent 2's real database), docs/runs/BLOCO2_PIS_COFINS_CROSS_DB_V1.json (5 installations), docs/fiscal_completeness_matrix.md, docs/dossie_contador.md (internal, real names) + central/static/docs/dossie_contador.md (public, anonymized - no CNPJ/razao social/product names), docs/AGENT2_UPGRADE_RUNBOOK_V1.md. - CLAUDE.md: recorded the permanent fiscal-mission rules (no AI/scraping, no Firebird schema changes, deploy only via deploy_hub.ps1, never reenroll, no remote access to client Agent hosts, no client data in unauthenticated static routes). No schema/migration change. Zero Firebird writes (new scripts never open a write connection). Hub only - Agent code untouched. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
309 lines
20 KiB
Python
309 lines
20 KiB
Python
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from central.app.field_contract import (ALREADY, CHANGE, CONTRACT, MANUAL, NA, WRITE_FIELDS, Context, infer_role, resolve_product)
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
REFS = {"cfop_valid": {"5101", "5102", "5401", "5405", "1102", "1101", "1403", "2102"}, "csosn": {"101", "102", "103", "300", "400", "500"}, "cst_icms": {"00", "20", "40", "41", "60"},
|
|
"cst_pis": {"01", "49", "98"}, "cst_cofins": {"01", "49", "98"}, "origem": set("012345678"), "fcp_uf": {}, "cbs_rate_2026": None, "cclasstrib_count": 0}
|
|
CO1 = {"uf": "BA", "crt": "1", "contribuinte_ipi": "N", "company_code": "000001"}
|
|
CO3 = {"uf": "BA", "crt": "3", "contribuinte_ipi": "S", "company_code": "000001"}
|
|
|
|
|
|
def product(**kw):
|
|
p = {f: None for f in WRITE_FIELDS}
|
|
p.update({"CODIGO": "000001", "PRODUTO": "CERVEJA LATA", "CODGRUPO": "000019", "TIPO": "VENDA", "PROD_COZINHA": None, "PRODUTO_SERVICO": None, "CLASSIFICACAO_FISCAL": "22030000",
|
|
"CFOP": "5102", "CSOSN": "102", "CST": "000", "SITUACAO_TRIBUTARIA": "T", "IAT": "A", "IPPT": "T", "INCIDENCIA_PISCOFINS": "NENHUMA", "PISCOFINS": "N"})
|
|
p.update(kw)
|
|
return p
|
|
|
|
|
|
def decision(**kw):
|
|
d = {"status": "EXACT_NCM", "cest_status": "NO_CEST_MAPPING_FOUND", "ipi_status": "IPI_RESOLVED", "ba_st_mva_status": "NO_MVA_MAPPING_FOUND", "dataset_versions": {},
|
|
"classification": {"ncm": "22030000", "ipi": {"rate_kind": "NUMERIC", "rate_numeric": 3.9, "rate": "3.9"}, "ipi_candidates": [], "cest": None, "cest_candidates": [], "ba_st_mva": None, "ba_st_mva_candidates": []}}
|
|
cls = kw.pop("classification", {})
|
|
d.update(kw)
|
|
d["classification"] = {**d["classification"], **cls}
|
|
return d
|
|
|
|
|
|
EV = {"recipes": set(), "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {"000001": True}}
|
|
|
|
|
|
IPI_POP = {"scope": "COMPANY", "scope_key": "", "key": "store_ipi_profile", "value": "POPULATE_REFERENCE_RATE"}
|
|
|
|
|
|
def run(p=None, d=None, co=CO1, ctx=None, ev=EV, manual=None, ipi_profile=IPI_POP):
|
|
return resolve_product(p or product(), d or decision(), co, REFS, ev, Context((ctx or []) + ([ipi_profile] if ipi_profile else [])), manual_ncm=manual)
|
|
|
|
|
|
def st(res, f):
|
|
return res[f]["state"], res[f]["proposed"]
|
|
|
|
|
|
def test_every_write_field_has_a_complete_contract():
|
|
keys = {"ERP_FIELD", "UI_LABEL", "SEMANTICS", "DATA_TYPE", "SOURCE_OF_TRUTH", "REQUIRED_INPUTS", "RESOLUTION_RULE", "VALID_VALUES", "NOT_APPLICABLE_RULE", "MANUAL_INPUT_RULE", "WRITE_ALLOWED", "VALIDATION_RULE", "PROVENANCE"}
|
|
assert set(CONTRACT) == set(WRITE_FIELDS) and len(WRITE_FIELDS) == 23
|
|
for f, c in CONTRACT.items():
|
|
assert keys <= set(c) and all(c[k] not in (None, "", []) for k in keys - {"WRITE_ALLOWED"}), f
|
|
|
|
|
|
def test_only_five_final_states_and_manual_always_names_its_input():
|
|
res = run()
|
|
for f in WRITE_FIELDS:
|
|
assert res[f]["state"] in (ALREADY, CHANGE, NA, MANUAL)
|
|
if res[f]["state"] == MANUAL:
|
|
assert res[f]["input_key"]
|
|
assert res[f]["write_eligible"] == (res[f]["state"] == CHANGE)
|
|
|
|
|
|
def test_ncm_states_and_operator_ncm_reprocessing():
|
|
bad = run(p=product(CLASSIFICACAO_FISCAL="12345678"), d=decision(status="INVALID_NCM"))
|
|
assert st(bad, "CLASSIFICACAO_FISCAL")[0] == MANUAL and bad["CLASSIFICACAO_FISCAL"]["input_key"] == "ncm"
|
|
for f in ("IPI", "EX_TIPI", "CEST", "CFOP", "CSOSN"):
|
|
assert bad[f]["state"] == MANUAL and bad[f]["reason"] == "BLOCKED_BY_NCM"
|
|
assert st(run(manual=("22030000", False), p=product(CLASSIFICACAO_FISCAL="12345678"), d=decision(status="INVALID_NCM")), "CLASSIFICACAO_FISCAL")[0] == MANUAL
|
|
fixed = run(manual=("22030000", True), p=product(CLASSIFICACAO_FISCAL="12345678"), d=decision())
|
|
assert st(fixed, "CLASSIFICACAO_FISCAL") == (CHANGE, "22030000") and fixed["IPI"]["state"] == CHANGE and fixed["IPI"]["proposed"] == 3.9 # dependents reprocessed with the manual NCM
|
|
|
|
|
|
def test_ipi_rules_never_write_nt_or_blank_and_ask_when_ambiguous():
|
|
assert st(run(), "IPI") == (CHANGE, 3.9)
|
|
assert st(run(p=product(IPI=3.9)), "IPI")[0] == ALREADY
|
|
nt = run(d=decision(classification={"ipi": {"rate_kind": "NON_NUMERIC", "rate_classification": "NT"}}))
|
|
assert nt["IPI"]["state"] == NA and not nt["IPI"]["write_eligible"]
|
|
blank = run(d=decision(classification={"ipi": {"rate_kind": "BLANK"}}))
|
|
assert blank["IPI"]["state"] == NA and "NEVER_ZERO" in blank["IPI"]["reason"]
|
|
amb = decision(ipi_status="IPI_CANDIDATES", classification={"ipi": None, "ipi_candidates": [{"ex": None, "rate": "5", "description": "base"}, {"ex": "1", "rate": "0", "description": "Chope"}]})
|
|
r = run(d=amb)
|
|
assert r["IPI"]["state"] == MANUAL and r["IPI"]["input_key"] == "ex_tipi" and len(r["IPI"]["choices"]) == 2
|
|
solved = run(d=amb, ctx=[{"scope": "PRODUCT", "scope_key": "000001", "key": "ex_tipi", "value": "NONE"}])
|
|
assert st(solved, "IPI") == (CHANGE, 5.0) and solved["EX_TIPI"]["state"] == NA
|
|
|
|
|
|
def test_cest_contract():
|
|
off = [{"code": "03.021.03", "official_description": "Cerveja em lata", "description_score": 1.0}, {"code": "03.021.05", "official_description": "PET", "description_score": 0.5}]
|
|
conf = run(d=decision(cest_status="CEST_CONFIRMED", classification={"cest": ["03.021.03"], "cest_candidates": off}))
|
|
assert st(conf, "CEST") == (CHANGE, "0302103")
|
|
assert run(p=product(CEST="0302103"), d=decision(cest_status="CEST_CANDIDATES", classification={"cest_candidates": off}))["CEST"]["state"] == ALREADY
|
|
many = run(d=decision(cest_status="CEST_CANDIDATES", classification={"cest_candidates": off}))
|
|
assert many["CEST"]["state"] == MANUAL and many["CEST"]["input_key"] == "cest" and len(many["CEST"]["choices"]) == 2
|
|
assert run(d=decision(cest_status="CEST_DESCRIPTION_MISMATCH", classification={"cest_candidates": off}))["CEST"]["state"] == NA
|
|
assert run()["CEST"]["state"] == NA
|
|
assert run(p=product(CEST="9999999"))["CEST"]["state"] == MANUAL
|
|
picked = run(d=decision(cest_status="CEST_CANDIDATES", classification={"cest_candidates": off}), ctx=[{"scope": "NCM", "scope_key": "22030000", "key": "cest", "value": "03.021.05"}])
|
|
assert st(picked, "CEST") == (CHANGE, "0302105")
|
|
|
|
|
|
def test_cfop_role_and_st_rules():
|
|
ok = run() # RESALE (stock entries), not subject to ST, declared 5102 == derived
|
|
assert ok["CFOP"]["state"] == ALREADY and ok["_facts"]["role"] == "RESALE"
|
|
assert st(run(p=product(CFOP="5405")), "CFOP") == (CHANGE, "5102") # 5405 but proven NOT_SUBJECT
|
|
assert st(run(p=product(CFOP=None)), "CFOP") == (CHANGE, "5102") # blank filled
|
|
unk = run(d=decision(ba_st_mva_status="MVA_CANDIDATES"), p=product(CFOP=None)) # ST unknown, blank -> asks
|
|
assert unk["CFOP"]["state"] == MANUAL and unk["CFOP"]["input_key"] == "st_status"
|
|
assert run(d=decision(ba_st_mva_status="MVA_CANDIDATES"))["CFOP"]["state"] == ALREADY # declared valid, not contradicted
|
|
role_unknown = run(ev={**EV, "stock_in": {}}, p=product(CFOP=None))
|
|
assert role_unknown["CFOP"]["state"] == MANUAL and role_unknown["CFOP"]["input_key"] == "product_role"
|
|
own = run(ev={**EV, "recipes": {"000001"}}, p=product(CFOP="5102"))
|
|
assert st(own, "CFOP") == (CHANGE, "5101")
|
|
assert run(p=product(TIPO="MATERIA-PRIMA"))["CFOP"]["state"] == NA
|
|
|
|
|
|
def test_cfop_entrada_needs_company_entry_scope_then_derives():
|
|
r = run(p=product(CFOP_ENTRADA=None))
|
|
assert r["CFOP_ENTRADA"]["state"] == MANUAL and r["CFOP_ENTRADA"]["input_key"] == "entry_scope"
|
|
r = run(p=product(CFOP_ENTRADA=None), ctx=[{"scope": "COMPANY", "scope_key": "", "key": "entry_scope", "value": "INTRA"}])
|
|
assert st(r, "CFOP_ENTRADA") == (CHANGE, "1102")
|
|
assert run(p=product(CFOP_ENTRADA="1102"))["CFOP_ENTRADA"]["state"] == ALREADY
|
|
|
|
|
|
def test_cfop_entrada_mixed_scope_is_operation_dependent_and_never_invents_a_cfop():
|
|
mixed = [{"scope": "COMPANY", "scope_key": "", "key": "entry_scope", "value": "MIXED_OR_DYNAMIC"}]
|
|
r = run(p=product(CFOP_ENTRADA=None), ctx=mixed)
|
|
e = r["CFOP_ENTRADA"]
|
|
# NA (not MANUAL): the audited entry_scope=MIXED_OR_DYNAMIC decision already answered "why no value" -
|
|
# this must never block product completeness nor re-ask a question the admin already answered (Missão 2 / Bloco 3.4).
|
|
assert e["state"] == NA and e["proposed"] is None and e.get("input_key") is None
|
|
assert e["reason"] == "CFOP_ENTRADA_OPERATION_DEPENDENT_MIXED_SCOPE"
|
|
r2 = run(p=product(CFOP_ENTRADA="2102"), ctx=mixed)
|
|
assert r2["CFOP_ENTRADA"]["state"] == ALREADY and r2["CFOP_ENTRADA"]["reason"] == "DECLARED_VALID_OPERATION_DEPENDENT_MIXED_SCOPE"
|
|
from central.app.context_keys import validate_entry
|
|
assert validate_entry({"scope": "COMPANY", "key": "entry_scope", "value": "MIXED_OR_DYNAMIC"})[0]["value"] == "MIXED_OR_DYNAMIC"
|
|
|
|
|
|
def test_icms_code_fields_follow_the_regime():
|
|
r1 = run()
|
|
assert r1["CST"]["state"] == NA and r1["CST_ENTRADA"]["state"] == NA and r1["CSOSN"]["state"] == ALREADY
|
|
assert run(p=product(CSOSN=None))["CSOSN"]["state"] == MANUAL
|
|
assert st(run(p=product(CSOSN=None), ctx=[{"scope": "GROUP", "scope_key": "000019", "key": "icms_treatment", "value": "EXEMPT"}]), "CSOSN") == (CHANGE, "400")
|
|
assert st(run(p=product(CSOSN="102"), ctx=[{"scope": "GROUP", "scope_key": "000019", "key": "icms_treatment", "value": "ST_WITHHELD"}]), "CSOSN") == (CHANGE, "500")
|
|
r3 = run(co=CO3, p=product(CSOSN=None))
|
|
assert r3["CSOSN"]["state"] == NA and r3["CST"]["state"] == ALREADY and r3["ALIQUOTA"]["state"] in (ALREADY, MANUAL)
|
|
blank = run(co=CO3, p=product(CST=None, COD_ORIGEM_MERCADORIA="0"))
|
|
assert blank["CST"]["state"] == MANUAL and blank["CST"]["input_key"] == "icms_treatment"
|
|
filled = run(co=CO3, p=product(CST=None, COD_ORIGEM_MERCADORIA="0"), ctx=[{"scope": "COMPANY", "scope_key": "", "key": "icms_treatment", "value": "TAXED"}])
|
|
assert st(filled, "CST") == (CHANGE, "000")
|
|
|
|
|
|
def test_aliquota_not_copied_blindly():
|
|
assert run()["ALIQUOTA"]["state"] == NA # Simples CSOSN 102 highlights no ICMS
|
|
assert run(p=product(CSOSN="101", ALIQUOTA=None))["ALIQUOTA"]["input_key"] == "credit_rate"
|
|
assert st(run(co=CO3, p=product(ALIQUOTA="17"), ctx=[{"scope": "COMPANY", "scope_key": "", "key": "icms_rate", "value": "20,5"}]), "ALIQUOTA") == (CHANGE, 20.5)
|
|
|
|
|
|
def test_origin_binding_rules():
|
|
r0 = run(p=product(COD_ORIGEM_MERCADORIA=None, CST="040"))["COD_ORIGEM_MERCADORIA"] # audit V2: the CST first digit is padding, never origin evidence
|
|
assert r0["state"] == MANUAL and r0["reason"] == "ORIGIN_NO_EVIDENCE" and r0["proposed"] is None
|
|
assert run(p=product(COD_ORIGEM_MERCADORIA="0", CST="040"))["COD_ORIGEM_MERCADORIA"]["state"] == ALREADY
|
|
assert run(p=product(COD_ORIGEM_MERCADORIA="2", CST="040"))["COD_ORIGEM_MERCADORIA"]["state"] == ALREADY # declared origin stays; CST digit is padding
|
|
assert run(p=product(COD_ORIGEM_MERCADORIA=None, CST="540"))["COD_ORIGEM_MERCADORIA"]["state"] == MANUAL # 5 is not on the SGV screen
|
|
assert st(run(p=product(COD_ORIGEM_MERCADORIA=None, CST="540"), ctx=[{"scope": "GROUP", "scope_key": "000019", "key": "origin", "value": "1"}]), "COD_ORIGEM_MERCADORIA") == (CHANGE, 1)
|
|
|
|
|
|
def test_cst_ipi_pis_cofins_incidence_iat_ippt():
|
|
assert run()["CST_IPI"]["state"] == NA
|
|
assert run(co=CO3)["CST_IPI"]["state"] == MANUAL
|
|
r = run()
|
|
assert r["COD_SAIDA_PIS"]["state"] == MANUAL and r["COD_SAIDA_PIS"]["input_key"] == "pis_cofins_profile" and r["ALIQ_COFINS"]["state"] == MANUAL
|
|
prof = '{"pis_out_cst": "49", "pis_in_cst": "98", "pis_rate": "0", "cofins_out_cst": "49", "cofins_in_cst": "98", "cofins_rate": "0"}'
|
|
r = run(ctx=[{"scope": "COMPANY", "scope_key": "", "key": "pis_cofins_profile", "value": prof}])
|
|
assert st(r, "COD_SAIDA_PIS") == (CHANGE, "49") and st(r, "COD_ENTRADA_COFINS") == (CHANGE, "98") and st(r, "ALIQ_PIS") == (CHANGE, "0")
|
|
bad = run(ctx=[{"scope": "COMPANY", "scope_key": "", "key": "pis_cofins_profile", "value": '{"pis_out_cst": "77"}'}])
|
|
assert bad["COD_SAIDA_PIS"]["state"] == MANUAL # not in the official CST table
|
|
assert run(p=product(IAT=None))["IAT"]["input_key"] == "rounding"
|
|
assert run()["INCIDENCIA_PISCOFINS"]["state"] == ALREADY
|
|
assert st(run(ev={**EV, "recipes": {"000001"}}), "IPPT") == (CHANGE, "P") and run()["IPPT"]["state"] == ALREADY
|
|
|
|
|
|
def test_pis_cofins_not_applicable_and_operation_time_require_an_audited_admin_decision():
|
|
"""Missão 2 / Bloco 2.4: NOT_APPLICABLE/OPERATION_TIME for PIS/COFINS must never be inferred (e.g. from emission
|
|
statistics) - only ever accepted from an explicit, audited FiscalContextEntry(key="pis_cofins_profile")."""
|
|
pis_cofins_fields = ("COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS")
|
|
# Sem decisão ADMIN => comportamento antigo (continua bloqueando, pedindo pis_cofins_profile).
|
|
r = run()
|
|
for f in pis_cofins_fields:
|
|
assert r[f]["state"] == MANUAL and r[f]["input_key"] == "pis_cofins_profile"
|
|
# Com decisão ADMIN NOT_APPLICABLE => NA em todos, nunca escreve, nunca bloqueia completude.
|
|
na_ctx = [{"scope": "COMPANY", "scope_key": "", "key": "pis_cofins_profile", "value": "NOT_APPLICABLE"}]
|
|
r_na = run(ctx=na_ctx)
|
|
for f in pis_cofins_fields:
|
|
assert r_na[f]["state"] == NA and r_na[f]["proposed"] is None and r_na[f]["reason"] == "ADMIN_DECISION_PIS_COFINS_NOT_APPLICABLE"
|
|
# Com decisão ADMIN OPERATION_TIME => NA em todos (não bloqueia), motivo distinto, nunca escreve.
|
|
ot_ctx = [{"scope": "COMPANY", "scope_key": "", "key": "pis_cofins_profile", "value": "OPERATION_TIME"}]
|
|
r_ot = run(ctx=ot_ctx)
|
|
for f in pis_cofins_fields:
|
|
assert r_ot[f]["state"] == NA and r_ot[f]["proposed"] is None and r_ot[f]["reason"] == "ADMIN_DECISION_PIS_COFINS_OPERATION_TIME"
|
|
# Uma decisão de PROFILE (JSON) continua funcionando exatamente como antes (não regressivo).
|
|
prof = '{"pis_out_cst": "49", "pis_in_cst": "98", "pis_rate": "0", "cofins_out_cst": "49", "cofins_in_cst": "98", "cofins_rate": "0"}'
|
|
r_prof = run(ctx=[{"scope": "COMPANY", "scope_key": "", "key": "pis_cofins_profile", "value": prof}])
|
|
assert st(r_prof, "COD_SAIDA_PIS") == (CHANGE, "49")
|
|
assert run(p=product(TIPO="MATERIA-PRIMA"))["IPPT"]["state"] == NA
|
|
|
|
|
|
def test_context_hierarchy_product_beats_group_beats_company():
|
|
ctx = Context([{"scope": "COMPANY", "scope_key": "", "key": "product_role", "value": "RESALE"}, {"scope": "GROUP", "scope_key": "G1", "key": "product_role", "value": "OWN_PRODUCTION"},
|
|
{"scope": "PRODUCT", "scope_key": "P1", "key": "product_role", "value": "INPUT"}])
|
|
assert ctx.get("product_role", "P1", "G1") == ("INPUT", "PRODUCT")
|
|
assert ctx.get("product_role", "P2", "G1") == ("OWN_PRODUCTION", "GROUP")
|
|
assert ctx.get("product_role", "P2", "G9") == ("RESALE", "COMPANY")
|
|
assert Context([]).get("product_role", "P2", "G9") == (None, None)
|
|
|
|
|
|
def test_role_inference_is_evidence_based_and_conflicts_stay_unknown():
|
|
assert infer_role(product(TIPO="MATERIA-PRIMA"), EV, Context([]), "000001", "1")[0] == "INPUT"
|
|
assert infer_role(product(), {**EV, "stock_in": {}}, Context([]), "000001", "1")[0] == "UNKNOWN"
|
|
assert infer_role(product(PROD_COZINHA=1), {**EV, "purchases": {"000001": [{}]}}, Context([]), "000001", "1")[0] == "UNKNOWN"
|
|
assert infer_role(product(PROD_COZINHA=1), EV, Context([]), "000001", "1")[0] == "UNKNOWN" # kitchen flag alone is not proof
|
|
assert infer_role(product(PROD_COZINHA=1), {**EV, "recipes": {"000001"}}, Context([]), "000001", "1")[0] == "OWN_PRODUCTION"
|
|
|
|
|
|
# ---------------- executor guards ----------------
|
|
def load_executor():
|
|
spec = importlib.util.spec_from_file_location("fwe_local", ROOT / "scripts" / "firebird_write_executor.py")
|
|
m = importlib.util.module_from_spec(spec); sys.modules[spec.name] = m; spec.loader.exec_module(m)
|
|
return m
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self, db):
|
|
self.db, self.rowcount, self._row = db, 0, None
|
|
|
|
def execute(self, sql, params=()):
|
|
s = " ".join(sql.split())
|
|
if "RDB$FIELD_TYPE" in s:
|
|
self._row = (37, 20)
|
|
elif s.startswith("SELECT") and "FROM C000025" in s:
|
|
fields = [x.strip() for x in s.split("SELECT")[1].split("FROM")[0].split(",")]
|
|
row = self.db["rows"].get(params[0])
|
|
self._row = tuple(row[f] for f in fields) if row else None
|
|
elif s.startswith("UPDATE"):
|
|
row = self.db["rows"].get(params[-1])
|
|
self.rowcount = 0 if row is None else self.db["update_rowcount"]
|
|
if row is not None and self.rowcount == 1:
|
|
for f, v in zip([x.split("=")[0].strip() for x in s.split("SET")[1].split("WHERE")[0].split(",")], params[:-1]):
|
|
self.db["pending"][f] = v
|
|
|
|
def fetchone(self):
|
|
return self._row
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class FakeConn:
|
|
def __init__(self, rows, update_rowcount=1):
|
|
self.db = {"rows": rows, "update_rowcount": update_rowcount, "pending": {}}
|
|
self.rolled_back = self.committed = False
|
|
|
|
def cursor(self):
|
|
return FakeCursor(self.db)
|
|
|
|
def commit(self):
|
|
for row in self.db["rows"].values():
|
|
row.update(self.db["pending"])
|
|
self.committed = True
|
|
|
|
def rollback(self):
|
|
self.db["pending"].clear(); self.rolled_back = True
|
|
|
|
|
|
ALLOWED = r"D:\PROJETOS\GTISISTEMAS\Projeto\Banco\CONSTRUCAO.FDB"
|
|
|
|
|
|
def test_executor_allowlist_is_exact_and_field_set_is_the_contract():
|
|
ex = load_executor()
|
|
assert ex._LOCAL_COMPLETION_FIELDS == set(WRITE_FIELDS)
|
|
allow = ex.load_local_allowlist()
|
|
assert len(allow) == 5
|
|
for path in (ALLOWED, r"D:\PROJETOS\GTISISTEMAS\Projeto\TESTESOFICIAIS.FDB"):
|
|
ex.assert_local_completion_target(path, allow)
|
|
for bad in (r"D:\PROJETOS\GTISISTEMAS\Banco\GTI.FDB", r"D:\PROJETOS\GTISISTEMAS\Projeto\Banco\OUTRA.FDB", r"D:\PROJETOS\GTISISTEMAS\Projeto\Banco\CONSTRUCAO.FDB.bak", r"D:\PROJETOS\GTISISTEMAS\Projeto\Banco"):
|
|
with pytest.raises(ex.WriteRefused):
|
|
ex.assert_local_completion_target(bad, allow)
|
|
|
|
|
|
def test_guarded_product_update_success_drift_rowcount_and_refusals():
|
|
ex = load_executor()
|
|
allow = ex.load_local_allowlist()
|
|
conn = FakeConn({"000001": {"IPI": None, "CEST": "1234567"}})
|
|
rec = ex.execute_guarded_product_update(conn, database_path=ALLOWED, product_code="000001", changes={"IPI": (None, 3.9), "CEST": ("1234567", "7654321")}, now="t", allowlist=allow)
|
|
assert rec["rowcount"] == 1 and rec["fields"]["IPI"]["after"] == 3.9 and conn.committed
|
|
with pytest.raises(ex.WriteRefused, match="drift"):
|
|
ex.execute_guarded_product_update(FakeConn({"000001": {"IPI": 1.0}}), database_path=ALLOWED, product_code="000001", changes={"IPI": (None, 3.9)}, now="t", allowlist=allow)
|
|
c2 = FakeConn({"000001": {"IPI": None}}, update_rowcount=2)
|
|
with pytest.raises(ex.WriteRefused, match="rowcount"):
|
|
ex.execute_guarded_product_update(c2, database_path=ALLOWED, product_code="000001", changes={"IPI": (None, 3.9)}, now="t", allowlist=allow)
|
|
assert c2.rolled_back
|
|
with pytest.raises(ex.WriteRefused, match="allowlist"):
|
|
ex.execute_guarded_product_update(conn, database_path=r"D:\PROJETOS\GTISISTEMAS\Banco\GTI.FDB", product_code="000001", changes={"IPI": (None, 1)}, now="t", allowlist=allow)
|
|
with pytest.raises(ex.WriteRefused, match="contract write target"):
|
|
ex.execute_guarded_product_update(conn, database_path=ALLOWED, product_code="000001", changes={"CODIGO": ("000001", "9")}, now="t", allowlist=allow)
|
|
with pytest.raises(ex.WriteRefused, match="null"):
|
|
ex.execute_guarded_product_update(conn, database_path=ALLOWED, product_code="000001", changes={"IPI": (None, None)}, now="t", allowlist=allow)
|
|
with pytest.raises(ex.WriteRefused, match="only C000025"):
|
|
ex.execute_guarded_product_update(conn, database_path=ALLOWED, product_code="000001", changes={"IPI": (None, 1)}, now="t", allowlist=allow, table="NFCE_ITENS")
|