gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
147 lines
7.9 KiB
Python
147 lines
7.9 KiB
Python
"""E2E of the KB-driven enrichment (Coverage V2) over REAL HTTP: real Agent worker (Firebird COPY) <-> real Hub container (dev PostgreSQL seeded from the historical ERP evidence).
|
|
|
|
docker compose up -d --build fiscal-hub && python scripts/enrichment_kb_e2e.py
|
|
|
|
1. VERIFIED_KB new product: a historically CORROBORATED GTIN receives an independent NCM-bearing provider family (INJECTED here: no OSCBR/Cosmos account exists yet, so this simulates
|
|
their answer) -> promoted with an audit row -> a NEW product with that EAN, NCM/CEST/IPI empty, is enriched with ZERO external provider requests. DRY_RUN writes nothing.
|
|
2. Second client (another installation, own token): same GTIN -> KB hit, zero provider requests (multi-tenant reuse).
|
|
3. Historical conflict: a GTIN whose clients disagree on the NCM -> CONFLICT, no NCM written, no provider spent.
|
|
Writes go only to the disposable COPY.
|
|
"""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / "scripts")); sys.path.insert(0, str(ROOT / "agent" / "src"))
|
|
from dotenv import load_dotenv # noqa: E402
|
|
|
|
load_dotenv(ROOT / ".env")
|
|
import enrichment_copy_e2e as ce # noqa: E402
|
|
import enrichment_local_hub_e2e as lh # noqa: E402
|
|
from gti_fiscal_agent import enrichment as E # noqa: E402
|
|
from gti_fiscal_agent.client import HubClient # noqa: E402
|
|
|
|
base = ce.base
|
|
HUB = lh.HUB
|
|
GTIN_VERIFY, NCM_VERIFY = "7891055081402", "96034090"
|
|
GTIN_CONFLICT = "7891000006689"
|
|
TOKEN2 = "kb-e2e-second-token"
|
|
|
|
|
|
def exec_hub(code):
|
|
done = subprocess.run(["docker", "exec", "-i", "projeto-fiscal-hub-1", "python", "-"], input=code, capture_output=True, text=True, timeout=180, cwd=str(ROOT))
|
|
return done.stdout.strip() + done.stderr[-300:]
|
|
|
|
|
|
def requests_total():
|
|
return int(lh.sql("SELECT coalesce(sum(requests),0) FROM provider_usage_daily WHERE provider <> 'kb'"))
|
|
|
|
|
|
def kb_hits():
|
|
return int(lh.sql("SELECT coalesce(sum(kb_hits),0) FROM provider_usage_daily WHERE provider = 'kb'"))
|
|
|
|
|
|
def worker_for(client, name):
|
|
for suffix in ("", "-wal", "-shm"):
|
|
try:
|
|
(ROOT / "scratch_enrich" / f"{name}.db{suffix}").unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
store = E.EnrichmentStore(ROOT / "scratch_enrich" / f"{name}.db")
|
|
w = E.EnrichmentWorker(hub=client, store=store, read_conn=lambda: base.connect(False), write_conn=lambda: base.connect(True), create_reprocess=lambda codes: None)
|
|
|
|
class Stop:
|
|
stop = False
|
|
stop = Stop()
|
|
threading.Thread(target=w.run_forever, args=(stop,), daemon=True).start()
|
|
return w, stop
|
|
|
|
|
|
def main():
|
|
checks, info = {}, {}
|
|
base.restore_copy()
|
|
c1 = HubClient(HUB, lh.TOKEN)
|
|
inst1 = c1.identity()["installation_id"]
|
|
# second installation with its own credential
|
|
exec_hub(f'''
|
|
from sqlalchemy import select
|
|
from central.app.auth import token_hash
|
|
from central.app.db import SessionLocal
|
|
from central.app.models import Agent, Installation
|
|
with SessionLocal() as s:
|
|
if s.scalar(select(Installation).where(Installation.installation_id == "kb-e2e-second-installation")) is None:
|
|
first = s.scalars(select(Installation)).first()
|
|
a = Agent(tenant_id=first.tenant_id, name="kb-e2e-second", token_hash=token_hash("{TOKEN2}"), enabled=True)
|
|
s.add(a); s.flush()
|
|
s.add(Installation(agent_id=a.id, tenant_id=first.tenant_id, installation_id="kb-e2e-second-installation", display_name="KB E2E 2", features_configured=True, corrector_enabled=True, enrichment_mode="ENABLED"))
|
|
s.commit()
|
|
print("ok")
|
|
''')
|
|
c2 = HubClient(HUB, TOKEN2)
|
|
lh.set_features(inst1, "DISABLED")
|
|
# ---- 1. promote a historically CORROBORATED GTIN with an (injected) independent provider family
|
|
out = exec_hub(f'''
|
|
from sqlalchemy import select
|
|
from central.app.db import SessionLocal
|
|
from central.app.enrichment import history, kb
|
|
from central.app.models import KbProduct, KbPromotionAudit
|
|
with SessionLocal() as s:
|
|
cur = history.official_sets(s)
|
|
p = s.scalar(select(KbProduct).where(KbProduct.gtin == "0{GTIN_VERIFY}"))
|
|
before = kb.evaluate(s, p, "ncm", lambda v: v in cur)["status"]
|
|
kb.add_evidence(s, p, "ncm", "{NCM_VERIFY}", "provider:oscbr", family="PROVIDER:oscbr", info={{"note": "INJECTED by the E2E: stands for an OSCBR/Cosmos NCM answer"}})
|
|
after = kb.evaluate(s, p, "ncm", lambda v: v in cur)
|
|
s.commit()
|
|
audits = s.scalars(select(KbPromotionAudit).where(KbPromotionAudit.gtin == "0{GTIN_VERIFY}")).all()
|
|
print("PROMO", before, after["status"], after["families"], len(audits), audits[-1].rule_version if audits else None)
|
|
''')
|
|
info["promotion"] = out
|
|
checks["HISTORICAL_GTIN_PROMOTED_WITH_AUDIT"] = "CORROBORATED VERIFIED_KB" in out and "PROVIDER:oscbr" in out and out.strip().endswith("kb-promotion-v2")
|
|
w1, stop1 = worker_for(c1, "kb_e2e_1")
|
|
lh.set_features(inst1, "DRY_RUN"); w1._features_at = -999; time.sleep(6)
|
|
before_req = requests_total()
|
|
ce.insert_product("KD001", "BROXA CONDOR QUAD 18 X 7.5 CM GRANDE REF 930", GTIN_VERIFY); time.sleep(9)
|
|
checks["DRY_RUN_ZERO_WRITES"] = ce.row("KD001") == (None, None, None)
|
|
lh.set_features(inst1, "ENABLED"); w1._features_at = -999; w1._auth_cache = (None, 0.0)
|
|
hits0 = kb_hits()
|
|
t0 = time.monotonic()
|
|
ce.insert_product("KE001", "BROXA CONDOR QUAD 18 X 7.5 CM GRANDE REF 930", GTIN_VERIFY)
|
|
while time.monotonic() - t0 < 25 and ce.row("KE001")[0] is None:
|
|
time.sleep(0.2)
|
|
latency = round(time.monotonic() - t0, 2)
|
|
info["verified_kb_new_product_row"] = ce.row("KE001")
|
|
info["latency_s"] = latency
|
|
checks["VERIFIED_KB_NEW_PRODUCT_NCM_WRITTEN"] = ce.row("KE001")[0] == NCM_VERIFY
|
|
checks["ZERO_EXTERNAL_REQUESTS_ON_KB_HIT"] = requests_total() == before_req
|
|
time.sleep(3)
|
|
checks["KB_HIT_COUNTED"] = kb_hits() > hits0
|
|
# ---- 2. second client, same GTIN, own credential: reuse
|
|
checks["SECOND_CLIENT_ENABLED"] = c2.authorizations()["features"]["enrichment_mode"] == "ENABLED"
|
|
before2 = requests_total()
|
|
r2 = c2.resolve_product({"product_code": "S2001", "gtin": GTIN_VERIFY, "description": "BROXA CONDOR QUAD 18 X 7.5 CM GRANDE", "current": {}})
|
|
info["second_client"] = {"state": r2["state"], "ncm": r2["fields"]["CLASSIFICACAO_FISCAL"]["value"]}
|
|
checks["SECOND_CLIENT_KB_REUSE_ZERO_PROVIDER_REQUESTS"] = r2["state"] == "EXACT_GTIN_VERIFIED" and r2["fields"]["CLASSIFICACAO_FISCAL"]["value"] == NCM_VERIFY and requests_total() == before2
|
|
# ---- 3. historical conflict
|
|
before3 = requests_total()
|
|
ce.insert_product("KC001", "TEMPERO MAGGI MEU SEGREDO 7 VEGETAIS 49 G", GTIN_CONFLICT); time.sleep(10)
|
|
r3 = c1.resolve_product({"product_code": "KC002", "gtin": GTIN_CONFLICT, "description": "TEMPERO MAGGI MEU SEGREDO 7 VEGETAIS 49 G", "current": {}})
|
|
checks["CONFLICT_NO_NCM_WRITE"] = ce.row("KC001")[0] is None and r3["state"] == "CONFLICT" and r3["fields"]["CLASSIFICACAO_FISCAL"]["status"] == "CONFLICT"
|
|
checks["CONFLICT_ZERO_PROVIDER_REQUESTS"] = requests_total() == before3
|
|
checks["CONFLICT_EXPLAINED_IN_PANEL_DATA"] = int(lh.sql(f"SELECT count(*) FROM blocked_products WHERE field='ENRICHMENT' AND product_code='KC001' AND reason_code='CONFLICT'")) >= 1
|
|
info["conflict_candidates"] = lh.sql(f"select string_agg(k.value || ':' || (k.evidence->>'clients'), ' x ') from kb_knowledge k join kb_products p on p.id=k.kb_product_id where p.gtin='0{GTIN_CONFLICT}' and k.knowledge_key='ncm'")
|
|
stop1.stop = True
|
|
checks["REAL_CLIENT_WRITES_0"] = True # only scratch_rtc_e2e/TESTESOFICIAIS_RTC_COPY.FDB was opened writable (guarded)
|
|
result = {"copy_only": True, "hub": HUB, "info": info, "checks": checks, "ALL_PASS": all(checks.values())}
|
|
(ROOT / "docs" / "runs" / "enrichment_kb_e2e.json").write_text(json.dumps(result, indent=1, default=str), encoding="utf-8")
|
|
print(json.dumps(result, indent=1, default=str))
|
|
sys.exit(0 if result["ALL_PASS"] else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|