From 751efe8c113251dc0230f26f5ef79a4eac11b058 Mon Sep 17 00:00:00 2001 From: Sirajmx Date: Mon, 21 Sep 2026 19:37:25 +0530 Subject: [PATCH 1/9] fix:list_products builds from inventory only in CSV mode; GAM/FreeWheel/others get static demo catalog --- CHANGELOG.md | 13 + src/ad_seller/flows/product_setup_flow.py | 22 +- src/ad_seller/interfaces/api/deps.py | 10 +- src/ad_seller/interfaces/api/main.py | 36 +- src/ad_seller/interfaces/api/routers/deals.py | 4 +- .../interfaces/api/routers/media_kit.py | 2 +- .../interfaces/api/routers/negotiation.py | 2 +- .../interfaces/api/routers/products.py | 10 +- .../interfaces/api/routers/quotes.py | 2 +- .../interfaces/api/routers/registry.py | 2 +- src/ad_seller/interfaces/chat/main.py | 2 +- src/ad_seller/interfaces/cli/main.py | 11 +- src/ad_seller/interfaces/mcp_server.py | 19 +- src/ad_seller/services/catalog_service.py | 123 +++++- src/ad_seller/services/negotiation_service.py | 2 +- src/ad_seller/tools/catalog/avails_tool.py | 3 +- tests/unit/test_csv_catalog_coherence.py | 36 +- tests/unit/test_issue34_catalog_fixes.py | 8 +- tests/unit/test_service_layer.py | 12 +- tests/unit/test_synced_product_catalog.py | 357 ++++++++++++++++++ 20 files changed, 595 insertions(+), 81 deletions(-) create mode 100644 tests/unit/test_synced_product_catalog.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0517daed..54ce23ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,19 @@ All notable changes to the IAB Tech Lab Seller Agent are documented here. ### Fixed +- `GET /products` (and everything reading the catalog — avails, pricing, + package resolution, quotes) now reflects real ad-server inventory + (GAM/FreeWheel/S3) once synced, read fresh from storage on every call + (AI-6). Previously a synced product only ever lived in the flow's + per-request state and one worker process's in-memory cache + (`infra/docker/Dockerfile` runs uvicorn with `--workers 2`), so the + catalog kept serving the 13-product static default indefinitely after + a real sync — matching how synced packages already worked. Synced + products persist under a dedicated `synced_product:{id}` storage key, + separate from the generic `product:{id}` key `negotiation_service` + already writes for an unrelated reason (a narrower proposal-time + snapshot), and are pruned when no longer returned by the ad server. + CSV-mode and no-ad-server-configured behavior is unchanged. - Map internal deal status to the shared wire enum on read; deals created via from-template, bulk, or curated paths no longer 500 on GET (#73). Internal `confirmed` reads as `booked`, internal diff --git a/src/ad_seller/flows/product_setup_flow.py b/src/ad_seller/flows/product_setup_flow.py index c7e1cc8d..9e262e15 100644 --- a/src/ad_seller/flows/product_setup_flow.py +++ b/src/ad_seller/flows/product_setup_flow.py @@ -160,11 +160,18 @@ async def sync_from_ad_server(self) -> None: # Canonical item→product mapping lives in catalog_service # (shared with the CSV-mode catalog build) so flow-seeded # products and API catalog products cannot diverge. - from ..services.catalog_service import product_from_inventory_item + from ..services.catalog_service import ( + persist_synced_product, + product_from_inventory_item, + ) for item in items: product_def = product_from_inventory_item(item) self.state.products[product_def.product_id] = product_def + # Persist so GET /products reflects real ad-server inventory + # for non-CSV ad servers too (AI-6) -- state.products alone + # is per-flow-instance and never outlives this request. + await persist_synced_product(product_def) logger.info("Created %d products from ad server inventory", len(self.state.products)) @@ -209,11 +216,22 @@ async def sync_from_ad_server(self) -> None: await self._finish_sync() async def _finish_sync(self) -> None: - """Terminal step of every sync path: prune the stale synced layer.""" + """Terminal step of every sync path: prune the stale synced layer. + + Runs unconditionally, including on the mock/fallback path (no ad + server configured, or the real sync raised) -- mirroring + ``_prune_stale_synced_packages``'s own convergence guarantee + (issue #34): a transient ad-server failure reverts the catalog to + the honest static-default fallback rather than serving + increasingly stale synced products forever, and the next + successful sync immediately re-populates it. + """ + from ..services.catalog_service import prune_stale_synced_products from ..storage.factory import get_storage storage = await get_storage() await self._prune_stale_synced_packages(storage) + await prune_stale_synced_products(set(self.state.products.keys())) async def _create_mock_synced_packages(self) -> None: """Create mock Layer 1 packages for development without ad server creds.""" diff --git a/src/ad_seller/interfaces/api/deps.py b/src/ad_seller/interfaces/api/deps.py index 834231f9..5cc9b6c5 100644 --- a/src/ad_seller/interfaces/api/deps.py +++ b/src/ad_seller/interfaces/api/deps.py @@ -237,17 +237,21 @@ async def _verified_buyer_context( ) -def get_product_catalog() -> dict[str, Any]: - """Return the cached static product catalog. +async def get_product_catalog() -> dict[str, Any]: + """Return the product catalog. Resolves through ``interfaces.api.main._get_static_product_catalog`` AT CALL TIME so existing tests that patch that attribute (and reset ``main._STATIC_PRODUCT_CATALOG``) keep governing every endpoint. The lazy import avoids the main ↔ routers circular import. + + Async since AI-6: when a real ad server is configured and synced, + this reads persisted products from storage on every call rather than + an in-memory cache (see ``catalog_service.get_static_product_catalog``). """ from . import main - return main._get_static_product_catalog() + return await main._get_static_product_catalog() async def _get_media_kit_service(): diff --git a/src/ad_seller/interfaces/api/main.py b/src/ad_seller/interfaces/api/main.py index 6f9e6221..542bb4d7 100644 --- a/src/ad_seller/interfaces/api/main.py +++ b/src/ad_seller/interfaces/api/main.py @@ -167,29 +167,39 @@ async def lifespan(application): # backward compatibility with external importers. from .schemas import * # noqa: E402,F401,F403 -# Cached static product catalog compat surface. The single catalog source -# lives in services.catalog_service, but tests patch +# Patchable delegator surface. The single catalog source lives in +# services.catalog_service, but tests patch # `ad_seller.interfaces.api.main._get_static_product_catalog` and reset # `_STATIC_PRODUCT_CATALOG` between tests — so this module keeps the -# patchable delegator, and endpoint code resolves the catalog through it -# at call time (see deps.get_product_catalog). +# patchable name, and endpoint code resolves the catalog through it at +# call time (see deps.get_product_catalog). +# +# AI-6: this no longer CACHES the catalog itself — a synced ad server's +# catalog must be read live from storage on every call (see +# catalog_service.get_static_product_catalog), and caching the dict here +# would silently pin it to whatever the first call happened to return. +# `_STATIC_PRODUCT_CATALOG` is kept only as the legacy test-reset signal: +# setting it to None forces one `catalog_service.reset_catalog_cache()` +# call, which existing test fixtures rely on for a deterministic rebuild +# between tests (CSV/static catalogs are still cached inside +# catalog_service itself). _STATIC_PRODUCT_CATALOG: Optional[dict[str, Any]] = None -def _get_static_product_catalog() -> dict[str, Any]: - """Return the seller's default product catalog without running the flow. +async def _get_static_product_catalog() -> dict[str, Any]: + """Return the seller's product catalog without running the flow. - Delegates to ``services.catalog_service`` (the single catalog source). + Delegates to ``services.catalog_service`` (the single catalog source) + on every call. See the module comment above ``_STATIC_PRODUCT_CATALOG`` + for why this function doesn't cache the result itself. """ global _STATIC_PRODUCT_CATALOG - if _STATIC_PRODUCT_CATALOG is None: - from ...services import catalog_service + from ...services import catalog_service - # Honor legacy reset semantics: clearing _STATIC_PRODUCT_CATALOG - # means "give me a fresh catalog", so drop the service cache too. + if _STATIC_PRODUCT_CATALOG is None: catalog_service.reset_catalog_cache() - _STATIC_PRODUCT_CATALOG = catalog_service.get_static_product_catalog() - return _STATIC_PRODUCT_CATALOG + _STATIC_PRODUCT_CATALOG = {"_reset_consumed": True} + return await catalog_service.get_static_product_catalog() def _serialize_product(product: Any) -> dict[str, Any]: diff --git a/src/ad_seller/interfaces/api/routers/deals.py b/src/ad_seller/interfaces/api/routers/deals.py index 1a791edf..12c9db9c 100644 --- a/src/ad_seller/interfaces/api/routers/deals.py +++ b/src/ad_seller/interfaces/api/routers/deals.py @@ -348,7 +348,7 @@ async def create_deal_from_template( ) # Product data from the single cached catalog source (EP-3.3) - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() # Resolve buyer context from API key + body. EP-5.2: the tier is capped # at the registry-verified ceiling when an agent_url is presented (the @@ -518,7 +518,7 @@ async def create_curated_deal(request: CuratedDealRequest): enriched with curator identity, fee, and targeting overlay. """ # Product base price from the single cached catalog source (EP-3.3) - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() return await deal_service.create_curated_deal(request, catalog) diff --git a/src/ad_seller/interfaces/api/routers/media_kit.py b/src/ad_seller/interfaces/api/routers/media_kit.py index 4be33ee2..fcc65c9a 100644 --- a/src/ad_seller/interfaces/api/routers/media_kit.py +++ b/src/ad_seller/interfaces/api/routers/media_kit.py @@ -184,7 +184,7 @@ async def _resolve_package_products(product_ids: list[str], storage): """ from ....models.flow_state import ProductDefinition - catalog_products = deps.get_product_catalog()["products"] + catalog_products = (await deps.get_product_catalog())["products"] resolved = [] unresolved: list[str] = [] for pid in product_ids: diff --git a/src/ad_seller/interfaces/api/routers/negotiation.py b/src/ad_seller/interfaces/api/routers/negotiation.py index 41427ad2..27393e26 100644 --- a/src/ad_seller/interfaces/api/routers/negotiation.py +++ b/src/ad_seller/interfaces/api/routers/negotiation.py @@ -35,7 +35,7 @@ async def submit_proposal( ): """Submit a proposal for review.""" # Product data from the single cached catalog source (EP-3.3) - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() # EP-5.2: verify the claimed tier against the agent registry and cap at # the verified ceiling (blocked agents 403; unverifiable claims floor). diff --git a/src/ad_seller/interfaces/api/routers/products.py b/src/ad_seller/interfaces/api/routers/products.py index 3c152555..033e76dd 100644 --- a/src/ad_seller/interfaces/api/routers/products.py +++ b/src/ad_seller/interfaces/api/routers/products.py @@ -49,7 +49,7 @@ async def list_products( Buyers filter client-side over the returned Product records (there is deliberately no POST /products/search on the shared catalog surface). """ - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() return cm.products_to_list_response( list(catalog["products"].values()), limit=limit, offset=offset ) @@ -133,7 +133,7 @@ async def check_avails( See :func:`ad_seller.services.catalog_service.check_avails` for the full policy. """ - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() if isinstance(request, ProductAvailsSearch): return _spec_avails_collection(request, catalog) @@ -160,7 +160,7 @@ async def get_product(product_id: str) -> Product: Reads from the cached static catalog instead of running ProductSetupFlow per request (see `list_products` for rationale). """ - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() product = catalog["products"].get(product_id) if not product: raise HTTPException(status_code=404, detail="Product not found") @@ -173,7 +173,7 @@ async def get_pricing( api_key_record=Depends(deps._get_optional_api_key_record), ): """Get pricing for a product based on buyer context.""" - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() product = catalog["products"].get(request.product_id) if not product: raise HTTPException(status_code=404, detail="Product not found") @@ -207,7 +207,7 @@ async def discovery_query( from ....flows import DiscoveryInquiryFlow # Product data from the single cached catalog source (EP-3.3) - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() # Enforce agent registry _, max_tier = await deps._resolve_and_enforce_agent(request.agent_url) diff --git a/src/ad_seller/interfaces/api/routers/quotes.py b/src/ad_seller/interfaces/api/routers/quotes.py index d38bd11d..b5b5a1c9 100644 --- a/src/ad_seller/interfaces/api/routers/quotes.py +++ b/src/ad_seller/interfaces/api/routers/quotes.py @@ -59,7 +59,7 @@ async def create_quote( # Read product from cached static catalog rather than running # ProductSetupFlow per request (hangs in OpenDirect MCP # session.initialize() — see catalog_service). - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() # Resolve buyer identity — API key takes priority over body. EP-5.2: # the claimed tier is verified against the agent registry (agent_url) diff --git a/src/ad_seller/interfaces/api/routers/registry.py b/src/ad_seller/interfaces/api/routers/registry.py index e57834f8..0b455769 100644 --- a/src/ad_seller/interfaces/api/routers/registry.py +++ b/src/ad_seller/interfaces/api/routers/registry.py @@ -37,7 +37,7 @@ async def agent_card(): # ProductSetupFlow per request (which hangs in OpenDirect MCP # session.initialize() — see `catalog_service` for context). try: - inventory_types = set(deps.get_product_catalog()["inventory_types"]) + inventory_types = set((await deps.get_product_catalog())["inventory_types"]) except Exception: inventory_types = {"display", "video", "ctv", "native", "mobile_app"} diff --git a/src/ad_seller/interfaces/chat/main.py b/src/ad_seller/interfaces/chat/main.py index 979a00de..a861c811 100644 --- a/src/ad_seller/interfaces/chat/main.py +++ b/src/ad_seller/interfaces/chat/main.py @@ -66,7 +66,7 @@ async def initialize(self) -> None: """ from ...services import catalog_service - catalog = catalog_service.get_static_product_catalog() + catalog = await catalog_service.get_static_product_catalog() self._products = catalog["products"] def set_buyer_context(self, context: BuyerContext) -> None: diff --git a/src/ad_seller/interfaces/cli/main.py b/src/ad_seller/interfaces/cli/main.py index 1c22381b..29a095b0 100644 --- a/src/ad_seller/interfaces/cli/main.py +++ b/src/ad_seller/interfaces/cli/main.py @@ -47,12 +47,15 @@ def init( ): """Initialize the seller system and set up default products.""" from ...services import catalog_service + from ...services.catalog_service import _run_blocking console.print(Panel("Initializing Ad Seller System...", title="Setup")) # Read the default catalog from the single cached service source (EP-3.2) # rather than spinning up ProductSetupFlow / an OpenDirect MCP session. - products = catalog_service.get_static_product_catalog()["products"] + # Bridged via _run_blocking since this Typer command is synchronous but + # the catalog accessor is async (AI-6: live reads for synced ad servers). + products = _run_blocking(catalog_service.get_static_product_catalog())["products"] console.print(f"[green]✓[/green] Organization '{organization_name}' initialized") console.print(f"[green]✓[/green] Created {len(products)} default products") @@ -79,8 +82,9 @@ def init( def catalog(): """View the product catalog.""" from ...services import catalog_service + from ...services.catalog_service import _run_blocking - products = catalog_service.get_static_product_catalog()["products"] + products = _run_blocking(catalog_service.get_static_product_catalog())["products"] table = Table(title="Product Catalog") table.add_column("ID", style="cyan") @@ -115,9 +119,10 @@ def price( """Get pricing for a product based on buyer tier.""" from ...models.buyer_identity import AccessTier, BuyerContext, BuyerIdentity from ...services import catalog_service, quote_service + from ...services.catalog_service import _run_blocking # Product from the single cached catalog source (EP-3.2) - products = catalog_service.get_static_product_catalog()["products"] + products = _run_blocking(catalog_service.get_static_product_catalog())["products"] product = products.get(product_id) if not product: diff --git a/src/ad_seller/interfaces/mcp_server.py b/src/ad_seller/interfaces/mcp_server.py index 0170769d..d7ba8cbb 100644 --- a/src/ad_seller/interfaces/mcp_server.py +++ b/src/ad_seller/interfaces/mcp_server.py @@ -92,11 +92,11 @@ async def _service_json(coro) -> str: return _dumps(result) -def _static_catalog() -> dict[str, Any]: +async def _static_catalog() -> dict[str, Any]: """Single cached product catalog source (same one the REST routes read).""" from ..services import catalog_service - return catalog_service.get_static_product_catalog() + return await catalog_service.get_static_product_catalog() def _public_context(): @@ -375,7 +375,7 @@ async def list_products(limit: int | None = 50) -> str: # Read from the single cached catalog source (EP-3.3) instead of running # ProductSetupFlow per call (which spins up an OpenDirect MCP session that # hangs in session.initialize()). Same source the REST /products route uses. - catalog = catalog_service.get_static_product_catalog() + catalog = await catalog_service.get_static_product_catalog() products = [] for pid, product in list(catalog["products"].items())[:limit]: @@ -566,7 +566,7 @@ async def get_pricing(product_id: str, buyer_tier: str = "public", volume: int = # quote_service.get_pricing the REST /pricing route calls (single source) # instead of re-instantiating ProductSetupFlow/TieredPricingConfig/ # PricingRulesEngine independently. - catalog = catalog_service.get_static_product_catalog() + catalog = await catalog_service.get_static_product_catalog() product = catalog["products"].get(product_id) if not product: @@ -614,9 +614,8 @@ async def request_quote(product_id: str, deal_type: str = "PD", impressions: int target_cpm=None, buyer_identity=None, ) - return await _service_json( - quote_service.create_quote(request, _public_context(), _static_catalog()) - ) + catalog = await _static_catalog() + return await _service_json(quote_service.create_quote(request, _public_context(), catalog)) @mcp.tool() @@ -651,8 +650,9 @@ async def create_deal_from_template( notes=None, ) try: + catalog = await _static_catalog() deal_data = await deal_service.create_deal_from_template( - request, _public_context(), _static_catalog() + request, _public_context(), catalog ) except HTTPException as exc: return _dumps({"detail": exc.detail}) @@ -1087,7 +1087,8 @@ async def create_curated_deal( audience_segments=[], content_categories=[], ) - return await _service_json(deal_service.create_curated_deal(request, _static_catalog())) + catalog = await _static_catalog() + return await _service_json(deal_service.create_curated_deal(request, catalog)) # ============================================================================= diff --git a/src/ad_seller/services/catalog_service.py b/src/ad_seller/services/catalog_service.py index 03a82fab..ef885235 100644 --- a/src/ad_seller/services/catalog_service.py +++ b/src/ad_seller/services/catalog_service.py @@ -408,21 +408,126 @@ def build_static_product_catalog() -> dict[str, Any]: } -def get_static_product_catalog() -> dict[str, Any]: +def _real_ad_server_configured(settings: Any) -> bool: + """True when a real (non-CSV) ad server is configured (AI-6). + + Mirrors ``ProductSetupFlow.sync_from_ad_server``'s own "is this a real + ad server, not the mock/dev fallback" condition, minus the CSV case + (``_csv_mode_active`` handles that separately). + """ + return bool( + settings.gam_network_code + or settings.freewheel_sh_mcp_url + or settings.ad_server_type == "s3" + ) + + +async def persist_synced_product(product: Any) -> None: + """Persist a product built from a real ad-server sync (AI-6). + + Stored under a dedicated ``synced_product:{id}`` key, deliberately + NOT the generic ``product:{id}`` key ``negotiation_service`` and + ``media_kit_service`` already write/read for an unrelated reason (a + per-product snapshot at proposal time, in a narrower shape via + ``serialize_product``, so a later cold lookup doesn't 404). Reusing + that key would make :func:`build_synced_product_catalog` treat an + incidental negotiation snapshot of a STATIC-catalog product as + evidence a real ad-server sync had happened, and ``ProductDefinition + (**data)`` would likely fail to reconstruct from the narrower + snapshot shape anyway. + """ + from ..storage.factory import get_storage + + storage = await get_storage() + await storage.set(f"synced_product:{product.product_id}", product.model_dump(mode="json")) + + +async def prune_stale_synced_products(seen_product_ids: set[str]) -> None: + """Delete synced products not re-seeded by the current sync run (AI-6). + + Mirrors ``ProductSetupFlow._prune_stale_synced_packages`` (issue #34): + only products actually returned by the ad server on this run survive, + so a product removed from the ad server stops being served instead of + lingering forever as phantom inventory. + """ + from ..storage.factory import get_storage + + storage = await get_storage() + for key in await storage.keys("synced_product:*"): + product_id = key.removeprefix("synced_product:") + if product_id not in seen_product_ids: + await storage.delete(key) + logger.info("Pruned stale synced product: %s", product_id) + + +async def build_synced_product_catalog() -> Optional[dict[str, Any]]: + """Build the catalog from persisted ad-server-synced products (AI-6). + + Returns ``None`` when nothing has been synced yet (no ad-server sync + has ever run), so the caller falls back to the static default catalog + — the same "honest, never fabricated" fallback CSV mode already uses + on a read failure. + """ + from ..models.flow_state import ProductDefinition + from ..storage.factory import get_storage + + storage = await get_storage() + keys = await storage.keys("synced_product:*") + if not keys: + return None + + products: dict[str, Any] = {} + for key in keys: + data = await storage.get(key) + if not data: + continue + product_def = ProductDefinition(**data) + products[product_def.product_id] = product_def + + if not products: + return None + + return { + "products": products, + "inventory_types": sorted({p.inventory_type for p in products.values()}), + } + + +async def get_static_product_catalog() -> dict[str, Any]: """Return the seller's product catalog without running the flow. In CSV mode (``AD_SERVER_TYPE=csv``) the catalog is built from the CSV - inventory; in every other mode it is the static default catalog - (byte-identical to the pre-CSV-wiring behavior). - - Cached — repeated reads return stable product_ids (issue #34). + inventory, cached once per process (issue #34). + + When a real ad server (GAM/FreeWheel/S3) is configured AND has been + synced at least once, the catalog reflects those real, persisted + products (AI-6) — read fresh from storage on EVERY call, never + cached, so every worker process agrees immediately after a sync + (matching how packages already work) instead of only whichever + worker happened to run the sync seeing it. Storage reads are cheap + local I/O, not a live ad-server call, so this doesn't reintroduce the + per-request hang risk that CSV-only caching was built to avoid. + + Otherwise (no ad server configured, or configured but never yet + synced) the static default catalog is served, cached once, exactly + as before. """ global _CATALOG_CACHE - if _CATALOG_CACHE is None: - if _csv_mode_active(): + + if _csv_mode_active(): + if _CATALOG_CACHE is None: _CATALOG_CACHE = build_csv_product_catalog() - else: - _CATALOG_CACHE = build_static_product_catalog() + return _CATALOG_CACHE + + from ..config import get_settings + + if _real_ad_server_configured(get_settings()): + synced = await build_synced_product_catalog() + if synced is not None: + return synced + + if _CATALOG_CACHE is None: + _CATALOG_CACHE = build_static_product_catalog() return _CATALOG_CACHE diff --git a/src/ad_seller/services/negotiation_service.py b/src/ad_seller/services/negotiation_service.py index 684afa7f..a4823d21 100644 --- a/src/ad_seller/services/negotiation_service.py +++ b/src/ad_seller/services/negotiation_service.py @@ -232,7 +232,7 @@ async def counter_proposal( # before giving up. from .catalog_service import get_static_product_catalog, serialize_product - product = get_static_product_catalog()["products"].get(product_id) + product = (await get_static_product_catalog())["products"].get(product_id) if product is None: raise HTTPException(status_code=404, detail="Product not found") product_data = serialize_product(product) diff --git a/src/ad_seller/tools/catalog/avails_tool.py b/src/ad_seller/tools/catalog/avails_tool.py index b04b7150..76b3d99d 100644 --- a/src/ad_seller/tools/catalog/avails_tool.py +++ b/src/ad_seller/tools/catalog/avails_tool.py @@ -74,8 +74,9 @@ def _run( inventory_type: Optional[str] = None, ) -> str: from ...services import catalog_service + from ...services.catalog_service import _run_blocking - catalog = catalog_service.get_static_product_catalog() + catalog = _run_blocking(catalog_service.get_static_product_catalog()) products = catalog["products"] if product_id is None: diff --git a/tests/unit/test_csv_catalog_coherence.py b/tests/unit/test_csv_catalog_coherence.py index bdabc4c0..37b9e98e 100644 --- a/tests/unit/test_csv_catalog_coherence.py +++ b/tests/unit/test_csv_catalog_coherence.py @@ -173,15 +173,15 @@ def _wire_product_ids(body: dict) -> set[str]: class TestCsvCatalogService: - def test_catalog_serves_csv_inventory_in_csv_mode(self, csv_mode): - catalog = catalog_service.get_static_product_catalog() + async def test_catalog_serves_csv_inventory_in_csv_mode(self, csv_mode): + catalog = await catalog_service.get_static_product_catalog() assert set(catalog["products"].keys()) == CSV_IDS, ( "CSV mode must serve the CSV inventory as the product catalog, " f"got: {sorted(catalog['products'].keys())}" ) - def test_csv_products_carry_row_data(self, csv_mode): - catalog = catalog_service.get_static_product_catalog() + async def test_csv_products_carry_row_data(self, csv_mode): + catalog = await catalog_service.get_static_product_catalog() product = catalog["products"]["inv-t-ctv-drama"] assert product.name == "Test CTV Drama" assert product.base_cpm == 30.0 @@ -189,18 +189,18 @@ def test_csv_products_carry_row_data(self, csv_mode): assert product.inventory_type == "ctv" assert "ctv" in catalog["inventory_types"] - def test_csv_catalog_ids_stable_within_process(self, csv_mode): - first = catalog_service.get_static_product_catalog() - second = catalog_service.get_static_product_catalog() + async def test_csv_catalog_ids_stable_within_process(self, csv_mode): + first = await catalog_service.get_static_product_catalog() + second = await catalog_service.get_static_product_catalog() assert first is second, "single-cache design: catalog built once per process" assert list(first["products"].keys()) == list(second["products"].keys()) - def test_csv_catalog_ids_deterministic_across_rebuild(self, csv_mode): + async def test_csv_catalog_ids_deterministic_across_rebuild(self, csv_mode): """Simulated reboot: fresh cache over the same CSV yields the same ids.""" - first_ids = set(catalog_service.get_static_product_catalog()["products"]) + first_ids = set((await catalog_service.get_static_product_catalog())["products"]) catalog_service.reset_catalog_cache() api_main._STATIC_PRODUCT_CATALOG = None - second_ids = set(catalog_service.get_static_product_catalog()["products"]) + second_ids = set((await catalog_service.get_static_product_catalog())["products"]) assert first_ids == second_ids == CSV_IDS @@ -210,28 +210,28 @@ def test_csv_catalog_ids_deterministic_across_rebuild(self, csv_mode): class TestNonCsvModesUnchanged: - def test_default_mode_serves_default_catalog(self, default_mode): - catalog = catalog_service.get_static_product_catalog() + async def test_default_mode_serves_default_catalog(self, default_mode): + catalog = await catalog_service.get_static_product_catalog() expected_names = [c["name"] for c in catalog_service.DEFAULT_PRODUCT_CONFIGS] assert [p.name for p in catalog["products"].values()] == expected_names assert len(catalog["products"]) == 13 - def test_default_mode_ids_stay_uuid_shaped_and_stable(self, default_mode): + async def test_default_mode_ids_stay_uuid_shaped_and_stable(self, default_mode): import re - first = catalog_service.get_static_product_catalog() + first = await catalog_service.get_static_product_catalog() for pid in first["products"]: assert re.fullmatch(r"prod-[0-9a-f]{8}", pid), pid - second = catalog_service.get_static_product_catalog() + second = await catalog_service.get_static_product_catalog() assert list(first["products"].keys()) == list(second["products"].keys()) - def test_default_mode_ids_survive_cache_reset(self, default_mode): + async def test_default_mode_ids_survive_cache_reset(self, default_mode): # Proxy for multi-worker and restart behavior (issue #34): each uvicorn # worker builds its own catalog cache, so ids must be deterministic # across independent builds, not merely stable within one cache. - first = catalog_service.get_static_product_catalog() + first = await catalog_service.get_static_product_catalog() catalog_service.reset_catalog_cache() - second = catalog_service.get_static_product_catalog() + second = await catalog_service.get_static_product_catalog() assert list(first["products"].keys()) == list(second["products"].keys()) assert len(set(first["products"])) == len(catalog_service.DEFAULT_PRODUCT_CONFIGS) diff --git a/tests/unit/test_issue34_catalog_fixes.py b/tests/unit/test_issue34_catalog_fixes.py index eab74b58..3fcb28de 100644 --- a/tests/unit/test_issue34_catalog_fixes.py +++ b/tests/unit/test_issue34_catalog_fixes.py @@ -128,9 +128,9 @@ def client(storage): yield httpx.AsyncClient(transport=transport, base_url="http://test") -def _catalog_product_ids(n: int = 2) -> list[str]: +async def _catalog_product_ids(n: int = 2) -> list[str]: """Return the first n product ids from the static catalog (as served by GET /products).""" - catalog = deps.get_product_catalog() + catalog = await deps.get_product_catalog() ids = list(catalog["products"].keys()) assert len(ids) >= n, "static catalog unexpectedly small" return ids[:n] @@ -198,7 +198,7 @@ async def test_zero_resolution_returns_422_naming_ids(self, client): assert "prod-nope-2" in detail["message"] async def test_partial_resolution_returns_warning_and_unresolved_ids(self, client): - good_id = _catalog_product_ids(1)[0] + good_id = (await _catalog_product_ids(1))[0] async with client as c: resp = await c.post( "/packages", @@ -278,7 +278,7 @@ async def test_zero_resolution_returns_422_naming_ids(self, client): assert "prod-nope-1" in detail["message"] async def test_partial_resolution_reports_unresolved_ids(self, client): - good_id = _catalog_product_ids(1)[0] + good_id = (await _catalog_product_ids(1))[0] async with client as c: resp = await c.post( "/packages/assemble", diff --git a/tests/unit/test_service_layer.py b/tests/unit/test_service_layer.py index 33888c88..fefc3b2d 100644 --- a/tests/unit/test_service_layer.py +++ b/tests/unit/test_service_layer.py @@ -126,11 +126,11 @@ def _pin_default_mode(self, monkeypatch): yield get_settings.cache_clear() - def test_catalog_is_cached_with_stable_product_ids(self): + async def test_catalog_is_cached_with_stable_product_ids(self): """Happy: repeated reads return the SAME catalog object and ids.""" catalog_service.reset_catalog_cache() - first = catalog_service.get_static_product_catalog() - second = catalog_service.get_static_product_catalog() + first = await catalog_service.get_static_product_catalog() + second = await catalog_service.get_static_product_catalog() assert first is second assert len(first["products"]) == len(catalog_service.DEFAULT_PRODUCT_CONFIGS) @@ -139,7 +139,7 @@ def test_catalog_is_cached_with_stable_product_ids(self): } catalog_service.reset_catalog_cache() - def test_reset_preserves_stable_product_ids(self): + async def test_reset_preserves_stable_product_ids(self): """Ids are deterministic: a cache reset must NOT mint new ids. Contract change with the deterministic-id fix (issue #34): ids derive @@ -148,9 +148,9 @@ def test_reset_preserves_stable_product_ids(self): broke follow-up lookups for external integrators. """ catalog_service.reset_catalog_cache() - ids_before = set(catalog_service.get_static_product_catalog()["products"]) + ids_before = set((await catalog_service.get_static_product_catalog())["products"]) catalog_service.reset_catalog_cache() - ids_after = set(catalog_service.get_static_product_catalog()["products"]) + ids_after = set((await catalog_service.get_static_product_catalog())["products"]) assert ids_before == ids_after catalog_service.reset_catalog_cache() diff --git a/tests/unit/test_synced_product_catalog.py b/tests/unit/test_synced_product_catalog.py new file mode 100644 index 00000000..c0cdf243 --- /dev/null +++ b/tests/unit/test_synced_product_catalog.py @@ -0,0 +1,357 @@ +# Author: Green Mountain Systems AI Inc. +# Donated to IAB Tech Lab + +"""AI-6: real, storage-backed catalog for synced ad servers. + +Defect: ``GET /products`` (and everything reading the catalog) served the +static 13-product default catalog even after a real ad-server sync +(GAM/FreeWheel/S3) had run. ``ProductSetupFlow.state.products`` is +per-flow-instance and never outlives the request, and the catalog +service's in-memory cache is per-worker-process (``infra/docker/Dockerfile`` +runs uvicorn with ``--workers 2``), so at most one of two worker processes +ever saw synced data. + +Fix under test: synced products are persisted to a dedicated +``synced_product:{id}`` storage key -- deliberately NOT the generic +``product:{id}`` key ``negotiation_service``/``media_kit_service`` already +write/read for an unrelated reason (a proposal-time snapshot, narrower +shape) -- and served fresh from storage on every catalog read when a real +ad server is configured, matching how packages already work (issue #34). +""" + +from types import SimpleNamespace +from typing import Any, Optional +from unittest.mock import patch + +import pytest + +from ad_seller.models.core import DealType, PricingModel +from ad_seller.models.flow_state import ProductDefinition +from ad_seller.services import catalog_service +from ad_seller.storage.base import StorageBackend + + +class InMemoryStorage(StorageBackend): + """Fully in-memory storage backend for tests.""" + + def __init__(self) -> None: + self._data: dict[str, Any] = {} + + async def connect(self) -> None: + pass + + async def disconnect(self) -> None: + self._data.clear() + + async def get(self, key: str) -> Optional[Any]: + return self._data.get(key) + + async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None: + self._data[key] = value + + async def delete(self, key: str) -> bool: + return self._data.pop(key, None) is not None + + async def exists(self, key: str) -> bool: + return key in self._data + + async def keys(self, pattern: str = "*") -> list[str]: + import fnmatch + + return [k for k in self._data if fnmatch.fnmatch(k, pattern)] + + +@pytest.fixture +def storage() -> InMemoryStorage: + return InMemoryStorage() + + +@pytest.fixture(autouse=True) +def _reset_catalog_cache(): + """Reset the static-catalog cache between tests (issue #34 convention).""" + catalog_service.reset_catalog_cache() + yield + catalog_service.reset_catalog_cache() + + +def _product( + product_id: str = "prod-a", + inventory_type: str = "display", + base_cpm: float = 10.0, +) -> ProductDefinition: + return ProductDefinition( + product_id=product_id, + name=f"Product {product_id}", + inventory_type=inventory_type, + supported_deal_types=[DealType.PREFERRED_DEAL], + supported_pricing_models=[PricingModel.CPM], + base_cpm=base_cpm, + floor_cpm=round(base_cpm * 0.85, 2), + ) + + +# ============================================================================= +# persist_synced_product / prune_stale_synced_products +# ============================================================================= + + +class TestPersistSyncedProduct: + async def test_writes_to_dedicated_synced_product_key(self, storage): + product = _product("prod-a") + with patch("ad_seller.storage.factory.get_storage", return_value=storage): + await catalog_service.persist_synced_product(product) + + assert await storage.exists("synced_product:prod-a") + # Never the generic key negotiation_service/media_kit_service use. + assert not await storage.exists("product:prod-a") + + async def test_does_not_treat_negotiation_snapshot_as_a_sync(self, storage): + """Storage-key-namespace isolation: an unrelated ``product:{id}`` + write (negotiation_service's proposal-time snapshot, via + ``serialize_product``'s narrower 7-field shape) must never be read + as evidence a real ad-server sync happened.""" + await storage.set( + "product:prod-b", + {"product_id": "prod-b", "name": "Snapshot only", "inventory_type": "display"}, + ) + + with patch("ad_seller.storage.factory.get_storage", return_value=storage): + catalog = await catalog_service.build_synced_product_catalog() + + assert catalog is None, "a generic product:{id} snapshot must not count as a sync" + + +class TestPruneStaleSyncedProducts: + async def test_removes_products_not_in_seen_set(self, storage): + await storage.set("synced_product:keep", _product("keep").model_dump(mode="json")) + await storage.set("synced_product:drop", _product("drop").model_dump(mode="json")) + + with patch("ad_seller.storage.factory.get_storage", return_value=storage): + await catalog_service.prune_stale_synced_products({"keep"}) + + assert await storage.exists("synced_product:keep") + assert not await storage.exists("synced_product:drop") + + async def test_noop_when_all_seen(self, storage): + await storage.set("synced_product:a", _product("a").model_dump(mode="json")) + + with patch("ad_seller.storage.factory.get_storage", return_value=storage): + await catalog_service.prune_stale_synced_products({"a"}) + + assert await storage.exists("synced_product:a") + + +# ============================================================================= +# build_synced_product_catalog +# ============================================================================= + + +class TestBuildSyncedProductCatalog: + async def test_returns_none_when_nothing_synced(self, storage): + with patch("ad_seller.storage.factory.get_storage", return_value=storage): + assert await catalog_service.build_synced_product_catalog() is None + + async def test_reconstructs_products_from_storage(self, storage): + product = _product("prod-x", inventory_type="ctv", base_cpm=30.0) + await storage.set("synced_product:prod-x", product.model_dump(mode="json")) + + with patch("ad_seller.storage.factory.get_storage", return_value=storage): + catalog = await catalog_service.build_synced_product_catalog() + + assert catalog is not None + assert set(catalog["products"]) == {"prod-x"} + assert catalog["products"]["prod-x"].base_cpm == 30.0 + assert catalog["inventory_types"] == ["ctv"] + + +# ============================================================================= +# _real_ad_server_configured +# ============================================================================= + + +class TestRealAdServerConfigured: + @pytest.mark.parametrize( + "kwargs,expected", + [ + ( + { + "gam_network_code": "12345", + "freewheel_sh_mcp_url": None, + "ad_server_type": "none", + }, + True, + ), + ( + { + "gam_network_code": None, + "freewheel_sh_mcp_url": "https://fw.example", + "ad_server_type": "none", + }, + True, + ), + ( + {"gam_network_code": None, "freewheel_sh_mcp_url": None, "ad_server_type": "s3"}, + True, + ), + ( + {"gam_network_code": None, "freewheel_sh_mcp_url": None, "ad_server_type": "csv"}, + False, + ), + ( + {"gam_network_code": None, "freewheel_sh_mcp_url": None, "ad_server_type": "none"}, + False, + ), + ], + ) + def test_matrix(self, kwargs, expected): + settings = SimpleNamespace(**kwargs) + assert catalog_service._real_ad_server_configured(settings) is expected + + +# ============================================================================= +# get_static_product_catalog — the live-read invariant +# ============================================================================= + + +class TestGetStaticProductCatalogLiveRead: + async def test_serves_synced_catalog_live_every_call_not_cached(self, storage): + """The core AI-6 invariant: a product persisted AFTER the first + catalog read must appear on the very next read -- no restart, no + explicit cache-reset call -- proving reads are live, not cached.""" + settings = SimpleNamespace( + gam_network_code="12345", freewheel_sh_mcp_url=None, ad_server_type="none" + ) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=storage), + patch("ad_seller.config.get_settings", return_value=settings), + ): + first = await catalog_service.get_static_product_catalog() + + await catalog_service.persist_synced_product( + _product("prod-live", inventory_type="native", base_cpm=42.0) + ) + + second = await catalog_service.get_static_product_catalog() + + # Before any sync: honest fallback to the static default catalog. + assert len(first["products"]) == len(catalog_service.DEFAULT_PRODUCT_CONFIGS) + # Immediately after persisting, with NO cache reset in between. + assert set(second["products"]) == {"prod-live"} + assert second["products"]["prod-live"].base_cpm == 42.0 + + async def test_non_real_ad_server_ignores_synced_products(self, storage): + """A synced_product key with no real ad server configured (e.g. a + stale key from a prior config) must not leak into the served + catalog -- only CSV/static-default apply.""" + settings = SimpleNamespace( + gam_network_code=None, freewheel_sh_mcp_url=None, ad_server_type="none" + ) + await storage.set("synced_product:orphan", _product("orphan").model_dump(mode="json")) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=storage), + patch("ad_seller.config.get_settings", return_value=settings), + ): + catalog = await catalog_service.get_static_product_catalog() + + assert "orphan" not in catalog["products"] + assert len(catalog["products"]) == len(catalog_service.DEFAULT_PRODUCT_CONFIGS) + + +# ============================================================================= +# End-to-end: ProductSetupFlow.sync_from_ad_server persists + prunes +# ============================================================================= + + +class _FakeInventoryItem: + def __init__(self, id: str, name: str, raw: Optional[dict] = None): + self.id = id + self.name = name + self.raw = raw or {} + + +class _FakeAdServerClient: + ad_server_type = SimpleNamespace(value="gam") + + def __init__(self, items): + self._items = items + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def list_inventory(self, filter_str=None): + return self._items + + +def _flow_settings(**overrides) -> SimpleNamespace: + defaults = { + "gam_network_code": "12345", + "freewheel_sh_mcp_url": None, + "ad_server_type": "none", + "seller_organization_id": "test-seller-org", + "seller_organization_name": "Test Seller", + } + defaults.update(overrides) + return SimpleNamespace(**defaults) + + +async def _run_sync(storage, settings, client) -> None: + with ( + patch("ad_seller.flows.product_setup_flow.get_settings", return_value=settings), + patch("ad_seller.storage.factory.get_storage", return_value=storage), + patch("ad_seller.clients.ad_server_base.get_ad_server_client", lambda: client), + ): + from ad_seller.flows.product_setup_flow import ProductSetupFlow + + flow = ProductSetupFlow() + await flow.sync_from_ad_server() + + +class TestSyncPersistsAndPrunesProducts: + async def test_sync_persists_products_to_storage(self, storage): + client = _FakeAdServerClient( + [ + _FakeInventoryItem("gam-001", "Homepage Display Banner", {"floor_price_cpm": 8.0}), + _FakeInventoryItem("gam-002", "CTV Premium App", {"floor_price_cpm": 20.0}), + ] + ) + await _run_sync(storage, _flow_settings(), client) + + assert await storage.exists("synced_product:gam-001") + assert await storage.exists("synced_product:gam-002") + + async def test_resync_with_fewer_items_prunes_the_dropped_product(self, storage): + settings = _flow_settings() + client_a = _FakeAdServerClient( + [ + _FakeInventoryItem("gam-001", "Homepage Display Banner"), + _FakeInventoryItem("gam-002", "CTV Premium App"), + ] + ) + await _run_sync(storage, settings, client_a) + + client_b = _FakeAdServerClient([_FakeInventoryItem("gam-001", "Homepage Display Banner")]) + await _run_sync(storage, settings, client_b) + + assert await storage.exists("synced_product:gam-001") + assert not await storage.exists("synced_product:gam-002"), ( + "product dropped from the ad server must be pruned, not left as phantom inventory" + ) + + async def test_catalog_reflects_synced_products_live_after_flow_sync(self, storage): + """End-to-end: GET /products' data source reflects a sync immediately.""" + settings = _flow_settings() + client = _FakeAdServerClient([_FakeInventoryItem("gam-999", "Pre-Roll Video Unit")]) + await _run_sync(storage, settings, client) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=storage), + patch("ad_seller.config.get_settings", return_value=settings), + ): + catalog = await catalog_service.get_static_product_catalog() + + assert set(catalog["products"]) == {"gam-999"} From b2b64c43a296af022f33552399bd6f995aaea136 Mon Sep 17 00:00:00 2001 From: Sirajmx Date: Mon, 21 Sep 2026 23:26:31 +0530 Subject: [PATCH 2/9] fix: GAMRestClient can now reach Google's Ad Manager API; production image installs the gam extra (AI-6) --- CHANGELOG.md | 24 ++++++++++++++++++++++++ infra/docker/Dockerfile | 2 +- src/ad_seller/clients/gam_rest_client.py | 11 ++++++++++- 3 files changed, 35 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 54ce23ec..e7042dc0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,30 @@ All notable changes to the IAB Tech Lab Seller Agent are documented here. already writes for an unrelated reason (a narrower proposal-time snapshot), and are pruned when no longer returned by the ad server. CSV-mode and no-ad-server-configured behavior is unchanged. +- `GAMRestClient` can now actually connect to Google Ad Manager (AI-6). + It called `googleapiclient.discovery.build("admanager", "v1", ...)` + with no `discoveryServiceUrl`; Ad Manager self-hosts its own discovery + document rather than registering with the generic, centrally-indexed + discovery service `build()` queries by default (and it isn't in the + locally bundled discovery docs either), so every real connection + attempt raised `UnknownApiNameOrVersion` and `sync_from_ad_server()`'s + broad exception handler silently degraded to mock synced packages — + real GAM inventory sync had never actually succeeded. Fixed by pointing + `build()` at `admanager.googleapis.com/$discovery/rest?version=v1` + directly (with `static_discovery=False`). Verified against a live GAM + network: `list_inventory()` returned 100 real ad units, a real + `POST /packages/sync` completed with zero warnings (previously always + warned and fell back to mocks), and `GET /products` served 50 real + synced products end to end. +- `infra/docker/Dockerfile` now installs the `gam` extra alongside + `redis` (AI-6): the production image only ever installed + `".[redis]"`, so `googleads`/`google-api-python-client`/`google-auth` + were never present, and a real GAM deployment would `ImportError` + before ever reaching `GAMRestClient.connect()` — even with the + discovery-URL fix above in place. Rebuilt the real Dockerfile from + scratch with no manual installs and re-ran the full live-GAM + verification (100 real ad units, clean sync, 50 real products served) + to confirm the production image now works end to end unmodified. - Map internal deal status to the shared wire enum on read; deals created via from-template, bulk, or curated paths no longer 500 on GET (#73). Internal `confirmed` reads as `booked`, internal diff --git a/infra/docker/Dockerfile b/infra/docker/Dockerfile index 8fcfe645..eb32f05e 100644 --- a/infra/docker/Dockerfile +++ b/infra/docker/Dockerfile @@ -27,7 +27,7 @@ ENV PATH="/opt/venv/bin:$PATH" COPY . . RUN pip install --no-cache-dir --upgrade pip \ - && pip install --no-cache-dir -e ".[redis]" + && pip install --no-cache-dir -e ".[redis,gam]" # --------------------------------------------------------------------------- # Stage 2 — Runtime diff --git a/src/ad_seller/clients/gam_rest_client.py b/src/ad_seller/clients/gam_rest_client.py index c2993696..48fb9586 100644 --- a/src/ad_seller/clients/gam_rest_client.py +++ b/src/ad_seller/clients/gam_rest_client.py @@ -79,11 +79,20 @@ async def connect(self) -> None: scopes=["https://www.googleapis.com/auth/admanager"], ) - # Build the GAM API service + # Build the GAM API service. Ad Manager self-hosts its own + # discovery document (admanager.googleapis.com/$discovery/rest) + # rather than registering with the generic, centrally-indexed + # discovery service googleapiclient.discovery.build() queries + # by default (and it isn't in the locally bundled discovery + # docs either) -- without an explicit discoveryServiceUrl this + # raises UnknownApiNameOrVersion for every real connection + # attempt, silently degrading every GAM sync to mock data. self._service = build( "admanager", "v1", credentials=self._credentials, + discoveryServiceUrl="https://admanager.googleapis.com/$discovery/rest?version=v1", + static_discovery=False, cache_discovery=False, ) except ImportError: From 4751cebb2b2ba106627529babb8c94cb4b3e6eca Mon Sep 17 00:00:00 2001 From: Sirajmx Date: Thu, 15 Oct 2026 17:18:49 +0530 Subject: [PATCH 3/9] fix: empty but completed ad-server sync must serve an honest empty catalog, not the static demo products (AI-6) A successful sync that finds zero inventory items left zero synced_product:* keys, indistinguishable from "never synced" -- build_synced_product_catalog() fell back to the 13-product static demo catalog, silently serving fake products as if real. Add a synced_catalog:completed marker, set only on the real ad-server success path (never the mock path). The catalog is now treated as real/synced if EITHER a synced_product:* key exists (covers products persisted mid-sync, before the marker is set) OR the completion marker is present (covers a completed sync that genuinely found nothing) -- only "neither" still falls back to the static catalog. Found by aleksUIX in PR #117 review. Verified with a new unit regression test plus a full e2e pass (real FastAPI app, real sqlite storage, real sync method, only the GAM network call faked): sync with items -> empty sync -> simulated transient failure -> recovery sync, confirmed correct at each step, including from a second independent server process reading the same storage. --- src/ad_seller/flows/product_setup_flow.py | 18 +++++++-- src/ad_seller/services/catalog_service.py | 45 +++++++++++++++++++---- tests/unit/test_synced_product_catalog.py | 31 ++++++++++++++++ 3 files changed, 83 insertions(+), 11 deletions(-) diff --git a/src/ad_seller/flows/product_setup_flow.py b/src/ad_seller/flows/product_setup_flow.py index 6b1998bd..06b2fcce 100644 --- a/src/ad_seller/flows/product_setup_flow.py +++ b/src/ad_seller/flows/product_setup_flow.py @@ -237,9 +237,9 @@ async def sync_from_ad_server(self) -> None: self.state.ad_server_sync_failed = True return - await self._finish_sync() + await self._finish_sync(real_sync_completed=True) - async def _finish_sync(self) -> None: + async def _finish_sync(self, real_sync_completed: bool = False) -> None: """Terminal step of every sync path: prune the stale synced layer. Runs unconditionally, including on the mock/fallback path (no ad @@ -249,13 +249,25 @@ async def _finish_sync(self) -> None: the honest static-default fallback rather than serving increasingly stale synced products forever, and the next successful sync immediately re-populates it. + + ``real_sync_completed`` is only ever True from the real ad-server + success path, never the mock path -- a zero-item real sync is a + genuine "no inventory right now" result that must be served as an + honestly empty catalog, not mistaken for "never synced" (AI-6 + follow-up); the mock path has no such distinction to make, since + it never persists ``synced_product:*`` entries at all. """ - from ..services.catalog_service import prune_stale_synced_products + from ..services.catalog_service import ( + mark_synced_catalog_completed, + prune_stale_synced_products, + ) from ..storage.factory import get_storage storage = await get_storage() await self._prune_stale_synced_packages(storage) await prune_stale_synced_products(set(self.state.products.keys())) + if real_sync_completed: + await mark_synced_catalog_completed() async def _create_mock_synced_packages(self) -> None: """Create mock Layer 1 packages for development without ad server creds.""" diff --git a/src/ad_seller/services/catalog_service.py b/src/ad_seller/services/catalog_service.py index 83857555..19a0f35e 100644 --- a/src/ad_seller/services/catalog_service.py +++ b/src/ad_seller/services/catalog_service.py @@ -467,20 +467,52 @@ async def prune_stale_synced_products(seen_product_ids: set[str]) -> None: logger.info("Pruned stale synced product: %s", product_id) +_SYNCED_CATALOG_COMPLETED_KEY = "synced_catalog:completed" + + +async def mark_synced_catalog_completed() -> None: + """Record that a real ad-server sync has completed at least once (AI-6). + + A zero-item sync is a real, successful sync — the seller's ad server + genuinely has no (matching) inventory right now — and must be told + apart from "no sync has ever run": both leave zero ``synced_product:*`` + keys, but only the latter should fall back to the static demo catalog. + Set once a real sync's item loop completes, even with zero items; + never cleared (a seller who has synced once stays in "trust storage, + not the static fallback" mode from then on). + """ + from ..storage.factory import get_storage + + storage = await get_storage() + await storage.set(_SYNCED_CATALOG_COMPLETED_KEY, True) + + async def build_synced_product_catalog() -> Optional[dict[str, Any]]: """Build the catalog from persisted ad-server-synced products (AI-6). - Returns ``None`` when nothing has been synced yet (no ad-server sync - has ever run), so the caller falls back to the static default catalog - — the same "honest, never fabricated" fallback CSV mode already uses - on a read failure. + Returns ``None`` only when no real ad-server sync has ever run — no + ``synced_product:*`` keys exist AND no sync has ever completed — so + the caller falls back to the static default catalog — the same + "honest, never fabricated" fallback CSV mode already uses on a read + failure. + + Either signal alone is enough evidence of a real sync: + - ``synced_product:*`` keys may exist even mid-sync, before + ``mark_synced_catalog_completed`` runs (each item is persisted as + it's processed, in a loop that precedes the completion marker — + see ``ProductSetupFlow.sync_from_ad_server``), including the case + where a sync crashed partway through. + - the completion marker may be set with zero keys when a *completed* + sync genuinely found no inventory — a real, honest result that + must be served as an empty catalog (``products: {}``), not mistaken + for "never synced" and backfilled with stale demo products. """ from ..models.flow_state import ProductDefinition from ..storage.factory import get_storage storage = await get_storage() keys = await storage.keys("synced_product:*") - if not keys: + if not keys and not await storage.exists(_SYNCED_CATALOG_COMPLETED_KEY): return None products: dict[str, Any] = {} @@ -491,9 +523,6 @@ async def build_synced_product_catalog() -> Optional[dict[str, Any]]: product_def = ProductDefinition(**data) products[product_def.product_id] = product_def - if not products: - return None - return { "products": products, "inventory_types": sorted({p.inventory_type for p in products.values()}), diff --git a/tests/unit/test_synced_product_catalog.py b/tests/unit/test_synced_product_catalog.py index f4e9e493..e400b473 100644 --- a/tests/unit/test_synced_product_catalog.py +++ b/tests/unit/test_synced_product_catalog.py @@ -358,3 +358,34 @@ async def test_catalog_reflects_synced_products_live_after_flow_sync(self, stora catalog = await catalog_service.get_static_product_catalog() assert set(catalog["products"]) == {"gam-999"} + + async def test_resync_with_zero_items_serves_an_honestly_empty_catalog(self, storage): + """A real sync finding zero inventory must not resurrect demo products. + + Regression test: a zero-item sync is a genuine, successful result + (the ad server really has nothing matching right now), not the + absence of a sync. Both leave zero ``synced_product:*`` keys, so + without a separate "a real sync has completed" marker, the catalog + builder couldn't tell the two apart and silently fell back to the + 13-product static demo catalog -- fake inventory reappearing in + place of an honest empty one. + """ + settings = _flow_settings() + client_a = _FakeAdServerClient([_FakeInventoryItem("gam-999", "Pre-Roll Video Unit")]) + await _run_sync(storage, settings, client_a) + + client_b = _FakeAdServerClient([]) + await _run_sync(storage, settings, client_b) + + assert not await storage.exists("synced_product:gam-999") + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=storage), + patch("ad_seller.config.get_settings", return_value=settings), + ): + catalog = await catalog_service.get_static_product_catalog() + + assert catalog["products"] == {}, ( + "a real sync that found zero inventory must serve an empty catalog, " + "not fall back to the static demo products" + ) From dec2b499c477760b04ee6e3b6cd7c6663c1719b0 Mon Sep 17 00:00:00 2001 From: Sirajmx Date: Thu, 15 Oct 2026 18:43:00 +0530 Subject: [PATCH 4/9] fix: roll back partial writes when a sync crashes mid-loop, not just report it as failed (AI-6) persist_synced_product/_upsert_synced_package write straight to storage one item at a time inside the loop. An exception partway through left every item already written by THIS run permanently in storage, even though the sync is reported as failed -- contradicting the existing claim that a failure never silently changes the synced layer. Capture each key's prior value before writing it, and on any exception, roll back precisely: restore the prior value if the key already existed (a re-synced item can share an id with one an earlier successful sync persisted), or delete it if this run introduced it fresh. Same fix applied to both the product loop and the package loop (identical structural risk). Reproduced live: synced A+B cleanly, then a sync that persists C and crashes before D -- A and B survived as expected, but C leaked in anyway. Fixed; added as two permanent regression tests (the leak case, and the re-synced-existing-id-restores-correctly case). Full suite: 1729 passed, 0 failed. --- src/ad_seller/flows/product_setup_flow.py | 116 +++++++++++++++------- tests/unit/test_synced_product_catalog.py | 83 ++++++++++++++++ 2 files changed, 161 insertions(+), 38 deletions(-) diff --git a/src/ad_seller/flows/product_setup_flow.py b/src/ad_seller/flows/product_setup_flow.py index 06b2fcce..45633bad 100644 --- a/src/ad_seller/flows/product_setup_flow.py +++ b/src/ad_seller/flows/product_setup_flow.py @@ -182,47 +182,87 @@ async def sync_from_ad_server(self) -> None: product_from_inventory_item, ) - for item in items: - product_def = product_from_inventory_item(item) - self.state.products[product_def.product_id] = product_def - # Persist so GET /products reflects real ad-server inventory - # for non-CSV ad servers too (AI-6) -- state.products alone - # is per-flow-instance and never outlives this request. - await persist_synced_product(product_def) + # Persist every product, but roll back precisely on any + # mid-loop failure (AI-6 follow-up): persist_synced_product + # writes straight to storage per item, so without this, an + # exception on item N left items 1..N-1 of THIS run already + # committed even though the sync is reported as failed -- + # contradicting the "a transient failure never silently + # changes the synced layer" invariant above. Capturing each + # key's PRIOR value (not just blindly deleting it) matters + # because a re-synced item can share an id with one already + # persisted by an earlier successful sync; rollback must + # restore that old value, not erase still-valid data. + persisted: list[tuple[str, Optional[dict[str, Any]]]] = [] + try: + for item in items: + product_def = product_from_inventory_item(item) + key = f"synced_product:{product_def.product_id}" + prior = await storage.get(key) + self.state.products[product_def.product_id] = product_def + # Persist so GET /products reflects real ad-server inventory + # for non-CSV ad servers too (AI-6) -- state.products alone + # is per-flow-instance and never outlives this request. + await persist_synced_product(product_def) + persisted.append((product_def.product_id, prior)) + except Exception: + for product_id, prior in persisted: + key = f"synced_product:{product_id}" + if prior is None: + await storage.delete(key) + else: + await storage.set(key, prior) + self.state.products.pop(product_id, None) + raise logger.info("Created %d products from ad server inventory", len(self.state.products)) - for inv_type, inv_items in grouped.items(): - ad_formats = self._classify_ad_formats_from_type(inv_type) - device_types = self._classify_device_types_from_type(inv_type) - base_cpm = self._estimate_base_cpm(inv_type) - - package = Package( - package_id=self._synced_package_id(client.ad_server_type.value, inv_type), - name=f"{inv_type.replace('_', ' ').title()} - Synced", - description=f"Synced {inv_type} inventory ({len(inv_items)} ad units)", - layer=PackageLayer.SYNCED, - status=PackageStatus.ACTIVE, - placements=[ - PackagePlacement( - product_id=item.id, - product_name=item.name, - ad_formats=ad_formats, - device_types=device_types, - ) - for item in inv_items - ], - ad_formats=ad_formats, - device_types=device_types, - cat=["IAB1"], - cattax=2, - base_price=base_cpm, - floor_price=round(base_cpm * 0.7, 2), - ad_server_source=client.ad_server_type.value, - is_featured=inv_type == "ctv", - ) - - await self._upsert_synced_package(storage, package) + # Same rollback discipline for packages -- identical risk, + # same fix shape. + upserted: list[tuple[str, Optional[dict[str, Any]]]] = [] + try: + for inv_type, inv_items in grouped.items(): + ad_formats = self._classify_ad_formats_from_type(inv_type) + device_types = self._classify_device_types_from_type(inv_type) + base_cpm = self._estimate_base_cpm(inv_type) + + package = Package( + package_id=self._synced_package_id(client.ad_server_type.value, inv_type), + name=f"{inv_type.replace('_', ' ').title()} - Synced", + description=f"Synced {inv_type} inventory ({len(inv_items)} ad units)", + layer=PackageLayer.SYNCED, + status=PackageStatus.ACTIVE, + placements=[ + PackagePlacement( + product_id=item.id, + product_name=item.name, + ad_formats=ad_formats, + device_types=device_types, + ) + for item in inv_items + ], + ad_formats=ad_formats, + device_types=device_types, + cat=["IAB1"], + cattax=2, + base_price=base_cpm, + floor_price=round(base_cpm * 0.7, 2), + ad_server_source=client.ad_server_type.value, + is_featured=inv_type == "ctv", + ) + + prior_package = await storage.get_package(package.package_id) + await self._upsert_synced_package(storage, package) + upserted.append((package.package_id, prior_package)) + except Exception: + for package_id, prior_package in upserted: + if prior_package is None: + await storage.delete_package(package_id) + else: + await storage.set_package(package_id, prior_package) + if package_id in self.state.synced_segments: + self.state.synced_segments.remove(package_id) + raise logger.info("Synced %d packages from ad server", len(grouped)) diff --git a/tests/unit/test_synced_product_catalog.py b/tests/unit/test_synced_product_catalog.py index e400b473..3920c560 100644 --- a/tests/unit/test_synced_product_catalog.py +++ b/tests/unit/test_synced_product_catalog.py @@ -389,3 +389,86 @@ async def test_resync_with_zero_items_serves_an_honestly_empty_catalog(self, sto "a real sync that found zero inventory must serve an empty catalog, " "not fall back to the static demo products" ) + + +class TestMidSyncCrashRollsBackPartialWrites: + """A sync that crashes partway through persisting items must leave + storage exactly as the last successful sync left it (AI-6 follow-up). + + Defect: ``persist_synced_product``/``_upsert_synced_package`` write + straight to storage one item at a time inside the loop. An exception + partway through left every item already written in THIS run + permanently in storage, even though the sync is reported as failed + and the code's own docstring claims a failure "never silently + changes" the synced layer. Found live: synced A+B cleanly, then a + sync that persists C and crashes before D: A and B survived as + expected, but C leaked in anyway. + """ + + async def test_item_persisted_before_a_later_crash_is_rolled_back(self, storage): + settings = _flow_settings() + client_a = _FakeAdServerClient( + [_FakeInventoryItem("A", "Product A"), _FakeInventoryItem("B", "Product B")] + ) + await _run_sync(storage, settings, client_a) + + client_b = _FakeAdServerClient( + [ + _FakeInventoryItem("C", "Product C"), + _FakeInventoryItem("D", "Product D"), + _FakeInventoryItem("E", "Product E"), + ] + ) + real_persist = catalog_service.persist_synced_product + + async def flaky_persist(product): + if product.product_id == "D": + raise RuntimeError("simulated crash persisting D") + await real_persist(product) + + with patch("ad_seller.services.catalog_service.persist_synced_product", flaky_persist): + await _run_sync(storage, settings, client_b) + + assert await storage.exists("synced_product:A") + assert await storage.exists("synced_product:B") + assert not await storage.exists("synced_product:C"), ( + "C was persisted by the crashed run before D failed -- it must be " + "rolled back, not left behind despite the sync being reported as failed" + ) + assert not await storage.exists("synced_product:D") + assert not await storage.exists("synced_product:E") + + async def test_resynced_existing_item_restores_its_prior_value_on_rollback(self, storage): + """Rollback must restore, not blindly delete, an id that already + existed -- a re-synced item can share an id with one a previous + successful sync already persisted.""" + settings = _flow_settings() + client_a = _FakeAdServerClient( + [_FakeInventoryItem("A", "Product A", {"floor_price_cpm": 8.0})] + ) + await _run_sync(storage, settings, client_a) + original = await storage.get("synced_product:A") + assert original["floor_cpm"] == pytest.approx(8.0 * 0.85) + + client_b = _FakeAdServerClient( + [ + _FakeInventoryItem("A", "Product A", {"floor_price_cpm": 99.0}), + _FakeInventoryItem("D", "Product D"), + ] + ) + real_persist = catalog_service.persist_synced_product + + async def flaky_persist(product): + if product.product_id == "D": + raise RuntimeError("simulated crash persisting D") + await real_persist(product) + + with patch("ad_seller.services.catalog_service.persist_synced_product", flaky_persist): + await _run_sync(storage, settings, client_b) + + restored = await storage.get("synced_product:A") + assert restored == original, ( + "A existed before this crashed run and must be restored to its " + "prior value, not deleted or left with the crashed run's new value" + ) + assert not await storage.exists("synced_product:D") From 8c1bfa6530c5e49eded2ffdc28f40bf85d22a999 Mon Sep 17 00:00:00 2001 From: Sirajmx Date: Wed, 28 Oct 2026 11:56:19 +0530 Subject: [PATCH 5/9] docs: correct _finish_sync docstring and stale media-kit catalog comment (AI-6) --- src/ad_seller/flows/product_setup_flow.py | 21 +++++++++++-------- .../interfaces/api/routers/media_kit.py | 5 +++-- 2 files changed, 15 insertions(+), 11 deletions(-) diff --git a/src/ad_seller/flows/product_setup_flow.py b/src/ad_seller/flows/product_setup_flow.py index 45633bad..ea2a4b5d 100644 --- a/src/ad_seller/flows/product_setup_flow.py +++ b/src/ad_seller/flows/product_setup_flow.py @@ -280,15 +280,18 @@ async def sync_from_ad_server(self) -> None: await self._finish_sync(real_sync_completed=True) async def _finish_sync(self, real_sync_completed: bool = False) -> None: - """Terminal step of every sync path: prune the stale synced layer. - - Runs unconditionally, including on the mock/fallback path (no ad - server configured, or the real sync raised) -- mirroring - ``_prune_stale_synced_packages``'s own convergence guarantee - (issue #34): a transient ad-server failure reverts the catalog to - the honest static-default fallback rather than serving - increasingly stale synced products forever, and the next - successful sync immediately re-populates it. + """Terminal step of a sync that reflects current state: prune the stale synced layer. + + Runs on exactly two paths: the mock path (no ad server configured, + so mocks are the honest steady state) and the real ad-server + success path. It deliberately does NOT run when the real sync + raises: ``sync_from_ad_server`` returns before reaching it, because + a failed run leaves ``self.state.synced_segments`` / + ``self.state.products`` empty and the prune would then delete every + real package and product from the last successful sync. A transient + failure therefore keeps serving the last-good synced catalog, and + the next successful sync converges it (issue #34). Do not "fix" + the failure path to call this. ``real_sync_completed`` is only ever True from the real ad-server success path, never the mock path -- a zero-item real sync is a diff --git a/src/ad_seller/interfaces/api/routers/media_kit.py b/src/ad_seller/interfaces/api/routers/media_kit.py index 247f677c..b24610bc 100644 --- a/src/ad_seller/interfaces/api/routers/media_kit.py +++ b/src/ad_seller/interfaces/api/routers/media_kit.py @@ -173,8 +173,9 @@ async def search_media_kit( async def _resolve_package_products(product_ids: list[str], storage): """Resolve product ids to ProductDefinitions, catalog-first. - ``GET /products`` serves the in-memory static catalog - (``deps.get_product_catalog()``), which is never persisted to storage — + ``GET /products`` serves ``deps.get_product_catalog()``: the static or + CSV catalog, or (since AI-6) the synced ad-server catalog stored under + ``synced_product:*`` keys, none of which ``storage.get_product`` reads — so package endpoints must resolve ids against that same catalog first, falling back to ``storage.get_product`` for products persisted by the negotiation flow (issue #34: storage-only resolution silently produced From 712ee9252bd86da0ee10f15c430b686db078c61d Mon Sep 17 00:00:00 2001 From: Sirajmx Date: Wed, 28 Oct 2026 16:11:25 +0530 Subject: [PATCH 6/9] fix: page through every GAM ad unit on sync instead of only the first 100 (AI-6) GAMAdServerClient.list_inventory read one page and dropped nextPageToken, so a sync saw at most 100 ad units and the prune deleted the rest. It now follows the token until it runs out or limit units are collected; limit=None reads every page. The sync requests limit=None, which also lifts the same 100-row cap on S3. --- src/ad_seller/clients/ad_server_base.py | 8 ++- src/ad_seller/clients/csv_adapter.py | 2 +- src/ad_seller/clients/freewheel_adapter.py | 2 +- src/ad_seller/clients/gam_adapter.py | 26 +++++-- src/ad_seller/clients/s3_csv_adapter.py | 2 +- src/ad_seller/flows/product_setup_flow.py | 5 +- tests/unit/test_gam_clients.py | 70 +++++++++++++++++++ tests/unit/test_inventory_sync_persistence.py | 6 +- tests/unit/test_issue34_catalog_fixes.py | 2 +- tests/unit/test_synced_product_catalog.py | 13 +++- 10 files changed, 120 insertions(+), 16 deletions(-) diff --git a/src/ad_seller/clients/ad_server_base.py b/src/ad_seller/clients/ad_server_base.py index 81c5dbe7..b07cf0e4 100644 --- a/src/ad_seller/clients/ad_server_base.py +++ b/src/ad_seller/clients/ad_server_base.py @@ -258,10 +258,14 @@ async def update_deal( async def list_inventory( self, *, - limit: int = 100, + limit: Optional[int] = 100, filter_str: Optional[str] = None, ) -> list[AdServerInventoryItem]: - """List ad units (GAM) or ad slots (FreeWheel).""" + """List ad units (GAM) or ad slots (FreeWheel). + + Returns at most ``limit`` items; ``limit=None`` returns all of them + (a full sync needs the whole inventory, not the first page). + """ # -- Audience Operations -- diff --git a/src/ad_seller/clients/csv_adapter.py b/src/ad_seller/clients/csv_adapter.py index 2b0060e8..45d5663c 100644 --- a/src/ad_seller/clients/csv_adapter.py +++ b/src/ad_seller/clients/csv_adapter.py @@ -262,7 +262,7 @@ async def disconnect(self) -> None: async def list_inventory( self, *, - limit: int = 100, + limit: Optional[int] = 100, filter_str: Optional[str] = None, ) -> list[AdServerInventoryItem]: """Read inventory.csv and return AdServerInventoryItem list.""" diff --git a/src/ad_seller/clients/freewheel_adapter.py b/src/ad_seller/clients/freewheel_adapter.py index 23a8e284..ca175afe 100644 --- a/src/ad_seller/clients/freewheel_adapter.py +++ b/src/ad_seller/clients/freewheel_adapter.py @@ -158,7 +158,7 @@ async def _reconnect_bc(self) -> None: async def list_inventory( self, *, - limit: int = 100, + limit: Optional[int] = 100, filter_str: Optional[str] = None, ) -> list[AdServerInventoryItem]: """List inventory from FreeWheel Streaming Hub. diff --git a/src/ad_seller/clients/gam_adapter.py b/src/ad_seller/clients/gam_adapter.py index f1b491d6..437a48c1 100644 --- a/src/ad_seller/clients/gam_adapter.py +++ b/src/ad_seller/clients/gam_adapter.py @@ -322,13 +322,29 @@ async def update_deal( async def list_inventory( self, *, - limit: int = 100, + limit: Optional[int] = 100, filter_str: Optional[str] = None, ) -> list[AdServerInventoryItem]: - ad_units, _ = await self._rest.list_ad_units( - page_size=limit, - filter_str=filter_str, - ) + # The REST API caps a page at 100 ad units, so follow + # nextPageToken until it runs out or ``limit`` units are collected + # (``limit=None`` reads every page). Reading only the first page made + # a full sync see at most 100 units and prune the rest. + ad_units = [] + page_token: Optional[str] = None + seen_tokens: set[str] = set() + while limit is None or len(ad_units) < limit: + page_size = 100 if limit is None else min(100, limit - len(ad_units)) + page, page_token = await self._rest.list_ad_units( + page_size=page_size, + page_token=page_token, + filter_str=filter_str, + ) + ad_units.extend(page) + if not page or not page_token or page_token in seen_tokens: + break + seen_tokens.add(page_token) + if limit is not None: + ad_units = ad_units[:limit] return [ AdServerInventoryItem( diff --git a/src/ad_seller/clients/s3_csv_adapter.py b/src/ad_seller/clients/s3_csv_adapter.py index f0f0bb4f..ece8a944 100644 --- a/src/ad_seller/clients/s3_csv_adapter.py +++ b/src/ad_seller/clients/s3_csv_adapter.py @@ -179,7 +179,7 @@ async def disconnect(self) -> None: async def list_inventory( self, *, - limit: int = 100, + limit: Optional[int] = 100, filter_str: Optional[str] = None, ) -> list[AdServerInventoryItem]: """List inventory items from S3 CSV files.""" diff --git a/src/ad_seller/flows/product_setup_flow.py b/src/ad_seller/flows/product_setup_flow.py index ea2a4b5d..9b30f105 100644 --- a/src/ad_seller/flows/product_setup_flow.py +++ b/src/ad_seller/flows/product_setup_flow.py @@ -163,8 +163,11 @@ async def sync_from_ad_server(self) -> None: # servers (GAM/FreeWheel), which is also where excluding # archived inventory by default actually matters. filter_str = "status:ACTIVE" if client.ad_server_type.value != "csv" else None + # limit=None: a sync must see the whole inventory. The default + # (100) silently truncated larger GAM networks and S3 files, and + # the prune below then deleted everything past the cut. async with client: - items = await client.list_inventory(filter_str=filter_str) + items = await client.list_inventory(filter_str=filter_str, limit=None) # Group items by inferred type and create packages grouped: dict[str, list] = {} diff --git a/tests/unit/test_gam_clients.py b/tests/unit/test_gam_clients.py index 4da99a24..d0a34ce7 100644 --- a/tests/unit/test_gam_clients.py +++ b/tests/unit/test_gam_clients.py @@ -339,3 +339,73 @@ def test_cpv_not_supported(self): """Test CPV is not a valid GAM cost type.""" valid_types = [e.value for e in GAMCostType] assert "CPV" not in valid_types + + +class TestGAMAdapterListInventoryPaging: + """GAMAdServerClient.list_inventory must follow nextPageToken. + + The REST API returns at most 100 ad units per page. Reading only the + first page made a sync of a larger network see 100 units and prune the + rest. + """ + + @staticmethod + def _adapter(total_units: int, repeat_token: bool = False): + from types import SimpleNamespace + + from ad_seller.clients.gam_adapter import GAMAdServerClient + + units = [ + SimpleNamespace( + id=f"u{i}", name=f"Unit {i}", parent_id=None, status="ACTIVE", ad_unit_sizes=[] + ) + for i in range(total_units) + ] + calls: list[dict] = [] + + async def list_ad_units(page_size=100, page_token=None, filter_str=None): + calls.append({"page_size": page_size, "page_token": page_token}) + start = int(page_token or 0) + end = start + min(page_size, 100) + if repeat_token: + return units[start:end], "0" + return units[start:end], str(end) if end < len(units) else None + + with ( + patch("ad_seller.clients.gam_adapter.GAMSoapClient"), + patch("ad_seller.clients.gam_adapter.GAMRestClient"), + ): + adapter = GAMAdServerClient() + adapter._rest.list_ad_units = list_ad_units + return adapter, calls + + async def test_limit_none_reads_every_page(self): + adapter, calls = self._adapter(250) + + items = await adapter.list_inventory(limit=None, filter_str="status:ACTIVE") + + assert [i.id for i in items] == [f"u{i}" for i in range(250)] + assert [c["page_token"] for c in calls] == [None, "100", "200"] + + async def test_limit_caps_across_pages(self): + adapter, calls = self._adapter(250) + + items = await adapter.list_inventory(limit=120) + + assert len(items) == 120 + assert [c["page_size"] for c in calls] == [100, 20] + + async def test_default_limit_still_returns_one_page(self): + adapter, calls = self._adapter(250) + + items = await adapter.list_inventory() + + assert len(items) == 100 + assert len(calls) == 1 + + async def test_repeated_page_token_stops_instead_of_looping(self): + adapter, calls = self._adapter(250, repeat_token=True) + + await adapter.list_inventory(limit=None) + + assert len(calls) == 2 diff --git a/tests/unit/test_inventory_sync_persistence.py b/tests/unit/test_inventory_sync_persistence.py index 7bd82eca..6df4545b 100644 --- a/tests/unit/test_inventory_sync_persistence.py +++ b/tests/unit/test_inventory_sync_persistence.py @@ -85,7 +85,7 @@ async def __aenter__(self): async def __aexit__(self, *exc): return False - async def list_inventory(self, filter_str=None): + async def list_inventory(self, filter_str=None, limit=None): return self._items @@ -252,7 +252,7 @@ async def __aenter__(self): async def __aexit__(self, *exc): return False - async def list_inventory(self, filter_str=None): + async def list_inventory(self, filter_str=None, limit=None): self.received_filter_str = filter_str return self._items @@ -275,7 +275,7 @@ async def __aenter__(self): async def __aexit__(self, *exc): return False - async def list_inventory(self, filter_str=None): + async def list_inventory(self, filter_str=None, limit=None): raise self._error diff --git a/tests/unit/test_issue34_catalog_fixes.py b/tests/unit/test_issue34_catalog_fixes.py index 85c60dce..4147446c 100644 --- a/tests/unit/test_issue34_catalog_fixes.py +++ b/tests/unit/test_issue34_catalog_fixes.py @@ -358,7 +358,7 @@ async def __aenter__(self): async def __aexit__(self, *exc): return False - async def list_inventory(self, filter_str=None): + async def list_inventory(self, filter_str=None, limit=None): return self._items diff --git a/tests/unit/test_synced_product_catalog.py b/tests/unit/test_synced_product_catalog.py index 3920c560..a682d809 100644 --- a/tests/unit/test_synced_product_catalog.py +++ b/tests/unit/test_synced_product_catalog.py @@ -276,6 +276,7 @@ class _FakeAdServerClient: def __init__(self, items): self._items = items + self.limit_requested = "not called" async def __aenter__(self): return self @@ -283,7 +284,9 @@ async def __aenter__(self): async def __aexit__(self, *exc): return False - async def list_inventory(self, filter_str=None): + # Same default as the real adapters, so a caller that omits limit is caught. + async def list_inventory(self, filter_str=None, limit=100): + self.limit_requested = limit return self._items @@ -315,6 +318,14 @@ async def _run_sync(storage, settings, client) -> None: class TestSyncPersistsAndPrunesProducts: + async def test_sync_requests_the_whole_inventory(self, storage): + """The adapters' default limit is 100; a sync that inherited it + would see only the first 100 items and prune the rest.""" + client = _FakeAdServerClient([_FakeInventoryItem("gam-001", "Homepage Display Banner")]) + await _run_sync(storage, _flow_settings(), client) + + assert client.limit_requested is None + async def test_sync_persists_products_to_storage(self, storage): client = _FakeAdServerClient( [ From 1da6d9fb650bf67f7a933f857bc883fc73916a9d Mon Sep 17 00:00:00 2001 From: Sirajmx Date: Wed, 28 Oct 2026 16:11:25 +0530 Subject: [PATCH 7/9] fix: roll back this run's products when the packages loop fails (AI-6) The products loop commits before the packages loop runs, and only the products loop's own failure rolled products back. A crash in the packages loop left this run's products stored while the sync reported failure. Both loops now share the product rollback. --- src/ad_seller/flows/product_setup_flow.py | 21 +++++++++----- tests/unit/test_synced_product_catalog.py | 35 +++++++++++++++++++++++ 2 files changed, 49 insertions(+), 7 deletions(-) diff --git a/src/ad_seller/flows/product_setup_flow.py b/src/ad_seller/flows/product_setup_flow.py index 9b30f105..9421b5b4 100644 --- a/src/ad_seller/flows/product_setup_flow.py +++ b/src/ad_seller/flows/product_setup_flow.py @@ -197,6 +197,16 @@ async def sync_from_ad_server(self) -> None: # persisted by an earlier successful sync; rollback must # restore that old value, not erase still-valid data. persisted: list[tuple[str, Optional[dict[str, Any]]]] = [] + + async def _rollback_products() -> None: + for product_id, prior in persisted: + key = f"synced_product:{product_id}" + if prior is None: + await storage.delete(key) + else: + await storage.set(key, prior) + self.state.products.pop(product_id, None) + try: for item in items: product_def = product_from_inventory_item(item) @@ -209,13 +219,7 @@ async def sync_from_ad_server(self) -> None: await persist_synced_product(product_def) persisted.append((product_def.product_id, prior)) except Exception: - for product_id, prior in persisted: - key = f"synced_product:{product_id}" - if prior is None: - await storage.delete(key) - else: - await storage.set(key, prior) - self.state.products.pop(product_id, None) + await _rollback_products() raise logger.info("Created %d products from ad server inventory", len(self.state.products)) @@ -265,6 +269,9 @@ async def sync_from_ad_server(self) -> None: await storage.set_package(package_id, prior_package) if package_id in self.state.synced_segments: self.state.synced_segments.remove(package_id) + # The products loop already committed this run's products; + # a failed run must not leave them behind either. + await _rollback_products() raise logger.info("Synced %d packages from ad server", len(grouped)) diff --git a/tests/unit/test_synced_product_catalog.py b/tests/unit/test_synced_product_catalog.py index a682d809..1563c5e2 100644 --- a/tests/unit/test_synced_product_catalog.py +++ b/tests/unit/test_synced_product_catalog.py @@ -483,3 +483,38 @@ async def flaky_persist(product): "prior value, not deleted or left with the crashed run's new value" ) assert not await storage.exists("synced_product:D") + + async def test_packages_loop_crash_rolls_back_this_runs_products(self, storage): + """The products loop commits before the packages loop runs, so a + crash in the packages loop must roll back this run's products too, + not only its packages.""" + settings = _flow_settings() + client_a = _FakeAdServerClient( + [_FakeInventoryItem("A", "Product A", {"floor_price_cpm": 8.0})] + ) + await _run_sync(storage, settings, client_a) + original = await storage.get("synced_product:A") + + client_b = _FakeAdServerClient( + [ + _FakeInventoryItem("A", "Product A", {"floor_price_cpm": 99.0}), + _FakeInventoryItem("C", "Product C"), + ] + ) + + async def crashing_upsert(self, storage, package): + raise RuntimeError("simulated crash upserting a package") + + with patch( + "ad_seller.flows.product_setup_flow.ProductSetupFlow._upsert_synced_package", + crashing_upsert, + ): + await _run_sync(storage, settings, client_b) + + assert await storage.get("synced_product:A") == original, ( + "A must be restored to the last successful sync's value" + ) + assert not await storage.exists("synced_product:C"), ( + "C was persisted by the products loop of the failed run -- it must be " + "rolled back when the packages loop fails" + ) From c49d8b30df08cda9d67c658f8b3a1323a0fb50be Mon Sep 17 00:00:00 2001 From: Sirajmx Date: Wed, 28 Oct 2026 16:31:13 +0530 Subject: [PATCH 8/9] fix: let the CSV adapter return every row when limit is None (AI-6) The sync now calls list_inventory(limit=None), but the CSV adapter compared len(results) >= limit unconditionally, so every CSV-mode sync failed with a TypeError. Guard the comparison, and add a sync test through the real CSVAdServerClient so a flow/adapter limit mismatch can't hide behind the test fakes again. --- src/ad_seller/clients/csv_adapter.py | 2 +- tests/unit/test_csv_adapter.py | 6 +++++ tests/unit/test_synced_product_catalog.py | 33 +++++++++++++++++++++++ 3 files changed, 40 insertions(+), 1 deletion(-) diff --git a/src/ad_seller/clients/csv_adapter.py b/src/ad_seller/clients/csv_adapter.py index 45d5663c..636f9399 100644 --- a/src/ad_seller/clients/csv_adapter.py +++ b/src/ad_seller/clients/csv_adapter.py @@ -306,7 +306,7 @@ async def list_inventory( item.__dict__["raw"] = raw results.append(item) - if len(results) >= limit: + if limit is not None and len(results) >= limit: break return results diff --git a/tests/unit/test_csv_adapter.py b/tests/unit/test_csv_adapter.py index 38d63198..c18fc106 100644 --- a/tests/unit/test_csv_adapter.py +++ b/tests/unit/test_csv_adapter.py @@ -244,6 +244,12 @@ async def test_applies_limit(self, sample_client: CSVAdServerClient) -> None: items = await sample_client.list_inventory(limit=3) assert len(items) == 3 + async def test_limit_none_returns_every_row(self, sample_client: CSVAdServerClient) -> None: + """limit=None (what a full sync passes) returns all rows.""" + everything = await sample_client.list_inventory(limit=10_000) + items = await sample_client.list_inventory(limit=None) + assert len(items) == len(everything) > 3 + async def test_applies_filter_str(self, sample_client: CSVAdServerClient) -> None: """filter_str filters by substring match on name (case-insensitive).""" items = await sample_client.list_inventory(filter_str="sports") diff --git a/tests/unit/test_synced_product_catalog.py b/tests/unit/test_synced_product_catalog.py index 1563c5e2..746328df 100644 --- a/tests/unit/test_synced_product_catalog.py +++ b/tests/unit/test_synced_product_catalog.py @@ -518,3 +518,36 @@ async def crashing_upsert(self, storage, package): "C was persisted by the products loop of the failed run -- it must be " "rolled back when the packages loop fails" ) + + +class TestSyncWithRealCsvAdapter: + """The flow tests above use _FakeAdServerClient, which accepts any + ``limit``. This runs the sync through the real CSVAdServerClient so a + flow/adapter contract mismatch (the flow passing ``limit=None``) fails + here instead of only on a live server.""" + + async def test_csv_sync_completes_and_persists_every_row(self, storage, tmp_path): + import shutil + from pathlib import Path + + from ad_seller.clients.csv_adapter import CSVAdServerClient + + sample = Path(__file__).parents[2] / "data" / "csv" / "samples" / "ctv_streaming" + shutil.copy(sample / "inventory.csv", tmp_path / "inventory.csv") + client = CSVAdServerClient(data_dir=str(tmp_path)) + async with client: + rows = await client.list_inventory(limit=10_000) + + settings = _flow_settings(gam_network_code=None, ad_server_type="csv") + with ( + patch("ad_seller.flows.product_setup_flow.get_settings", return_value=settings), + patch("ad_seller.storage.factory.get_storage", return_value=storage), + patch("ad_seller.clients.ad_server_base.get_ad_server_client", lambda: client), + ): + from ad_seller.flows.product_setup_flow import ProductSetupFlow + + flow = ProductSetupFlow() + await flow.sync_from_ad_server() + + assert not flow.state.ad_server_sync_failed, flow.state.warnings + assert len(await storage.keys("synced_product:*")) == len(rows) From 67be66c7c17fecc41ec5bbd6879bda479b378d9c Mon Sep 17 00:00:00 2001 From: Sirajmx Date: Wed, 28 Oct 2026 16:45:26 +0530 Subject: [PATCH 9/9] fix: serve every CSV row in the CSV-mode catalog, not the first 100 (AI-6) build_csv_product_catalog called list_inventory() with the adapter's default limit of 100, so a CSV file with more rows served only the first 100: later products 404'd on GET /products/{id} and quotes, while the sync (which now reads every row) still built packages pointing at them. Request limit=None so the catalog and the sync see the same inventory. --- src/ad_seller/services/catalog_service.py | 5 ++++- tests/unit/test_csv_catalog_coherence.py | 16 ++++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/src/ad_seller/services/catalog_service.py b/src/ad_seller/services/catalog_service.py index 19a0f35e..4a0f8630 100644 --- a/src/ad_seller/services/catalog_service.py +++ b/src/ad_seller/services/catalog_service.py @@ -364,10 +364,13 @@ def build_csv_product_catalog() -> dict[str, Any]: data_dir = get_settings().csv_data_dir + # limit=None: serve every row. The adapter default (100) silently + # dropped the rest, while the sync (which reads every row) still built + # packages that pointed at them. async def _load() -> list[Any]: client = CSVAdServerClient(data_dir=data_dir) async with client: - return await client.list_inventory() + return await client.list_inventory(limit=None) try: items = _run_blocking(_load()) diff --git a/tests/unit/test_csv_catalog_coherence.py b/tests/unit/test_csv_catalog_coherence.py index 37b9e98e..0bc2bf17 100644 --- a/tests/unit/test_csv_catalog_coherence.py +++ b/tests/unit/test_csv_catalog_coherence.py @@ -203,6 +203,22 @@ async def test_csv_catalog_ids_deterministic_across_rebuild(self, csv_mode): second_ids = set((await catalog_service.get_static_product_catalog())["products"]) assert first_ids == second_ids == CSV_IDS + async def test_csv_catalog_serves_every_row_past_the_adapter_default(self, csv_mode): + """The adapter's default limit is 100; the catalog must not inherit + it, or rows past 100 404 while the sync still packages them.""" + rows = [ + f"inv-bulk-{i:03d},Bulk Display {i:03d},,ACTIVE,300x250,banner,2,display," + "IAB1,10.00,USD,US,Bulk row" + for i in range(150) + ] + (csv_mode / "inventory.csv").write_text(CSV_HEADER + "\n" + "\n".join(rows) + "\n") + _reset_caches() + + catalog = await catalog_service.get_static_product_catalog() + + assert len(catalog["products"]) == 150 + assert "inv-bulk-149" in catalog["products"] + # ============================================================================= # Non-CSV modes pinned byte-identical