gustavosantos/GTI_GERENCIADOR_FISCAL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: RTC observed live with provenance, non-circular PIS/COFINS applicability, canonical CST normalization (1.1.8)

Browse Source 
- RTC: Agent reads ALL PARAMETROS rows every sync (conflict => off), reports rtc_source/rtc_observed_at/rtc_param_rows; legal_effective_context is informational only
- PIS/COFINS: applicability from CRT + PISCOFINS flag + incidence only (never from empty fields); Simples w/o special treatment NOT_APPLICABLE, taxed regime with explicit N OPTIONAL, unreadable declaration NEEDS_REVIEW
- CST: one digit with an official zero-padded form is VALID_AFTER_NORMALIZATION (non-blocking warning), nothing is written to the ERP
- tests: RTC toggle 0->1->0 across syncs (Agent + Hub stale closing), non-circularity, normalization, panel provenance

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
gti
2026-09-24 17:46:45 -03:00
co-authored by Claude Sonnet 5
parent 1430042322
commit 88f4318072
18 changed files with 679 additions and 814 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/tests/test_unified_fiscal.py
+47 -7
View File
@@ -16,8 +16,8 @@ CLASSES = [("000001", "000", date(2025, 5, 5), None), ("200003", "200", date(202
class FakeCursor:
"""READ ONLY fake: answers only SELECTs and records every statement so tests can prove nothing else ran."""
def __init__(self, param=0, schema=SCHEMA_ROWS, products=PRODUCTS, classes=CLASSES, cst=(("000",), ("200",))):
self.param, self.schema, self.products, self.classes, self.cst, self.statements, self._rows = param, schema, products, classes, cst, [], []
def __init__(self, param=0, params=None, schema=SCHEMA_ROWS, products=PRODUCTS, classes=CLASSES, cst=(("000",), ("200",))):
self.param, self.params, self.schema, self.products, self.classes, self.cst, self.statements, self._rows = param, params, schema, products, classes, cst, [], []
def execute(self, sql, *args):
self.statements.append(sql)
@@ -26,7 +26,7 @@ class FakeCursor:
if "FROM RDB$RELATION_FIELDS" in text:
self._rows = list(self.schema)
elif "FROM PARAMETROS" in text:
self._rows = [(self.param,)]
self._rows = [(v,) for v in (self.params if self.params is not None else [self.param])]
elif "RDB$RELATION_CONSTRAINTS" in text:
self._rows = [(1,)]
elif "FROM RTC_CCLASSTRIB" in text:
@@ -62,7 +62,7 @@ def test_rtc_off_in_the_erp_is_out_of_scope_and_never_blocks():
out = uf.assess_from_connection(conn, {"crt": "1"}, today=date(2026, 9, 24))
assert out["context"]["rtc_enabled"] is False and out["context"]["rtc_supported"] is True and out["context"]["rtc_catalog_valid"] is True
assert not [b for f in out["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
assert kinds(out["findings"]["1"]) == [] and kinds(out["findings"]["2"]) == [] and kinds(out["findings"]["3"]) == ["COFINS_CST_INVALID", "PIS_CST_INVALID"]
assert kinds(out["findings"]["1"]) == [] and kinds(out["findings"]["2"]) == [] and kinds(out["findings"]["3"]) == [] # Simples + explicit "no special treatment": not applicable, typed "2" is only a warning
assert all(s.strip().upper().startswith("SELECT") for s in conn.cur.statements) # read-only: SELECT statements only
@@ -70,19 +70,19 @@ def test_rtc_on_in_the_erp_brings_cclasstrib_into_the_same_findings():
out = uf.assess_from_connection(FakeConn(param=1), {"crt": "1"}, today=date(2026, 9, 24))
assert out["context"]["rtc_enabled"] is True
assert kinds(out["findings"]["1"]) == ["CCLASSTRIB_MISSING"] and kinds(out["findings"]["2"]) == []
assert kinds(out["findings"]["3"]) == ["CCLASSTRIB_MISSING", "COFINS_CST_INVALID", "PIS_CST_INVALID"] # one queue, one verdict per product
assert kinds(out["findings"]["3"]) == ["CCLASSTRIB_MISSING"] # one queue, one verdict per product
def test_simulation_flag_never_touches_the_erp_parameter():
conn = FakeConn(param=0)
on = uf.assess_from_connection(conn, {"crt": "1"}, rtc_override=True)
assert on["context"]["rtc_enabled"] is True and kinds(on["findings"]["1"]) == ["CCLASSTRIB_MISSING"]
assert on["context"]["rtc_param"] == 0 and not any("UPDATE" in s.upper() for s in conn.cur.statements)
assert on["context"]["rtc_param"] == "0" and on["context"]["rtc_source"] == "SIMULATION" and not any("UPDATE" in s.upper() for s in conn.cur.statements)
def test_old_schema_without_rtc_columns_reports_unsupported_and_still_evaluates_pis_cofins():
old = [("C000025", c) for c in ("CODIGO", "COD_SAIDA_PIS", "PISCOFINS", "INCIDENCIA_PISCOFINS")]
rows = [("1", None, "N", "NENHUMA"), ("2", "9", "N", "NENHUMA")]
rows = [("1", None, "N", "NENHUMA"), ("2", "049", "N", "NENHUMA")]
out = uf.assess_from_connection(FakeConn(schema=old, products=rows, classes=[], cst=()), {"crt": "3"})
assert out["context"]["rtc_supported"] is False and out["context"]["rtc_enabled"] is False
assert kinds(out["findings"]["1"]) == [] and kinds(out["findings"]["2"]) == ["PIS_CST_INVALID"]
@@ -108,3 +108,43 @@ def test_input_fingerprint_changes_with_the_fiscal_values_and_the_rtc_switch():
changed = [("1", None, None, "99", None, None, None, None, "N", "NENHUMA")] + PRODUCTS[1:]
c = uf.assess_from_connection(FakeConn(param=0, products=changed), {"crt": "1"})["findings"]["1"]["input_fingerprint"]
assert len({a, b, c}) == 3 and re.fullmatch(r"[0-9a-f]{64}", a)
def test_rtc_state_is_read_from_this_database_now_with_provenance_and_timestamp():
out = uf.assess_from_connection(FakeConn(param=1), {"crt": "1"})
ctx = out["context"]
assert ctx["rtc_enabled"] is True and ctx["erp_rtc_enabled"] is True and ctx["rtc_source"] == "PARAMETROS.USAR_REFORMA_TRIBUTARIA" and ctx["rtc_param"] == "1" and ctx["rtc_param_rows"] == 1
assert ctx["rtc_param_conflict"] is False and ctx["rtc_observed_at"].endswith("+00:00")
assert ctx["legal_effective_context"]["informational_only"] is True and ctx["legal_effective_context"]["crt"] == "1"
def test_toggle_between_syncs_is_followed_in_both_directions_without_any_cache():
conn = FakeConn(param=0)
first = uf.assess_from_connection(conn, {"crt": "1"})
assert first["context"]["rtc_enabled"] is False and not [b for f in first["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
conn.cur.param = 1 # the operator switches the reform on in the ERP between two syncs
second = uf.assess_from_connection(conn, {"crt": "1"})
assert second["context"]["rtc_enabled"] is True and kinds(second["findings"]["1"]) == ["CCLASSTRIB_MISSING"]
assert second["findings"]["1"]["input_fingerprint"] != first["findings"]["1"]["input_fingerprint"] # a new blocker fingerprint: the old (RTC off) state is not reused
conn.cur.param = 0 # and off again
third = uf.assess_from_connection(conn, {"crt": "1"})
assert third["context"]["rtc_enabled"] is False and not [b for f in third["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
assert third["findings"]["1"]["input_fingerprint"] == first["findings"]["1"]["input_fingerprint"]
def test_conflicting_or_repeated_parameter_rows_never_guess_on():
same = uf.assess_from_connection(FakeConn(params=[1, 1]), {"crt": "1"})["context"]
assert same["rtc_enabled"] is True and same["rtc_param_rows"] == 2 and same["rtc_param_conflict"] is False
conflict = uf.assess_from_connection(FakeConn(params=[1, 0]), {"crt": "1"})
assert conflict["context"]["rtc_enabled"] is False and conflict["context"]["rtc_param_conflict"] is True
assert not [b for f in conflict["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
assert uf.assess_from_connection(FakeConn(params=[]), {"crt": "1"})["context"]["rtc_enabled"] is False
def test_non_canonical_cst_warnings_are_reported_without_blocking():
rows = [("1", None, None, "4", None, None, "4", None, "N", "NENHUMA"), ("2", None, None, None, None, None, None, None, "N", "NENHUMA")]
out = uf.assess_from_connection(FakeConn(products=rows), {"crt": "3"})
assert kinds(out["findings"]["1"]) == [] and kinds(out["findings"]["2"]) == []
w = out["context"]["non_canonical_warnings"]
assert w["products"] == 1 and w["by_type"] == {"CST_COFINS_NON_CANONICAL_FORMAT": 1, "CST_PIS_NON_CANONICAL_FORMAT": 1} and w["samples"][0]["product_code"] == "1"
assert out["context"]["pis_cofins_applicability"] == {"OPTIONAL": 2}