111 lines
5.2 KiB
Python
111 lines
5.2 KiB
Python
"""OSCBR backfill: prioritised selection, promotion only on agreement, external conflict, quota (5/min, 50/day, reserve, reset), Cosmos out of the chain."""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import httpx
|
|
from sqlalchemy import select
|
|
|
|
from central.app.enrichment import backfill, history, kb, quota
|
|
from central.app.enrichment.providers import Oscbr, build_providers
|
|
from central.app.enrichment.quota import ProviderGateway, QuotaManager
|
|
from central.app.models import KbProduct, KbPromotionAudit, ProviderUsageDaily
|
|
from central.tests.test_enrichment_history import run, staging
|
|
from central.tests.test_product_enrichment_resolver import env, gtin # noqa: F401
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _fresh_windows():
|
|
quota._windows.clear()
|
|
yield
|
|
quota._windows.clear()
|
|
|
|
|
|
def _oscbr(ncm_by_gtin, calls):
|
|
def handler(request):
|
|
if request.url.path == "/oauth/token":
|
|
return httpx.Response(200, json={"token": "t"})
|
|
g = request.url.path.rsplit("/", 1)[1]
|
|
calls.append(g)
|
|
if g in ncm_by_gtin:
|
|
return httpx.Response(200, json={"ean": g, "nome": "PRODUTO X", "marca": "M", "ncm": ncm_by_gtin[g], "cest": "0300700"})
|
|
return httpx.Response(404, json={})
|
|
return Oscbr(httpx.Client(transport=httpx.MockTransport(handler)), None, "u", "p")
|
|
|
|
|
|
def _seed(env, n=6):
|
|
gs = [gtin(f"78960000{i:04d}") for i in range(n)]
|
|
for i, g in enumerate(gs):
|
|
run(env, [staging("A", g, "22021000", years={2023: 30 + i, 2024: 40}, code=str(i), desc=f"PRODUTO TESTE {i} 350ML")])
|
|
return gs
|
|
|
|
|
|
def test_backfill_promotes_only_on_agreement_and_keeps_corroborated_when_no_ncm(env):
|
|
gs = _seed(env, 4)
|
|
prov = _oscbr({gs[0]: "22021000", gs[1]: "19059090"}, [])
|
|
cur = history.official_sets(env)
|
|
rep = backfill.run(env, ProviderGateway(env, [prov]), prov, limit=10, reserve_percent=20, is_official=lambda v: v in cur, wait_rate_window=False)
|
|
by = {r["gtin"]: r for r in rep["records"]}
|
|
assert by[gs[0]]["kb_status"] == "VERIFIED_KB" and by[gs[1]]["kb_status"] == "CONFLICT" and by[gs[1]]["external_conflict"]
|
|
assert by[gs[2]]["kb_status"] == "CORROBORATED" and by[gs[2]]["found"] is False
|
|
assert env.scalars(select(KbPromotionAudit)).all()[0].sources["providers"] == ["PROVIDER:oscbr"]
|
|
cest = [r for r in by.values() if r["cest"]]
|
|
assert cest and cest[0]["cest"] == "0300700" # CEST kept as evidence only
|
|
|
|
|
|
def test_backfill_never_selects_verified_or_conflict_or_rechecked(env):
|
|
gs = _seed(env, 3)
|
|
p = env.scalar(select(KbProduct).where(KbProduct.gtin == "0" + gs[0]))
|
|
kb.add_evidence(env, p, "ncm", "19059090", "x", family=kb.CLIENT_FAMILY, client_label="Z", lineage={"docs": 50, "years": {"2024": 25, "2025": 25}})
|
|
env.commit()
|
|
cur = history.official_sets(env)
|
|
sample = backfill.select_sample(env, lambda v: v in cur, 10, "oscbr")
|
|
assert "0" + gs[0] not in {s["gtin_key"] for s in sample} and len(sample) == 2
|
|
prov = _oscbr({}, [])
|
|
backfill.run(env, ProviderGateway(env, [prov]), prov, limit=10, reserve_percent=20, is_official=lambda v: v in cur, wait_rate_window=False)
|
|
assert backfill.select_sample(env, lambda v: v in cur, 10, "oscbr") == [] # checked once, next_checked_at in the future
|
|
|
|
|
|
def test_oscbr_quota_5_per_minute_50_per_day_reserve_and_reset(env):
|
|
prov = _oscbr({}, [])
|
|
gw = ProviderGateway(env, [prov])
|
|
quota._windows["oscbr"].clear()
|
|
outs = [gw.lookup(prov, gtin(f"78970000{i:04d}")).state for i in range(7)]
|
|
assert outs[:5] == ["NOT_FOUND"] * 5 and outs[5:] == ["SKIPPED", "SKIPPED"]
|
|
assert gw.lookup(prov, gtin("789700009991")).reason == "RATE_WINDOW"
|
|
usage = env.scalar(select(ProviderUsageDaily).where(ProviderUsageDaily.provider == "oscbr"))
|
|
assert usage.requests == 5
|
|
quota._windows["oscbr"].clear()
|
|
usage.requests = 40
|
|
assert gw.lookup(prov, gtin("789700009992"), "BACKFILL", 20).reason == "OPERATIONAL_RESERVE"
|
|
assert gw.lookup(prov, gtin("789700009993"), "OPERATIONAL").state == "NOT_FOUND"
|
|
quota._windows["oscbr"].clear()
|
|
usage.requests = 50
|
|
assert gw.lookup(prov, gtin("789700009994"), "OPERATIONAL").reason == "RATE_LIMITED"
|
|
quota._windows["oscbr"].clear()
|
|
tomorrow = QuotaManager(env, now=lambda: datetime.now(timezone.utc) + timedelta(days=1, minutes=1))
|
|
ok, reason, _ = tomorrow.allow(prov)
|
|
assert ok and reason is None
|
|
|
|
|
|
def test_cosmos_out_of_chain_and_token_preferred():
|
|
class S:
|
|
cosmos_token = "SECRET-C"
|
|
cosmos_enabled = False
|
|
oscbr_token = "SECRET-T"
|
|
oscbr_user = ""
|
|
oscbr_password = ""
|
|
kodebar_api_key = ""
|
|
allow_paid_providers = False
|
|
names = [p.name for p in build_providers(S())]
|
|
assert "cosmos" not in names and names[0] == "oscbr" and build_providers(S())[0].configured
|
|
S.cosmos_enabled = True
|
|
assert build_providers(S())[0].name == "cosmos"
|
|
|
|
|
|
def test_provider_query_canonical_forms():
|
|
assert backfill.provider_query("00000078936683") == "78936683" # GTIN-8
|
|
assert backfill.provider_query("07891000100103") == "7891000100103" # EAN-13
|
|
assert backfill.provider_query("17891000100100") == "17891000100100" # GTIN-14 with indicator
|