GTI_GERENCIADOR_FISCAL/fiscal/central/tests/test_oscbr_backfill.py

"""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={})
    prov = Oscbr(httpx.Client(transport=httpx.MockTransport(handler)), None, "u", "p")
    prov.provides_ncm = True                       # simulates a plan/answer that DOES carry NCM: exercises the backfill/promotion logic (the real V3 role is IDENTITY)
    return prov


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


def test_oscbr_real_role_is_identity_and_uses_the_v3_endpoint():
    seen = []

    def handler(request):
        seen.append(request.url.path)
        return httpx.Response(200, json={"ean": "7891000100103", "ean_tipo": "EAN13", "nome": "COLA 350ML", "marca": "X", "categoria": "Bebidas", "link_foto": "u"})
    p = Oscbr(httpx.Client(transport=httpx.MockTransport(handler)), "tok")
    r = p.lookup_gtin("7891000100103")
    assert seen == ["/api/v3/gtin/7891000100103"] and r.found and r.title == "COLA 350ML" and r.ncm is None and r.cest is None
    assert Oscbr.provides_ncm is False
    legacy = Oscbr(httpx.Client(transport=httpx.MockTransport(lambda q: httpx.Response(200, json={"nome": "FEIJAO", "ncm": 7133399}))), "tok").lookup_gtin("7896116900029")
    assert legacy.ncm == "07133399"                                              # 7-digit integer restored to the 8-digit NCM