GTI_GERENCIADOR_FISCAL/agent/tests/test_unified_fiscal.py

import re
from datetime import date

import pytest

from gti_fiscal_agent import unified_fiscal as uf

pytestmark = pytest.mark.real_unified_fiscal

SCHEMA_ROWS = ([("C000025", c) for c in ("CODIGO", "CCLASSTRIB", "COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS", "PISCOFINS", "INCIDENCIA_PISCOFINS")]
               + [("RTC_CCLASSTRIB", c) for c in ("CCLASSTRIB", "CSTIBSCBS", "INICIO_VIGENCIA", "FIM_VIGENCIA")] + [("RTC_CST", "CSTIBSCBS")] + [("PARAMETROS", "USAR_REFORMA_TRIBUTARIA")])
PRODUCT_COLS = ("CODIGO", "CCLASSTRIB", "COD_ENTRADA_PIS", "COD_SAIDA_PIS", "ALIQ_PIS", "COD_ENTRADA_COFINS", "COD_SAIDA_COFINS", "ALIQ_COFINS", "PISCOFINS", "INCIDENCIA_PISCOFINS")
PRODUCTS = [("1", None, None, None, None, None, None, None, "N", "NENHUMA"), ("2", "200003", None, "04", None, None, "04", None, "S", "MONOFÁSICA"), ("3", None, "2", "2", "2", "2", "2", "2", "N", "NENHUMA")]
CLASSES = [("000001", "000", date(2025, 5, 5), None), ("200003", "200", date(2025, 5, 5), None)]


class FakeCursor:
    """READ ONLY fake: answers only SELECTs and records every statement so tests can prove nothing else ran."""
    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)
        text = " ".join(sql.split()).upper()
        assert text.startswith("SELECT"), f"non-SELECT statement: {sql}"
        if "FROM RDB$RELATION_FIELDS" in text:
            self._rows = list(self.schema)
        elif "FROM PARAMETROS" in text:
            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:
            self._rows = list(self.classes)
        elif "FROM RTC_CST" in text:
            self._rows = list(self.cst)
        elif "FROM C000025" in text:
            self._rows = list(self.products)
        else:
            raise AssertionError(sql)

    def fetchall(self):
        return self._rows

    def fetchone(self):
        return self._rows[0] if self._rows else None


class FakeConn:
    def __init__(self, **kw):
        self.cur = FakeCursor(**kw)

    def cursor(self):
        return self.cur


def kinds(found):
    return sorted({b["type"] for b in found["blockers"]})


def test_rtc_off_in_the_erp_is_out_of_scope_and_never_blocks():
    conn = FakeConn(param=0)
    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"]) == []          # 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


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_UNRESOLVED"] and kinds(out["findings"]["2"]) == []
    assert kinds(out["findings"]["3"]) == ["CCLASSTRIB_MISSING_UNRESOLVED"]       # 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_UNRESOLVED"]
    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", "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"]


@pytest.mark.parametrize("value,expected", [(None, False), (0, False), ("0", False), (1, True), ("1", True), ("S", True), ("N", False)])
def test_parameter_values_are_read_fail_closed(value, expected):
    caps = uf.detect_rtc_capabilities({"C000025": {"CCLASSTRIB"}, "RTC_CCLASSTRIB": {"CCLASSTRIB", "CSTIBSCBS"}, "RTC_CST": {"CSTIBSCBS"}}, param_value=value)
    assert caps["RTC_ENABLED_BY_PARAM"] is 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"})
    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"]


def test_input_fingerprint_changes_with_the_fiscal_values_and_the_rtc_switch():
    a = uf.assess_from_connection(FakeConn(param=0), {"crt": "1"})["findings"]["1"]["input_fingerprint"]
    b = uf.assess_from_connection(FakeConn(param=1), {"crt": "1"})["findings"]["1"]["input_fingerprint"]
    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_UNRESOLVED"]
    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}