diff --git a/central/tests/test_kb_attest_wrong_gtin.py b/central/tests/test_kb_attest_wrong_gtin.py new file mode 100644 index 0000000..40032eb --- /dev/null +++ b/central/tests/test_kb_attest_wrong_gtin.py @@ -0,0 +1,84 @@ +"""Regression for the 2026-09-27 production incident: an ADMIN attested the right NCM on the WRONG (but very similar) KB product — "COCA ZERO LATA 350 ML" +instead of plain "COCA LATA 350 ML" — because both appeared as CONFLICT with an overlapping candidate list. The pipeline behaved correctly (no write for +the unresolved product); the mistake was purely a fiscal data-entry error. These tests lock in: (1) two locally-created Firebird rows with the SAME +description never share KB identity through anything but the KB product itself; (2) attesting one GTIN group never resolves an unrelated, distinct one, +even when their aliases and candidate NCMs overlap; (3) EAN_AUSENTE is never a terminal state — it becomes resolvable the moment its OWN GTIN group gets +verified; (4) a stale (rejected) attest target does not block a later, correct attest. +""" +from sqlalchemy import select + +from central.app.enrichment import history, kb +from central.app.enrichment.quota import ProviderGateway +from central.app.enrichment.resolver import resolve +from central.app.models import KbProduct, ProductEnrichment +from central.tests.test_enrichment_history import run as seed_history, staging +from central.tests.test_product_enrichment_api import H, setup +from central.tests.test_product_enrichment_resolver import env # noqa: F401 + + +def gtin(body12): + total = sum(int(d) * (3 if i % 2 == 0 else 1) for i, d in enumerate(reversed(body12))) + return body12 + str((10 - total % 10) % 10) + + +REGULAR, ZERO = gtin("789490001000"), gtin("789490070000") + + +def _seed_two_similar_products(env): + seed_history(env, [staging("RINCAO", REGULAR, "19059090", years={2024: 900, 2025: 900}, desc="COCA LATA 350 ML"), + staging("UNIAO", REGULAR, "22021000", years={2024: 20, 2025: 13}, desc="REFRIGERANTE COCA COLA LATA 350 ML")]) + seed_history(env, [staging("RINCAO", ZERO, "19059090", years={2024: 900, 2025: 900}, desc="COCA ZERO LATA 350 ML"), + staging("UNIAO", ZERO, "22021000", years={2024: 40, 2025: 30}, desc="REFRIGERANTE COCA COLA ZERO LATA 350 ML")]) + + +def test_attesting_the_wrong_similar_gtin_leaves_the_intended_product_unresolved(env): + _seed_two_similar_products(env) + cur = history.official_sets(env) + kb.attest(env, "0" + ZERO, "22021000", "Admin", lambda v: v in cur) # attest the WRONG one, as happened in production + result = resolve(env, ProviderGateway(env, []), gtin_raw="", description="COCA-COLA ORIGINAL LATA 350 ML", current={}) + assert result["fields"]["CLASSIFICACAO_FISCAL"]["status"] != "AUTO_FILL_SAFE" # still not resolved: attesting a DIFFERENT KB product proves nothing here + kb.attest(env, "0" + REGULAR, "22021000", "Admin", lambda v: v in cur) # attest the CORRECT one - same rule, correct target + result = resolve(env, ProviderGateway(env, []), gtin_raw="", description="COCA-COLA ORIGINAL LATA 350 ML", current={}) + assert result["state"] == "EXACT_KB_NAME_VERIFIED" and result["fields"]["CLASSIFICACAO_FISCAL"]["value"] == "22021000" + + +def test_ean_ausente_is_not_terminal_becomes_resolved_once_its_own_gtin_is_verified(monkeypatch): + client, engine = setup(monkeypatch) + from sqlalchemy.orm import Session + with Session(engine) as s: + _seed_two_similar_products(s) + s.commit() + body = {"product_code": "P1", "gtin": "", "description": "COCA-COLA ORIGINAL LATA 350 ML", "current": {}} + first = client.post("/v1/products/resolve", json=body, headers=H).json() + with Session(engine) as s: + row = s.scalars(select(ProductEnrichment).where(ProductEnrichment.product_code == "P1")).one() + assert row.ui_status == "EAN_AUSENTE" + cur = history.official_sets(s) + kb.attest(s, "0" + REGULAR, "22021000", "Admin", lambda v: v in cur) + from central.app.enrichment.status import mark_for_reevaluation + inst_id = row.installation_id + mark_for_reevaluation(s, inst_id) + s.commit() + assert s.scalars(select(ProductEnrichment).where(ProductEnrichment.product_code == "P1")).one().reevaluate is True + second = client.post("/v1/products/resolve", json=body, headers=H).json() + assert second["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "AUTO_FILL_SAFE" # EAN_AUSENTE was never a dead end + with Session(engine) as s: + assert s.scalars(select(ProductEnrichment).where(ProductEnrichment.product_code == "P1")).one().ui_status == "PRONTO_PARA_PREENCHER" + + +def test_two_local_products_with_the_identical_description_both_resolve_from_one_kb_verification(monkeypatch): + """Mirrors production 001437/001439: two DIFFERENT Firebird rows, same description, no EAN. One KB verification resolves BOTH (they resolve + independently, by name, not through any shared local-row identity).""" + client, engine = setup(monkeypatch) + from sqlalchemy.orm import Session + with Session(engine) as s: + _seed_two_similar_products(s) + cur = history.official_sets(s) + kb.attest(s, "0" + REGULAR, "22021000", "Admin", lambda v: v in cur) + s.commit() + for code in ("A1", "A2"): + body = {"product_code": code, "gtin": "", "description": "COCA-COLA ORIGINAL LATA 350 ML", "current": {}} + r = client.post("/v1/products/resolve", json=body, headers=H).json() + assert r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "AUTO_FILL_SAFE" and r["fields"]["CLASSIFICACAO_FISCAL"]["value"] == "22021000" + with Session(engine) as s: + assert s.query(KbProduct).count() == 2 # regular and ZERO stayed two distinct KB identities throughout