gti
7e83d17533
field_contract.resolve_product was only exercised by its own unit tests. /v1/products/resolve now calls it for every product whose NCM resolves, merging CFOP/CST/CSOSN/PIS-COFINS/CST_IPI/SITUACAO_TRIBUTARIA/IAT/IPPT decisions alongside CLASSIFICACAO_FISCAL/CEST/IPI - reusing the same engine, the same FiscalContextEntry table the blocker-resolution workflow already writes to, no duplicated rules. Closes three real transport/discovery gaps found while wiring it in: - Agent heartbeat never sent company_crt (env-var UF only) or company_contribuinte_ipi, so Installation.company_crt stayed permanently empty even though company_context.resolve() already proves both from C000004/EMISSOR for every coverage job. Resolved once, cached, retried only while unresolved. Migration 0032 adds Installation.company_contribuinte_ipi (NULL=unknown, never coerced to "N"). - The Agent's enrichment payload only ever reported 3 old fields (CLASSIFICACAO_FISCAL/CEST/IPI) as "current", so the Hub could never see the product's own CFOP/CST/CSOSN/... nor the classification columns (TIPO/PROD_COZINHA/PRODUTO_SERVICO/CODGRUPO) infer_role needs - separated into a read-only CURRENT_REPORT_FIELDS list, the write allowlist (PRODUCT_ENRICHMENT_WRITE_V1) untouched. - Added a targeted, product_code-scoped evidence query (firebird_reader.read_product_evidence: recipes/composed/purchases/stock, indexed WHERE, never a full-table scan) so st_withheld/product_role can be proven for a single ad-hoc resolve, not only during a full sync job. Verified read-only against the real database (D:\PROJETOS\GTISISTEMAS\Banco\GTI.FDB): CRT=1, CONTRIBUINTE_IPI=N, company_context.resolve() reaches HIGH confidence; infer_role correctly resolved RESALE/OWN_PRODUCTION for 5/8 sample real products with zero operator questions. 1121 tests passed (tests + central/tests + agent/tests), 0 failed. Version 1.3.7 (single GTI_FISCAL_VERSION shared by Hub and Agent per central/app/version.py's own design - never independently versioned).
135 lines
7.4 KiB
Python
135 lines
7.4 KiB
Python
import pytest
|
|
|
|
from gti_fiscal_agent.company_context import CompanyContextError, config_from_env, mask_cnpj, preflight_report, resolve
|
|
from gti_fiscal_agent.firebird_reader import parse_company_context
|
|
|
|
CN_A, CN_B, CN_C = "11111111000111", "22222222000122", "33333333000133"
|
|
|
|
|
|
def co(code, cnpj, uf="BA", crt="1", optante="S"):
|
|
return {"code": code, "cnpj": cnpj, "uf": uf, "crt": crt, "optante": optante, "name": "X"}
|
|
|
|
|
|
def ev(companies, emissor=None, docs=None):
|
|
return {"c000004": companies, "emissor": emissor if emissor is not None else [], "nfce_keys": docs or [], "nf_by_filial": {}}
|
|
|
|
|
|
def emis(cnpj=CN_A, uf="BA", crt="1"):
|
|
return {"cnpj": cnpj, "uf": uf, "crt": crt}
|
|
|
|
|
|
DOC_A = [{"cuf": "29", "cnpj": CN_A, "n": 100}]
|
|
|
|
|
|
def test_single_company_all_sources_agree_is_auto_resolved():
|
|
r = resolve(ev([co("000001", CN_A)], [emis()], DOC_A))
|
|
assert r["status"] == "PASS" and r["classification"] == "AUTO_RESOLVED" and r["company_uf"] == "BA" and r["company_crt"] == "1" and r["confidence"] == "HIGH"
|
|
assert "DOCUMENT_KEYS" in r["source_of_truth"]["uf"]
|
|
|
|
|
|
def test_multiple_rows_resolved_only_through_emitter_identity_never_first_row():
|
|
rows = [co("000001", CN_B, "SP", None, "N"), co("000005", CN_A), co("000007", CN_C, "BA", None, "N")]
|
|
r = resolve(ev(rows, [emis(CN_A)], DOC_A))
|
|
assert r["status"] == "PASS" and r["company_code"] == "000005" and r["source_of_truth"]["company"] == "EMISSOR_CNPJ_MATCH"
|
|
|
|
|
|
def test_multiple_rows_without_disambiguation_requires_config():
|
|
rows = [co("000001", CN_A), co("000002", CN_B)]
|
|
r = resolve(ev(rows, []))
|
|
assert r["status"] == "FAIL" and r["classification"] == "CONFIG_REQUIRED" and r["company_code"] is None
|
|
assert resolve(ev(rows, [emis(CN_C)]))["classification"] == "CONFIG_REQUIRED" # emitter matches none
|
|
ok = resolve(ev(rows, [], []), {"company_code": "000002"})
|
|
assert ok["status"] == "PASS" and ok["company_code"] == "000002" and ok["classification"] == "CONFIGURED"
|
|
|
|
|
|
def test_configured_company_must_exist_and_agree_with_emitter():
|
|
rows = [co("000001", CN_A), co("000002", CN_B)]
|
|
assert resolve(ev(rows, [emis(CN_A)]), {"company_cnpj": "99999999999999"})["classification"] == "CONFIG_REQUIRED"
|
|
r = resolve(ev(rows, [emis(CN_A)], DOC_A), {"company_code": "000002"})
|
|
assert r["status"] == "FAIL" and r["classification"] == "CONFLICT" # selected company differs from the emitter: unexplained
|
|
|
|
|
|
def test_documents_only_from_another_cnpj_conflict_but_mixed_history_is_a_warning():
|
|
r = resolve(ev([co("000001", CN_A)], [emis()], [{"cuf": "29", "cnpj": CN_B, "n": 10}]))
|
|
assert r["status"] == "FAIL" and r["classification"] == "CONFLICT"
|
|
r = resolve(ev([co("000001", CN_A)], [emis()], [{"cuf": "29", "cnpj": CN_A, "n": 10}, {"cuf": "29", "cnpj": CN_B, "n": 1}]))
|
|
assert r["status"] == "PASS" and any("HISTORICAL_NFCE_KEYS_FROM_OTHER_CNPJ" in w for w in r["warnings"])
|
|
|
|
|
|
def test_uf_conflict_and_unknown():
|
|
assert resolve(ev([co("1", CN_A, "BA")], [emis(uf="SP")]))["classification"] == "CONFLICT"
|
|
assert resolve(ev([co("1", CN_A, "BA")], [emis()], [{"cuf": "35", "cnpj": CN_A, "n": 5}]))["classification"] == "CONFLICT" # key says SP
|
|
assert resolve(ev([co("1", CN_A, "")], [emis(uf="")]))["classification"] == "UNRESOLVED"
|
|
|
|
|
|
def test_crt_conflict_needs_explicit_source_and_records_it():
|
|
base = ev([co("1", CN_A, crt="3")], [emis(crt="1")], DOC_A)
|
|
r = resolve(base)
|
|
assert r["status"] == "FAIL" and r["classification"] == "CONFLICT" and r["crt_candidates"] == {"EMISSOR": "1", "C000004": "3"}
|
|
ok = resolve(base, {"crt_source": "EMISSOR"})
|
|
assert ok["status"] == "PASS" and ok["company_crt"] == "1" and ok["classification"] == "CONFIGURED" and ok["confidence"] == "MEDIUM"
|
|
assert resolve(base, {"crt_source": "C000004"})["company_crt"] == "3"
|
|
assert resolve(base, {"crt_source_invalid": True})["classification"] == "CONFIG_REQUIRED"
|
|
|
|
|
|
def test_crt_unknown_fails_and_single_source_is_flagged():
|
|
assert resolve(ev([co("1", CN_A, crt=None)], [emis(crt="")]))["classification"] == "UNRESOLVED"
|
|
r = resolve(ev([co("1", CN_A, crt=None)], [emis(crt="1")]))
|
|
assert r["status"] == "PASS" and "CRT_HAS_A_SINGLE_SOURCE" in r["warnings"]
|
|
|
|
|
|
def test_optante_inconsistency_is_a_warning_not_a_guess():
|
|
r = resolve(ev([co("1", CN_A, crt="3", optante="S")], [emis(crt="3")]))
|
|
assert r["status"] == "PASS" and any("OPTANTE_SIMPLES" in w for w in r["warnings"])
|
|
|
|
|
|
def test_contribuinte_ipi_is_auto_discovered_from_the_same_selected_c000004_row():
|
|
"""C000004.CONTRIBUINTE_IPI (real column, same row already proven to be THE company by CRT/UF resolution) closes the fiscal-completion gap
|
|
where CST_IPI could not be resolved without asking an operator for a fact the ERP already proves."""
|
|
row = co("000001", CN_A)
|
|
row["contribuinte_ipi"] = "N"
|
|
r = resolve(ev([row], [emis()], DOC_A))
|
|
assert r["company_contribuinte_ipi"] == "N" and r["source_of_truth"]["contribuinte_ipi"] == "C000004"
|
|
|
|
|
|
def test_contribuinte_ipi_absent_or_invalid_stays_unresolved_never_guessed():
|
|
r = resolve(ev([co("000001", CN_A)], [emis()], DOC_A)) # co() never sets contribuinte_ipi
|
|
assert r["company_contribuinte_ipi"] is None
|
|
row = co("000002", CN_B); row["contribuinte_ipi"] = "X" # garbage value: never trusted
|
|
r2 = resolve(ev([row], [emis(cnpj=CN_B)], [{"cuf": "29", "cnpj": CN_B, "n": 1}]))
|
|
assert r2["company_contribuinte_ipi"] is None
|
|
|
|
|
|
def test_full_cnpj_is_never_exposed():
|
|
r = resolve(ev([co("1", CN_A)], [emis()], DOC_A))
|
|
dump = str(preflight_report(r)) + str(r)
|
|
assert CN_A not in dump and mask_cnpj(CN_A) == "11**********11"
|
|
|
|
|
|
def test_read_company_context_fails_closed(monkeypatch):
|
|
from gti_fiscal_agent import company_context as cc, firebird_reader as fr
|
|
monkeypatch.setattr(cc, "preflight_company_context", lambda env: resolve(ev([co("1", CN_A), co("2", CN_B)], [])))
|
|
with pytest.raises(CompanyContextError):
|
|
fr.read_company_context({})
|
|
monkeypatch.setattr(cc, "preflight_company_context", lambda env: resolve(ev([co("1", CN_A)], [emis()], DOC_A)))
|
|
assert fr.read_company_context({})["uf"] == "BA"
|
|
|
|
|
|
def test_config_from_env_and_legacy_parser_unchanged():
|
|
assert config_from_env({"COMPANY_CODE": " 000005 ", "COMPANY_CNPJ": "11.111.111/0001-11", "COMPANY_CRT_SOURCE": "emissor"}) == {"company_code": "000005", "company_cnpj": "11111111000111", "crt_source": "EMISSOR", "crt_source_invalid": False, "uf_override": None}
|
|
assert config_from_env({"COMPANY_CRT_SOURCE": "x"})["crt_source_invalid"] is True
|
|
assert parse_company_context([("ba", "1")]) == {"uf": "BA", "crt": "1"}
|
|
|
|
|
|
def test_auto_crt_source_and_uf_override_semantics():
|
|
assert config_from_env({"COMPANY_CRT_SOURCE": "AUTO"}) ["crt_source"] is None and config_from_env({"COMPANY_CRT_SOURCE": "AUTO"})["crt_source_invalid"] is False
|
|
conflict = ev([co("1", CN_A, crt="3")], [emis(crt="1")], DOC_A)
|
|
assert resolve(conflict, config_from_env({"COMPANY_CRT_SOURCE": "AUTO"}))["status"] == "FAIL" # AUTO never hides a CRT conflict
|
|
good = ev([co("1", CN_A)], [emis()], DOC_A)
|
|
assert resolve(good, {"uf_override": "BA"})["status"] == "PASS"
|
|
r = resolve(good, {"uf_override": "SP"})
|
|
assert r["status"] == "FAIL" and r["classification"] == "CONFLICT" # override never replaces the resolver silently
|
|
blank = ev([co("1", CN_A, uf="")], [emis(uf="")])
|
|
r = resolve(blank, {"uf_override": "BA"})
|
|
assert r["status"] == "PASS" and "UF_FROM_EXPLICIT_OVERRIDE" in r["warnings"]
|