GTI_GERENCIADOR_FISCAL/fiscal/central/tests/test_fiscal_completion.py

"""Post-NCM fiscal completion: wires the existing (previously dormant) field_contract.resolve_product into the live pipeline. Reuses the SAME engine,
the SAME FiscalContextEntry table the blocker-resolution workflow already writes to, and company_uf/company_crt already auto-discovered from every
heartbeat - no new rule, no guessing."""
from sqlalchemy import select
from sqlalchemy.orm import Session

from central.app.enrichment import fiscal_completion as fcomp
from central.app.field_contract import WRITE_FIELDS
from central.app.models import Agent, Base, CfopEntry, Dataset, FiscalCodeEntry, FiscalContextEntry, Installation, Tenant


def _seed_codes(session, ncm="22030000"):
    ds = Dataset(kind="NFE_MOC_TABLES", provider="x", sha256="a" * 64, version_label="v", act="a", source_url="u", status="ACTIVE", record_count=0)
    session.add(ds); session.flush()
    for domain, codes in (("CSOSN", ["101", "102", "103", "300", "400", "500"]), ("CST_ICMS", ["00", "20", "40", "41", "60"]),
                          ("CST_PIS", ["01", "49", "98"]), ("CST_COFINS", ["01", "49", "98"]), ("ORIGEM_MERCADORIA", list("012345678"))):
        for code in codes:
            session.add(FiscalCodeEntry(dataset_id=ds.id, domain=domain, code=code, description=f"{domain} {code}", source_section="test"))
    cfop_ds = Dataset(kind="CFOP", provider="x", sha256="b" * 64, version_label="v", act="a", source_url="u", status="ACTIVE", record_count=0)
    session.add(cfop_ds); session.flush()
    for code in ("5101", "5102", "5401", "1102", "1101"):
        session.add(CfopEntry(dataset_id=cfop_ds.id, code=code, title=f"CFOP {code}"))
    session.commit()


def _installation(session, uf="BA", crt="1", company_key="000001", contribuinte_ipi=None):
    tenant = Tenant(id=1, name="Acme"); session.add(tenant); session.flush()
    agent = Agent(tenant_id=1, name="a", token_hash="x" * 64); session.add(agent); session.flush()
    inst = Installation(agent_id=agent.id, tenant_id=1, installation_id="11111111-1111-4111-8111-111111111111", company_uf=uf, company_crt=crt, company_code=company_key,
                        company_contribuinte_ipi=contribuinte_ipi)
    session.add(inst); session.flush()
    return inst


def _ctx(session, tenant_id, company_key, entries):
    for e in entries:
        session.add(FiscalContextEntry(tenant_id=tenant_id, company_key=company_key, scope=e[0], scope_key=e.get("scope_key", "") if isinstance(e, dict) else "", key=e[1] if not isinstance(e, dict) else e["key"],
                                       value=e[2] if not isinstance(e, dict) else e["value"], source="OPERATOR"))
    session.commit()


def env():
    from sqlalchemy import create_engine
    from sqlalchemy.pool import StaticPool
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(engine)
    return Session(engine)


def test_field_shapes():
    assert fcomp.is_writable_shape("CFOP", "5102") and not fcomp.is_writable_shape("CFOP", "51A2") and not fcomp.is_writable_shape("CFOP", "510")
    assert fcomp.is_writable_shape("CST", "000") and fcomp.is_writable_shape("CST", "0000")
    assert fcomp.is_writable_shape("ALIQUOTA", "17.5") and fcomp.is_writable_shape("ALIQ_PIS", "1,65") and not fcomp.is_writable_shape("ALIQUOTA", "abc")
    assert fcomp.is_writable_shape("SITUACAO_TRIBUTARIA", "t") and not fcomp.is_writable_shape("SITUACAO_TRIBUTARIA", "X")
    assert fcomp.is_writable_shape("IAT", "A") and fcomp.is_writable_shape("IPPT", "P") and fcomp.is_writable_shape("PISCOFINS", "N")
    assert not fcomp.is_writable_shape("CFOP", "")


def test_company_auto_discovered_from_heartbeat_no_guessing():
    s = env()
    inst = _installation(s)
    assert fcomp.build_company(inst) == {"uf": "BA", "crt": "1"}


def test_resolve_fiscal_fields_with_full_context_fills_independent_safe_fields():
    s = env()
    _seed_codes(s)
    inst = _installation(s)
    import json
    pis_cofins_profile = json.dumps({"pis_in_cst": "01", "pis_out_cst": "01", "pis_rate": "1.65", "cofins_in_cst": "01", "cofins_out_cst": "01", "cofins_rate": "7.6"})
    _ctx(s, 1, "000001", [
        ("COMPANY", "product_role", "RESALE"), ("COMPANY", "icms_treatment", "TAXED"), ("COMPANY", "origin", "0"),
        ("COMPANY", "pis_cofins_profile", pis_cofins_profile), ("COMPANY", "piscofins", "S"), ("COMPANY", "rounding", "A"),
        ("COMPANY", "store_ipi_profile", "LEAVE_NOT_APPLICABLE"), ("COMPANY", "icms_rate", "17"), ("COMPANY", "entry_scope", "INTRA"),
    ])
    out = fcomp.resolve_fiscal_fields(s, product_code="X1", description="CERVEJA LATA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1)
    safe = {f: v["value"] for f, v in out.items() if v["status"] == "AUTO_FILL_SAFE"}
    assert "CSOSN" in safe and safe["CSOSN"] == "102"
    assert "PISCOFINS" in safe or out["PISCOFINS"]["status"] == "ALREADY_SET"
    for f, v in out.items():
        assert v["status"] in ("AUTO_FILL_SAFE", "ALREADY_SET", "NOT_APPLICABLE", "NEEDS_REVIEW", "INSUFFICIENT_EVIDENCE"), f


def test_field_independence_one_missing_context_does_not_block_others():
    s = env()
    _seed_codes(s)
    inst = _installation(s)
    _ctx(s, 1, "000001", [("COMPANY", "product_role", "RESALE"), ("COMPANY", "rounding", "A")])   # icms_treatment / pis_cofins_profile intentionally missing
    out = fcomp.resolve_fiscal_fields(s, product_code="X2", description="CERVEJA LATA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1)
    assert out["IAT"]["status"] == "AUTO_FILL_SAFE" and out["IAT"]["value"] == "A"                # resolved independently
    assert out["CSOSN"]["status"] == "NEEDS_REVIEW" and out["CSOSN"]["missing_key"] == "icms_treatment"
    assert out["COD_ENTRADA_PIS"]["status"] == "NEEDS_REVIEW" and out["COD_ENTRADA_PIS"]["missing_key"] == "pis_cofins_profile"


def test_contribuinte_ipi_auto_discovered_from_installation_wins_over_operator_context():
    """Installation.company_contribuinte_ipi (auto-discovered at heartbeat from C000004.CONTRIBUINTE_IPI) is an ERP fact; a FiscalContextEntry
    operator answer is only a fallback for an Agent version that hasn't reported it yet. ERP fact must win when both exist."""
    s = env()
    _seed_codes(s)
    inst = _installation(s, contribuinte_ipi="S")
    _ctx(s, 1, "000001", [("COMPANY", "contribuinte_ipi", "N")])       # stale/wrong operator answer - the ERP fact must override it
    out = fcomp.resolve_fiscal_fields(s, product_code="X9", description="MAQUINA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1)
    assert out["CST_IPI"]["status"] == "NEEDS_REVIEW"           # contribuinte_ipi=S alone doesn't fill CST_IPI (still needs a declared value), but it's no longer NOT_APPLICABLE
    assert out["CST_IPI"]["reason"] != "contribuinte de IPI não confirmado"    # proves the ERP fact (S) was used, not "unknown"


def test_targeted_product_evidence_feeds_role_inference_st_withheld_never_a_full_scan():
    """`ev` (firebird_reader.read_product_evidence's targeted, per-product recipes/composed/purchases/stock) reaches field_contract.resolve_product
    the same way the full-scan `read_evidence` already does for sync jobs - just scoped to ONE product, never the whole catalog."""
    s = env()
    _seed_codes(s)
    inst = _installation(s)
    out = fcomp.resolve_fiscal_fields(s, product_code="X10", description="BOLO CASEIRO", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1,
                                      ev={"recipes": {"X10"}, "composed": set(), "purchases": {}, "stock_entry": {}, "stock_in": {}})
    assert out["CFOP"]["status"] == "NEEDS_REVIEW"          # OWN_PRODUCTION now provable from the recipe evidence alone - no product_role question, only the CFOP direction/context remains
    assert out["CFOP"]["missing_key"] != "product_role"


def test_extra_columns_feed_field_contracts_own_role_inference_no_operator_question_needed():
    """PRODUTO_SERVICO is a real C000025 column field_contract.infer_role already reads as objective ERP evidence (no ctx.get("product_role")
    needed). Proves `extra_columns` (the Agent's classification columns) actually reach `resolve_product`: with PRODUTO_SERVICO set, CFOP is
    blocked on the SERVICE-specific `service_cfop` key, not the generic `product_role` key a fully-unknown product would need."""
    s = env()
    _seed_codes(s)
    inst = _installation(s)
    out = fcomp.resolve_fiscal_fields(s, product_code="X7", description="CONSULTORIA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1,
                                      extra_columns={"PRODUTO_SERVICO": "1"})
    assert out["CFOP"]["status"] == "NEEDS_REVIEW" and out["CFOP"]["missing_key"] == "service_cfop"


def test_prod_cozinha_string_from_agent_json_payload_is_normalized_to_int_for_infer_role():
    """The Agent sends `current`/extra columns as JSON strings ("1", not 1). field_contract.infer_role compares PROD_COZINHA == 1 (a real
    Firebird SMALLINT). Without normalization the kitchen-routing signal would silently never fire."""
    s = env()
    _seed_codes(s)
    inst = _installation(s)
    out = fcomp.resolve_fiscal_fields(s, product_code="X8", description="PRATO FEITO", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1,
                                      extra_columns={"PROD_COZINHA": "1"})
    assert out["CFOP"]["status"] == "NEEDS_REVIEW"    # PROD_COZINHA=1 alone is UNKNOWN (not proof), but it must be READ, not silently ignored as a non-numeric string
    assert out["CFOP"]["missing_key"] == "product_role"


def test_cst_ipi_unknown_contribuinte_ipi_is_needs_review_not_a_silent_not_applicable():
    """C000004.CONTRIBUINTE_IPI is not yet sent at heartbeat - `company` never carries it unless an operator answered the context key. field_contract's
    own rule reads a missing fact the same as a proven "N" (NOT_APPLICABLE); this module must not let that false negative through silently."""
    s = env()
    _seed_codes(s)
    inst = _installation(s)                                       # no contribuinte_ipi context entry: genuinely unknown
    out = fcomp.resolve_fiscal_fields(s, product_code="X5", description="CERVEJA LATA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1)
    assert out["CST_IPI"]["status"] == "NEEDS_REVIEW" and out["CST_IPI"]["missing_key"] == "contribuinte_ipi"
    _ctx(s, 1, "000001", [("COMPANY", "contribuinte_ipi", "N")])
    out = fcomp.resolve_fiscal_fields(s, product_code="X6", description="CERVEJA LATA", current={}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1)
    assert out["CST_IPI"]["status"] == "NOT_APPLICABLE"           # once the fact is genuinely proven "N", the engine's own NA decision is trusted again


def test_not_official_ncm_blocks_everything_with_explicit_reason():
    s = env()
    inst = _installation(s)
    out = fcomp.resolve_fiscal_fields(s, product_code="X3", description="X", current={}, ncm="00000000", ncm_official=False, installation=inst, tenant_id=1)
    assert all(v["status"] == "INSUFFICIENT_EVIDENCE" for v in out.values())


def test_existing_value_preserved_not_overwritten():
    """CSOSN 500 declares ICMS already withheld by ST - a DIFFERENT treatment than a plain TAXED declaration (CSOSN_FAMILY in field_contract.py
    maps icms_treatment=ST_WITHHELD -> {"500"} and icms_treatment=TAXED -> {"101","102","103"}). With the matching treatment declared, the engine
    proves the existing value correct and does not touch it."""
    s = env()
    _seed_codes(s)
    inst = _installation(s)
    _ctx(s, 1, "000001", [("COMPANY", "product_role", "RESALE"), ("COMPANY", "icms_treatment", "ST_WITHHELD"), ("COMPANY", "origin", "0")])
    out = fcomp.resolve_fiscal_fields(s, product_code="X4", description="CERVEJA LATA", current={"CSOSN": "500"}, ncm="22030000", ncm_official=True, installation=inst, tenant_id=1)
    assert out["CSOSN"]["status"] == "ALREADY_SET"


def test_coverage_status_complete_vs_safe_blocked():
    complete = {f: {"status": "ALREADY_SET"} for f in fcomp.FISCAL_COMPLETION_FIELDS}
    assert fcomp.coverage_status(complete) == "COMPLETE"
    blocked = dict(complete, CFOP={"status": "NEEDS_REVIEW"})
    assert fcomp.coverage_status(blocked) == "SAFE_BLOCKED"


def test_all_write_fields_covered_by_the_module_except_the_explicit_exclusions():
    excluded = {"CLASSIFICACAO_FISCAL", "CEST", "IPI", "COD_ORIGEM_MERCADORIA", "EX_TIPI"}
    assert set(WRITE_FIELDS) - excluded == set(fcomp.FISCAL_COMPLETION_FIELDS)