feat: audited-decision gate for PIS/COFINS and CFOP_ENTRADA NOT_APPLICABLE/OPERATION_TIME
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>
This commit is contained in:
@@ -125,7 +125,9 @@ def test_cfop_entrada_mixed_scope_is_operation_dependent_and_never_invents_a_cfo
|
||||
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"]
|
||||
assert e["state"] == MANUAL and e["proposed"] is None and e.get("input_key") is None
|
||||
# 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"
|
||||
@@ -175,6 +177,30 @@ def test_cst_ipi_pis_cofins_incidence_iat_ippt():
|
||||
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
|
||||
|
||||
|
||||
|
||||
@@ -27,6 +27,11 @@ def test_validation_rules():
|
||||
{"scope": "COMPANY", "key": "icms_rate", "value": "abc"}, {"scope": "COMPANY", "key": "nope", "value": "1"}):
|
||||
assert validate_entry(bad)[1]
|
||||
assert validate_entry({"scope": "COMPANY", "key": "pis_cofins_profile", "value": "{\"pis_out_cst\": \"49\"}"})[1] is None
|
||||
# Missão 2 / Bloco 2.3: the two audited sentinels must validate (case-insensitive) alongside the JSON profile shape.
|
||||
ok, err = validate_entry({"scope": "COMPANY", "key": "pis_cofins_profile", "value": "not_applicable"})
|
||||
assert err is None and ok["value"] == "NOT_APPLICABLE"
|
||||
ok, err = validate_entry({"scope": "COMPANY", "key": "pis_cofins_profile", "value": "OPERATION_TIME"})
|
||||
assert err is None and ok["value"] == "OPERATION_TIME"
|
||||
|
||||
|
||||
def test_import_is_all_or_nothing_upserting_and_listed():
|
||||
|
||||
@@ -17,7 +17,7 @@ def test_health_is_unavailable_when_database_is_down():
|
||||
|
||||
def test_version_endpoint_reports_release_and_fiscal_versions():
|
||||
body = TestClient(create_app(database_probe=lambda: True)).get("/version").json()
|
||||
assert body["version"] == "1.3.9" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"]
|
||||
assert body["version"] == "1.3.10" and body["api_compat"] == "1.0" and body["engine_version"] and body["decision_version"]
|
||||
|
||||
|
||||
def test_production_refuses_the_dev_token(monkeypatch):
|
||||
|
||||
Reference in New Issue
Block a user