"""Conservative LIVE smoke of the free providers (separate from the offline test suite): 1 lookup per provider on a famous, stable GTIN. python scripts/enrichment_live_smoke.py Goes through the real ProviderGateway (cache + quota + circuit breaker) on an in-memory database; tokens are never printed. Exit 0 when every CONFIGURED provider answered without an infrastructure error (NOT_FOUND is a valid answer; a rate limit is reported, not treated as a failure of our code). """ import json import os import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from sqlalchemy import create_engine # noqa: E402 from sqlalchemy.orm import Session # noqa: E402 from sqlalchemy.pool import StaticPool # noqa: E402 from central.app.enrichment.providers import build_providers # noqa: E402 from central.app.enrichment.quota import ProviderGateway # noqa: E402 from central.app.models import Base # noqa: E402 GTIN = "7894900011517" # Coca-Cola 350 ml (BR): stable, widely catalogued def main(): engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(engine) settings = type("S", (), {"cosmos_token": os.environ.get("COSMOS_TOKEN", ""), "allow_paid_providers": False})() out, ok = [], True with Session(engine) as session: gateway = ProviderGateway(session, build_providers(settings)) for provider in gateway.providers: result = gateway.lookup(provider, GTIN) row = {"provider": provider.name, "configured": provider.configured, "state": result.state, "reason": result.reason, "title": result.product.title if result.product else None} out.append(row) if provider.configured and result.state == "ERROR" and result.reason not in ("RATE_LIMITED",): ok = False print(json.dumps({"gtin": GTIN, "results": out, "OK": ok}, ensure_ascii=False, indent=1)) sys.exit(0 if ok else 1) if __name__ == "__main__": main()