142 lines
9.8 KiB
Python
142 lines
9.8 KiB
Python
"""Visible status, re-evaluation (panel / Reprocessar / Sincronizar / KB decision), operator attestation, deterministic name-only match, RTC != Enrichment."""
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from central.app.enrichment import api as enrich_api, history, kb, status
|
|
from central.app.enrichment.quota import ProviderGateway
|
|
from central.app.enrichment.resolver import resolve
|
|
from central.app.jobs import create_job
|
|
from central.app.models import Agent, AuditEvent, Installation, KbKnowledge, KbProduct, KbPromotionAudit, ProductEnrichment
|
|
from central.tests.test_control_panel_pages import _panel_client
|
|
from central.tests.test_enrichment_history import G, run as seed_history, staging
|
|
from central.tests.test_product_enrichment_api import H, _second_agent, resolve as api_resolve, setup
|
|
from central.tests.test_product_enrichment_resolver import COSMOS, Fake, G1, env, gtin # noqa: F401
|
|
from central.tests.test_rtc_security import _become, _post
|
|
|
|
|
|
def _kb_conflict(env):
|
|
seed_history(env, [staging("RINCAO", G1, "19059090", years={2024: 900, 2025: 900}, desc="COCA LATA 350 ML"), staging("UNIAO", G1, "22021000", years={2024: 20, 2025: 13}, desc="REFRIGERANTE COCA COLA LATA 350 ML")])
|
|
|
|
|
|
def test_attestation_resolves_a_conflict_with_audit_and_only_among_candidates(env):
|
|
_kb_conflict(env)
|
|
cur = history.official_sets(env)
|
|
p = env.scalar(select(KbProduct).where(KbProduct.gtin == G))
|
|
assert kb.evaluate(env, p, "ncm", lambda v: v in cur)["status"] == kb.CONFLICT
|
|
for bad, why in (("21069090", "NCM_IS_NOT_A_CANDIDATE_OF_THIS_GTIN"), ("22030000x", "NCM_IS_NOT_A_CANDIDATE_OF_THIS_GTIN")):
|
|
try:
|
|
kb.attest(env, G, bad, "Rafael", lambda v: v in cur)
|
|
assert False
|
|
except ValueError as error:
|
|
assert str(error) == why
|
|
res = kb.attest(env, G, "22021000", "Rafael", lambda v: v in cur, note="lata 350ml = refrigerante")
|
|
assert res["status"] == kb.VERIFIED_KB and res["value"] == "22021000" and set(res["families"]) == {"OPERATOR_ATTESTATION", "CLIENT_DATA"}
|
|
audit = env.scalars(select(KbPromotionAudit)).one()
|
|
assert audit.sources["operator"] == "Rafael" and audit.new_status == "VERIFIED_KB"
|
|
rejected = env.scalar(select(KbKnowledge).where(KbKnowledge.knowledge_key == "ncm", KbKnowledge.value == "19059090"))
|
|
assert rejected.status == kb.STALE and rejected.evidence["info"]["historical_state"] == "REJECTED_BY_OPERATOR"
|
|
|
|
|
|
def test_name_only_match_is_deterministic_and_conservative(env):
|
|
_kb_conflict(env)
|
|
cur = history.official_sets(env)
|
|
kb.attest(env, G, "22021000", "Rafael", lambda v: v in cur)
|
|
zero = gtin("789490070001")
|
|
seed_history(env, [staging("UNIAO", zero, "22021000", years={2024: 20, 2025: 13}, desc="REFRIGERANTE COCA COLA ZERO LATA 350 ML", code="9")])
|
|
env.commit()
|
|
r = resolve(env, ProviderGateway(env, []), gtin_raw="", description="COCA-COLA ORIGINAL LATA 350 ML", current={"CLASSIFICACAO_FISCAL": "00000000"})
|
|
assert r["state"] == "EXACT_KB_NAME_VERIFIED" and r["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "AUTO_FILL_SAFE" and r["fields"]["CLASSIFICACAO_FISCAL"]["value"] == "22021000"
|
|
ambiguous = resolve(env, ProviderGateway(env, []), gtin_raw="", description="COCA COLA ZERO LATA 350 ML", current={})
|
|
assert ambiguous["fields"]["CLASSIFICACAO_FISCAL"]["status"] != "AUTO_FILL_SAFE" # ZERO is not a neutral extra word: no cross-match
|
|
generic = resolve(env, ProviderGateway(env, []), gtin_raw="", description="REFRIGERANTE", current={})
|
|
assert generic["state"] == "AMBIGUOUS"
|
|
assert not kb.alias_match(frozenset({"COCA", "COLA", "LATA", "350", "ML"}), frozenset({"COCA", "COLA", "ZERO", "LATA", "350", "ML"}))
|
|
assert kb.alias_match(frozenset({"COCA", "COLA", "LATA", "350", "ML"}), frozenset({"REFRIGERANTE", "COCA", "COLA", "LATA", "350", "ML"}))
|
|
|
|
|
|
def test_every_tracked_product_has_a_visible_status(monkeypatch):
|
|
client, engine = setup(monkeypatch)
|
|
out = api_resolve(client, "N1", "", "COCA-COLA ORIGINAL LATA 350 ML", {"CLASSIFICACAO_FISCAL": "00000000"})
|
|
with Session(engine) as s:
|
|
row = s.scalars(select(ProductEnrichment)).one()
|
|
assert row.ui_status == "EAN_AUSENTE" and row.reason_code == "NAME_ONLY_NO_VERIFIED_MATCH" and row.last_attempt_at is not None
|
|
assert out["fields"]["CLASSIFICACAO_FISCAL"]["technical_placeholder"] == "00000000"
|
|
generic = api_resolve(client, "N2", "", "REFRIGERANTE", {})
|
|
invalid = api_resolve(client, "N3", "7891000100104", "PRODUTO X", {})
|
|
with Session(engine) as s:
|
|
st = {r.product_code: r.ui_status for r in s.scalars(select(ProductEnrichment))}
|
|
assert st["N2"] == "NOME_AMBIGUO" and st["N3"] == "EAN_INVALIDO"
|
|
assert all(status.LABELS.get(v) for v in st.values())
|
|
from central.tests.test_product_enrichment_resolver import G1 as g
|
|
done = api_resolve(client, "N4", g, "REFRIGERANTE COLA LATA 350ML", {})
|
|
with Session(engine) as s:
|
|
assert s.scalars(select(ProductEnrichment).where(ProductEnrichment.product_code == "N4")).one().ui_status == "PRONTO_PARA_PREENCHER"
|
|
client.post("/v1/products/enrichment/applied", json={"product_code": "N4", "fields": [{"field": "CLASSIFICACAO_FISCAL", "outcome": "APPLIED", "old": "00000000", "new": "22021000"}]}, headers=H)
|
|
client.post("/v1/products/enrichment/applied", json={"product_code": "N1", "fields": [{"field": "CLASSIFICACAO_FISCAL", "outcome": "ABORTED", "reason": "STALE_CONFLICT"}]}, headers=H)
|
|
with Session(engine) as s:
|
|
st = {r.product_code: (r.ui_status, r.reason_code) for r in s.scalars(select(ProductEnrichment))}
|
|
assert st["N4"][0] == "ENRIQUECIDO" and st["N1"] == ("WRITE_FAILED", "STALE_CONFLICT")
|
|
|
|
|
|
def test_reprocess_sync_and_panel_flag_only_unresolved_products_for_reevaluation(monkeypatch):
|
|
client, engine = setup(monkeypatch)
|
|
api_resolve(client, "U1", "", "COCA-COLA ORIGINAL LATA 350 ML", {"CLASSIFICACAO_FISCAL": "00000000"})
|
|
api_resolve(client, "E1", G1[:0] or gtin("789100010010"), "REFRIGERANTE COLA LATA 350ML", {})
|
|
client.post("/v1/products/enrichment/applied", json={"product_code": "E1", "fields": [{"field": "CLASSIFICACAO_FISCAL", "outcome": "APPLIED", "old": None, "new": "22021000"}]}, headers=H)
|
|
assert client.get("/v1/products/enrichment/pending", headers=H).json() == {"codes": []}
|
|
with Session(engine) as s:
|
|
agent = s.scalars(select(Agent)).first()
|
|
for kind in ("REPROCESS_ALL", "SYNC_INCREMENTAL"):
|
|
create_job(s, agent, kind, {})
|
|
assert s.scalars(select(ProductEnrichment).where(ProductEnrichment.reevaluate.is_(True))).all()[0].product_code == "U1"
|
|
for r in s.scalars(select(ProductEnrichment)):
|
|
r.reevaluate = False
|
|
s.commit()
|
|
for j in s.scalars(select(__import__("central.app.models", fromlist=["Job"]).Job)):
|
|
j.status = "COMPLETED"
|
|
s.commit()
|
|
with Session(engine) as s:
|
|
create_job(s, s.scalars(select(Agent)).first(), "REPROCESS_PRODUCTS", {"codigos": ["U1"]})
|
|
assert client.get("/v1/products/enrichment/pending", headers=H).json() == {"codes": ["U1"]} # the enriched product E1 is never re-evaluated
|
|
api_resolve(client, "U1", "", "COCA-COLA ORIGINAL LATA 350 ML", {"CLASSIFICACAO_FISCAL": "00000000"})
|
|
assert client.get("/v1/products/enrichment/pending", headers=H).json() == {"codes": []} # consumed by the new evaluation
|
|
|
|
|
|
def test_panel_reevaluate_and_attest_routes_are_csrf_admin_and_audited(monkeypatch):
|
|
client, agent_id, _ = _panel_client()
|
|
with client._session_factory() as s:
|
|
seed_history(s, [staging("RINCAO", G1, "19059090", years={2024: 900, 2025: 900}, desc="COCA LATA 350 ML"), staging("UNIAO", G1, "22021000", years={2024: 20, 2025: 13}, desc="REFRIGERANTE COCA COLA LATA 350 ML")])
|
|
from central.tests.test_product_enrichment_resolver import seed
|
|
seed(s)
|
|
i = s.scalars(select(Installation)).one()
|
|
i.enrichment_mode, i.features_configured = "ENABLED", True
|
|
s.commit()
|
|
_become(client, "PADRAO")
|
|
assert _post(client, "/kb/attest", gtin="0" + G1, ncm="22021000").status_code == 403
|
|
_become(client, "ADMIN")
|
|
assert client.post("/kb/attest", data={"csrf_token": "bad", "gtin": "0" + G1, "ncm": "22021000"}, follow_redirects=False).status_code in (400, 403)
|
|
assert _post(client, "/kb/attest", gtin="0" + G1, ncm="21069090").status_code == 422
|
|
assert _post(client, "/kb/attest", gtin="0" + G1, ncm="22021000").status_code == 303
|
|
assert _post(client, f"/agents/{agent_id}/enrichment/reevaluate", code="").status_code == 303
|
|
with client._session_factory() as s:
|
|
acts = [e.action for e in s.scalars(select(AuditEvent))]
|
|
assert "KB_NCM_ATTESTED" in acts and "KB_ATTEST_DENIED" in acts and "ENRICHMENT_REEVALUATE" in acts
|
|
page = client.get("/?kb_q=COCA COLA LATA 350 ML").text
|
|
assert "VERIFIED_KB" in page and "decidido por" in page
|
|
|
|
|
|
def test_rtc_mode_and_enrichment_mode_are_independent_switches_with_distinct_labels():
|
|
client, agent_id, _ = _panel_client()
|
|
_become(client, "ADMIN")
|
|
assert _post(client, f"/agents/{agent_id}/features", enrichment_mode="DRY_RUN", corrector_enabled="1").status_code == 303
|
|
with client._session_factory() as s:
|
|
i = s.scalars(select(Installation)).one()
|
|
assert i.enrichment_mode == "DRY_RUN" and i.rtc_write_mode == "DISABLED"
|
|
assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="DRY_RUN").status_code == 303
|
|
assert _post(client, f"/agents/{agent_id}/rtc-mode", mode="ENABLED", confirm="PREENCHER").status_code == 303
|
|
with client._session_factory() as s:
|
|
i = s.scalars(select(Installation)).one()
|
|
assert i.rtc_write_mode == "ENABLED" and i.enrichment_mode == "DRY_RUN" # turning cClassTrib to Automatic never touches the enrichment
|
|
page = client.get(f"/agents/{agent_id}").text
|
|
assert "Preenchimento cClassTrib (RTC)" in page and "Enriquecimento de produtos novos" in page
|