135 lines
8.1 KiB
Python
135 lines
8.1 KiB
Python
"""RTC (cClassTrib / CST IBS-CBS) and PIS/COFINS: capabilities, validators, applicability, legacy vs expanded coverage. Read-only."""
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parents[1] / "src"))
|
|
|
|
from gti_fiscal_agent import rtc_pis_cofins as r # noqa: E402
|
|
|
|
FULL_SCHEMA = {"C000025": {"CODIGO", "CCLASSTRIB", *r.PIS_COFINS_COLUMNS}, "RTC_CCLASSTRIB": {"CCLASSTRIB", "CSTIBSCBS"}, "RTC_CST": {"CSTIBSCBS"}}
|
|
CATALOG = {"classes": {"000001": "000", "200003": "200", "999999": None}, "cst": {"000", "200"}, "duplicates": 0}
|
|
|
|
|
|
def caps(schema=FULL_SCHEMA, param=0, catalog=CATALOG):
|
|
c = r.detect_rtc_capabilities(schema, param_value=param, catalog=catalog, physical_fk=True)
|
|
c["_PIS_COFINS_SCHEMA"] = all(col in schema.get("C000025", ()) for col in r.PIS_COFINS_COLUMNS[:6])
|
|
return c
|
|
|
|
|
|
def test_database_without_rtc_fails_closed_and_is_not_blocked():
|
|
c = caps({"C000025": {"CODIGO"}}, param=None, catalog=None)
|
|
assert not any(c[k] for k in ("HAS_PRODUCT_CCLASSTRIB", "HAS_RTC_CCLASSTRIB_TABLE", "HAS_RTC_CST_TABLE", "HAS_RTC_RELATIONSHIP", "RTC_CATALOG_VALID"))
|
|
assert r.rtc_applicability(c) == "NOT_SUPPORTED"
|
|
found = r.expanded_findings({"CODIGO": 1}, c, None)
|
|
assert found["blockers"] == [] and found["unknown"] == []
|
|
|
|
|
|
def test_partial_rtc_column_without_catalog_tables_is_not_a_relationship():
|
|
c = caps({"C000025": {"CODIGO", "CCLASSTRIB"}}, param=1, catalog=None)
|
|
assert c["HAS_PRODUCT_CCLASSTRIB"] and not c["HAS_RTC_RELATIONSHIP"] and not c["RTC_CATALOG_VALID"]
|
|
assert r.rtc_applicability(c) == "NOT_SUPPORTED"
|
|
|
|
|
|
def test_full_model_and_valid_catalog_capabilities():
|
|
good = {"classes": {"000001": "000"}, "cst": {"000"}, "duplicates": 0}
|
|
c = caps(catalog=good)
|
|
assert c["HAS_RTC_RELATIONSHIP"] and c["RTC_CATALOG_VALID"] and c["HAS_PHYSICAL_RTC_FK"] and c["RTC_ENABLED_BY_PARAM"] is False and c["LOCAL_CATALOG_ONLY"]
|
|
assert not caps(catalog=CATALOG)["RTC_CATALOG_VALID"] # a class pointing to a missing CST makes the catalog invalid
|
|
|
|
|
|
def test_cclasstrib_validator_states():
|
|
assert r.validate_cclasstrib(None, CATALOG) == "EMPTY" and r.validate_cclasstrib(" ", CATALOG) == "EMPTY"
|
|
assert r.validate_cclasstrib("12A456", CATALOG) == "INVALID_FORMAT" and r.validate_cclasstrib("12345", CATALOG) == "INVALID_FORMAT"
|
|
assert r.validate_cclasstrib("123456", CATALOG) == "NOT_IN_LOCAL_CATALOG"
|
|
assert r.validate_cclasstrib("200003", CATALOG) == "VALID_WITH_CST"
|
|
assert r.validate_cclasstrib("999999", CATALOG) == "CST_MISSING"
|
|
assert r.validate_cclasstrib("000001", {"classes": {"000001": None}, "cst": set()}) == "VALID_LOCAL"
|
|
assert r.validate_cclasstrib("000001", None) == "NOT_IN_LOCAL_CATALOG"
|
|
|
|
|
|
def test_pis_cofins_validators():
|
|
assert r.validate_pis_cofins_cst("") == "EMPTY" and r.validate_pis_cofins_cst(None) == "EMPTY"
|
|
assert r.validate_pis_cofins_cst("01") == "VALID" and r.validate_pis_cofins_cst("73") == "VALID"
|
|
assert r.validate_pis_cofins_cst("2") == "INVALID_FORMAT" and r.validate_pis_cofins_cst("10") == "INVALID_FORMAT"
|
|
assert r.validate_pis_cofins_rate("0,65") == "VALID" and r.validate_pis_cofins_rate("3") == "VALID" and r.validate_pis_cofins_rate("") == "EMPTY"
|
|
assert r.validate_pis_cofins_rate("abc") == "INVALID_RATE" and r.validate_pis_cofins_rate("150") == "INVALID_RATE"
|
|
ok = r.validate_pis_cofins({"PISCOFINS": "S", "INCIDENCIA_PISCOFINS": "MONOFASICA", "COD_SAIDA_PIS": "04", "COD_SAIDA_COFINS": "04"})
|
|
assert ok["FLAG_CONSISTENCY"] == "CONSISTENT"
|
|
bad = r.validate_pis_cofins({"PISCOFINS": "N", "INCIDENCIA_PISCOFINS": "NENHUMA", "COD_SAIDA_PIS": "01"})
|
|
assert bad["FLAG_CONSISTENCY"] == "INCONSISTENT"
|
|
|
|
|
|
def test_empty_pis_cofins_is_unknown_applicability_not_a_blocker():
|
|
found = r.expanded_findings({"CODIGO": 1, "PISCOFINS": "N", "INCIDENCIA_PISCOFINS": "NENHUMA"}, caps(), CATALOG)
|
|
assert found["blockers"] == [] and found["unknown"] == ["PIS_COFINS"] and found["rtc_applicability"] == "NOT_ENABLED_BY_ERP"
|
|
|
|
|
|
def test_populated_pis_cofins_is_applicable_and_valid_data_completes():
|
|
row = {"CODIGO": 1, "PISCOFINS": "S", "INCIDENCIA_PISCOFINS": "SUBST.TRIBUTARIA", "COD_ENTRADA_PIS": "73", "COD_SAIDA_PIS": "01", "ALIQ_PIS": "0,65",
|
|
"COD_ENTRADA_COFINS": "73", "COD_SAIDA_COFINS": "01", "ALIQ_COFINS": "3"}
|
|
found = r.expanded_findings(row, caps(), CATALOG)
|
|
assert found["blockers"] == [] and found["unknown"] == [] and found["pis_cofins_applicability"] == "APPLICABLE"
|
|
|
|
|
|
def test_invalid_values_where_applicable_create_specific_blockers():
|
|
row = {"CODIGO": 2, "PISCOFINS": "N", "INCIDENCIA_PISCOFINS": "NENHUMA", "COD_ENTRADA_PIS": "2", "COD_SAIDA_PIS": "2", "ALIQ_PIS": "2",
|
|
"COD_ENTRADA_COFINS": "2", "COD_SAIDA_COFINS": "2", "ALIQ_COFINS": "x"}
|
|
kinds = {k for k, _ in r.expanded_findings(row, caps(), CATALOG)["blockers"]}
|
|
assert {"PIS_CST_INVALID", "COFINS_CST_INVALID", "COFINS_RATE_INVALID", "PIS_COFINS_FLAG_INCONSISTENT"} <= kinds
|
|
|
|
|
|
def test_rtc_becomes_applicable_only_when_the_erp_enables_it():
|
|
on = caps(param=1)
|
|
assert r.rtc_applicability(on) == "APPLICABLE"
|
|
assert {k for k, _ in r.expanded_findings({"CODIGO": 1}, on, CATALOG)["blockers"]} == {"CCLASSTRIB_MISSING"}
|
|
assert {k for k, _ in r.expanded_findings({"CODIGO": 1, "CCLASSTRIB": "123456"}, on, CATALOG)["blockers"]} == {"CCLASSTRIB_NOT_IN_CATALOG"}
|
|
assert {k for k, _ in r.expanded_findings({"CODIGO": 1, "CCLASSTRIB": "999999"}, on, CATALOG)["blockers"]} == {"RTC_CST_INVALID"}
|
|
assert {k for k, _ in r.expanded_findings({"CODIGO": 1, "CCLASSTRIB": "12x"}, on, CATALOG)["blockers"]} == {"CCLASSTRIB_INVALID"}
|
|
assert not [k for k, _ in r.expanded_findings({"CODIGO": 1, "CCLASSTRIB": "200003"}, on, CATALOG)["blockers"] if k.startswith(("CCLASS", "RTC"))]
|
|
assert r.rtc_applicability(caps(param=None)) == "UNKNOWN_APPLICABILITY"
|
|
|
|
|
|
def test_legacy_and_expanded_coverage_are_reported_side_by_side_and_close():
|
|
products = [{"CODIGO": 1}, {"CODIGO": 2}, {"CODIGO": 3},
|
|
{"CODIGO": 4, "PISCOFINS": "S", "INCIDENCIA_PISCOFINS": "MONOFASICA", "COD_SAIDA_PIS": "01", "COD_SAIDA_COFINS": "01", "COD_ENTRADA_PIS": "73", "COD_ENTRADA_COFINS": "73"},
|
|
{"CODIGO": 5, "COD_SAIDA_PIS": "2", "PISCOFINS": "S"}]
|
|
states = {"1": "COMPLETE", "2": "BLOCKED_MISSING_EVIDENCE", "3": "FAILED_TECHNICAL", "4": "COMPLETE", "5": "CORRECTED"}
|
|
out = r.expanded_scope_summary(products, states, caps(), CATALOG)
|
|
assert out["legacy_scope"] == {"complete": 3, "safe_blocked": 1, "failed": 1}
|
|
# 1 unknown (no PIS/COFINS evidence), 1 complete (#4 valid), blocked = legacy-blocked #2 + #5 (invalid CST), failed #3
|
|
assert out["expanded_scope"] == {"complete": 1, "safe_blocked": 2, "failed": 1, "unknown_applicability": 1}
|
|
assert out["invariant_pass"] and out["blockers_by_type"].get("PIS_CST_INVALID") == 1 and out["local_catalog_only"]
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self, tables, rows):
|
|
self.tables, self.rows, self.result = tables, rows, []
|
|
|
|
def execute(self, sql):
|
|
assert sql.lstrip().upper().startswith("SELECT") # read-only by construction
|
|
if "RDB$RELATION_FIELDS" in sql:
|
|
self.result = [(t, c) for t, cols in self.tables.items() for c in cols]
|
|
elif "USAR_REFORMA_TRIBUTARIA" in sql:
|
|
self.result = [(0,)]
|
|
elif "RDB$RELATION_CONSTRAINTS" in sql:
|
|
self.result = [(1,)]
|
|
elif "FROM RTC_CCLASSTRIB" in sql:
|
|
self.result = [("000001", "000")]
|
|
elif "FROM RTC_CST" in sql:
|
|
self.result = [("000",)]
|
|
elif "FROM C000025" in sql:
|
|
self.result = self.rows
|
|
|
|
def fetchall(self):
|
|
return self.result
|
|
|
|
def fetchone(self):
|
|
return self.result[0] if self.result else None
|
|
|
|
|
|
def test_loader_reads_schema_catalog_and_param_with_select_only():
|
|
cur = FakeCursor({"C000025": ["CODIGO", "CCLASSTRIB"], "RTC_CCLASSTRIB": ["CCLASSTRIB", "CSTIBSCBS"], "RTC_CST": ["CSTIBSCBS"], "PARAMETROS": ["USAR_REFORMA_TRIBUTARIA"]}, [(1, None)])
|
|
schema, catalog, extras = r.load_rtc_context(cur)
|
|
assert catalog == {"classes": {"000001": "000"}, "cst": {"000"}, "duplicates": 0} and extras == {"param_value": 0, "physical_fk": True}
|
|
assert r.read_products_rtc_pis_cofins(cur, schema) == [{"CODIGO": 1, "CCLASSTRIB": None}]
|