fix: gate cClassTrib applicability by CRT and date
This commit is contained in:
@@ -235,9 +235,14 @@ def test_batch_limit_bounds_the_blast_radius():
|
||||
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
|
||||
kinds = {b["type"] for f in out["findings"].values() for b in f["blockers"]}
|
||||
assert "CCLASSTRIB_MISSING_UNRESOLVED" in kinds and "CCLASSTRIB_MISSING" not in kinds
|
||||
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():
|
||||
|
||||
@@ -66,17 +66,39 @@ def test_rtc_off_in_the_erp_is_out_of_scope_and_never_blocks():
|
||||
assert all(s.strip().upper().startswith("SELECT") for s in conn.cur.statements) # read-only: SELECT statements only
|
||||
|
||||
|
||||
def test_rtc_on_in_the_erp_brings_cclasstrib_into_the_same_findings():
|
||||
def test_rtc_on_in_the_erp_does_not_make_simples_cclasstrib_applicable_before_effective_date():
|
||||
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_UNRESOLVED"] and kinds(out["findings"]["2"]) == []
|
||||
assert kinds(out["findings"]["3"]) == ["CCLASSTRIB_MISSING_UNRESOLVED"] # one queue, one verdict per product
|
||||
assert out["context"]["cclasstrib_applicability"]["state"] == "NOT_APPLICABLE"
|
||||
assert not [b for f in out["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
|
||||
|
||||
|
||||
def test_rtc_on_reports_simples_cclasstrib_as_not_applicable_and_keeps_other_fields():
|
||||
products = [("1", None, None, None, None, None, None, None, "S", "MONOFÁSICA")]
|
||||
conn = FakeConn(param=1, products=products)
|
||||
|
||||
out = uf.assess_from_connection(conn, {"crt": "1"}, today=date(2026, 9, 25))
|
||||
|
||||
assert out["context"]["cclasstrib_applicability"]["state"] == "NOT_APPLICABLE"
|
||||
assert out["context"]["cclasstrib_applicability"]["source"] == "C000004.CRT + RTC effective-date rules"
|
||||
assert kinds(out["findings"]["1"]) == ["COFINS_CST_MISSING", "PIS_CST_MISSING"]
|
||||
assert all(sql.strip().upper().startswith("SELECT") for sql in conn.cur.statements)
|
||||
|
||||
|
||||
def test_cclasstrib_applicability_change_changes_product_input_fingerprint():
|
||||
simples = uf.assess_from_connection(FakeConn(param=1), {"crt": "1"}, today=date(2026, 9, 25))
|
||||
normal = uf.assess_from_connection(FakeConn(param=1), {"crt": "3"}, today=date(2026, 9, 25))
|
||||
|
||||
assert simples["findings"]["1"]["input_fingerprint"] != normal["findings"]["1"]["input_fingerprint"]
|
||||
assert kinds(simples["findings"]["1"]) == []
|
||||
assert any(blocker["category"] == "RTC" for blocker in normal["findings"]["1"]["blockers"])
|
||||
|
||||
|
||||
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_UNRESOLVED"]
|
||||
assert on["context"]["rtc_enabled"] is True and on["context"]["cclasstrib_applicability"]["state"] == "NOT_APPLICABLE"
|
||||
assert not [b for f in on["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
|
||||
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)
|
||||
|
||||
|
||||
@@ -96,7 +118,7 @@ def test_parameter_values_are_read_fail_closed(value, expected):
|
||||
|
||||
def test_catalog_with_orphan_or_duplicate_is_invalid_and_flagged_only_when_rtc_is_on():
|
||||
bad = FakeConn(param=1, classes=CLASSES + [("200003", "200", date(2025, 5, 5), None)]) # duplicate class
|
||||
on = uf.assess_from_connection(bad, {"crt": "1"})
|
||||
on = uf.assess_from_connection(bad, {"crt": "3"})
|
||||
assert on["context"]["rtc_catalog_valid"] is False and kinds(on["findings"]["2"]) == ["RTC_CATALOG_INVALID"]
|
||||
off = uf.assess_from_connection(FakeConn(param=0, classes=CLASSES + [("200003", "200", date(2025, 5, 5), None)]), {"crt": "1"})
|
||||
assert not [b for f in off["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
|
||||
@@ -115,6 +137,7 @@ def test_rtc_state_is_read_from_this_database_now_with_provenance_and_timestamp(
|
||||
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["cclasstrib_applicability"]["state"] == "NOT_APPLICABLE" and ctx["cclasstrib_applicability"]["source"] == "C000004.CRT + RTC effective-date rules"
|
||||
assert ctx["legal_effective_context"]["informational_only"] is True and ctx["legal_effective_context"]["crt"] == "1"
|
||||
|
||||
|
||||
@@ -124,7 +147,8 @@ def test_toggle_between_syncs_is_followed_in_both_directions_without_any_cache()
|
||||
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_UNRESOLVED"]
|
||||
assert second["context"]["rtc_enabled"] is True and second["context"]["cclasstrib_applicability"]["state"] == "NOT_APPLICABLE"
|
||||
assert not [b for f in second["findings"].values() for b in f["blockers"] if b["category"] == "RTC"]
|
||||
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"})
|
||||
|
||||
Reference in New Issue
Block a user