diff --git a/CHANGELOG.md b/CHANGELOG.md index eb2ec09f..9f730aea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,13 @@ All notable changes to the IAB Tech Lab Seller Agent are documented here. ### Changed +- Index Exchange SSP connector rewritten against the real `/v3/deals` API + (create, get, list, update, clone, troubleshoot, seat-ID to DSP + resolution). Seat IDs that resolve to several DSPs pause for operator + approval instead of guessing; approvals, their listing and resume are + operator-only, and a resumed approval is consumed so it cannot create the + deal twice. `INDEX_EXCHANGE_API_URL` has no default, so an unconfigured + deployment never calls production. - Booking (POST /api/v1/deals) now requires a verified buyer key matching the quote; the MCP deal-from-template tool is operator-gated. Previously anonymous callers could book any quote. diff --git a/docs/api/openapi.json b/docs/api/openapi.json index 4711c219..ea50d335 100644 --- a/docs/api/openapi.json +++ b/docs/api/openapi.json @@ -2875,7 +2875,7 @@ "Approvals" ], "summary": "List Pending Approvals", - "description": "List all pending approval requests. Requires authentication.", + "description": "List all pending approval requests. Requires an operator credential.", "operationId": "list_pending_approvals_approvals_get", "parameters": [ { @@ -2939,7 +2939,7 @@ "Approvals" ], "summary": "Get Approval", - "description": "Get a specific approval request and its response (if any).\n\nRequires authentication.", + "description": "Get a specific approval request and its response (if any).\n\nRequires an operator credential.", "operationId": "get_approval_approvals__approval_id__get", "parameters": [ { @@ -3012,7 +3012,7 @@ "Approvals" ], "summary": "Decide Approval", - "description": "Submit a human decision for a pending approval. Requires authentication.\n\nThe verified principal from the authenticated API key is stamped into\nthe audit record; ``body.decided_by`` is kept as a display label only.", + "description": "Submit a human decision for a pending approval. Requires an operator credential.\n\nThe verified principal from the authenticated API key is stamped into\nthe audit record; ``body.decided_by`` is kept as a display label only.", "operationId": "decide_approval_approvals__approval_id__decide_post", "parameters": [ { @@ -3095,7 +3095,7 @@ "Approvals" ], "summary": "Resume Flow", - "description": "Resume a flow after an approval decision has been submitted.\n\nRequires authentication. Loads the flow state snapshot, applies the\ndecision, and returns the final result without re-running expensive\ncrew evaluations.", + "description": "Resume a flow after an approval decision has been submitted.\n\nRequires an operator credential. Loads the flow state snapshot, applies the\ndecision, and returns the final result without re-running expensive\ncrew evaluations.", "operationId": "resume_flow_approvals__approval_id__resume_post", "parameters": [ { @@ -10424,6 +10424,13 @@ "additionalProperties": true, "type": "object" }, + { + "items": { + "additionalProperties": true, + "type": "object" + }, + "type": "array" + }, { "type": "null" } diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md index 5b271e56..74656931 100644 --- a/docs/guides/configuration.md +++ b/docs/guides/configuration.md @@ -68,8 +68,8 @@ ad-seller freewheel-login --provider bc | `SSP_ROUTING_RULES` | `str` | `""` | Routing rules: `ctv:pubmatic,display:index_exchange` | | `PUBMATIC_MCP_URL` | `str` | `None` | PubMatic MCP endpoint (e.g., `https://mcp.pubmatic.com/sses`) | | `PUBMATIC_API_KEY` | `str` | `None` | PubMatic API key | -| `INDEX_EXCHANGE_API_URL` | `str` | `None` | Index Exchange REST API URL | -| `INDEX_EXCHANGE_API_KEY` | `str` | `None` | Index Exchange API key | +| `INDEX_EXCHANGE_API_URL` | `str` | `None` | Index Exchange REST API FQDN only (no path), e.g. `https://app.indexexchange.com` — the connector applies path prefixes per-endpoint | +| `INDEX_EXCHANGE_API_KEY` | `str` | `None` | Keycloak JWT bearer token (not a static API key — expires and must be refreshed) | | `MAGNITE_API_URL` | `str` | `None` | Magnite REST API URL | | `MAGNITE_API_KEY` | `str` | `None` | Magnite API key | diff --git a/docs/guides/developer-setup.md b/docs/guides/developer-setup.md index 6f343808..6904dc7d 100644 --- a/docs/guides/developer-setup.md +++ b/docs/guides/developer-setup.md @@ -78,9 +78,9 @@ SSP_CONNECTORS=pubmatic,index_exchange PUBMATIC_MCP_URL=https://mcp.pubmatic.com/sses PUBMATIC_API_KEY=your-pubmatic-key -# Index Exchange (REST) -INDEX_EXCHANGE_API_URL=https://api.indexexchange.com -INDEX_EXCHANGE_API_KEY=your-ix-key +# Index Exchange (REST) — FQDN only, no path; the connector applies /api/deals itself +INDEX_EXCHANGE_API_URL=https://app.indexexchange.com +INDEX_EXCHANGE_API_KEY=your-keycloak-jwt-bearer-token # Routing rules (optional) SSP_ROUTING_RULES=ctv:pubmatic,display:index_exchange diff --git a/docs/guides/troubleshooting.md b/docs/guides/troubleshooting.md index eb60b5eb..a686328f 100644 --- a/docs/guides/troubleshooting.md +++ b/docs/guides/troubleshooting.md @@ -110,19 +110,40 @@ The deal was created in the seller agent but not yet pushed to PubMatic. ### Index Exchange API Errors +**Error: `Index Exchange: 401 Unauthorized`** + +`INDEX_EXCHANGE_API_KEY` must be a valid Keycloak JWT bearer token, not a +static API key — it expires and must be refreshed via the client-credentials +grant. This is not automated yet; re-issue a fresh token in your `.env`. + **Error: `Index Exchange: 403 Forbidden`** -Your API key doesn't have permission for that operation. +Your token doesn't have permission for that operation. + +1. Verify `INDEX_EXCHANGE_API_KEY` in your `.env` is current (not expired). +2. Contact your Index Exchange account manager to confirm the token has deal management scope. + +**Error: `ValueError: external_deal_id is required` / `account_id is required` / `dsp_id is required`** -1. Verify `INDEX_EXCHANGE_API_KEY` in your `.env`. -2. Contact your Index Exchange account manager to confirm the key has deal management scope. +`/v3/deals` requires `externalDealID`, `account.accountID`, and `dspID` +(`directConfigurations.dspID` for Direct deal types, or +`marketplaceConfigurations.dspID` for Marketplace Package) on every create +call. `external_deal_id` is supplied automatically by the seller-agent's +deal flow. `dsp_id` can be resolved automatically from `buyer_seat_ids` (set +at deal-creation time, e.g. via `create_curated_deal`'s `buyer_seat_ids` +param) — if the seat ID(s) match more than one DSP, the deal is held +pending human approval instead of guessing (see `approve_or_reject` / +`resume_approval`). `account_id` has no automatic source yet; pass it +explicitly via `distribute_deal_via_ssp`'s `account_id` param if it isn't +already on the stored deal. **Error: `Index Exchange: 422 Unprocessable Entity`** Deal data failed validation on the Index Exchange side. -- Check the response body for a `errors` field — it will name the invalid field. -- Common causes: floor price below IX minimum, unsupported ad format, missing `seller_id`. +- Check the response body for an `errors` field — it will name the invalid field. +- Common causes: `floor` below IX minimum, `externalDealID` starting with `0` + or outside 3–64 chars, missing `directConfigurations.dspID` on a Direct deal. --- diff --git a/docs/reference/mcp-tools.md b/docs/reference/mcp-tools.md index a4899f37..4a5d2a44 100644 --- a/docs/reference/mcp-tools.md +++ b/docs/reference/mcp-tools.md @@ -6,7 +6,7 @@ Tools exposed by the seller MCP server (`ad_seller.interfaces.mcp_server`), enumerated from the live FastMCP registry. -**Total: 47 tools** +**Total: 48 tools** | Tool | Description | | --- | --- | @@ -49,6 +49,7 @@ enumerated from the live FastMCP registry. | `push_deal_to_buyers` | Push a deal to buyer endpoints via IAB Deals API v1.0. | | `register_buyer_agent` | Discover and register a buyer agent by URL. | | `request_quote` | Request a non-binding price quote for a product. | +| `resume_approval` | Resume a flow that was paused for approval, after a decision has been | | `revoke_api_key` | Revoke an API key. | | `set_agent_trust` | Set trust level for a buyer agent. | | `set_approval_gates` | Configure approval gates. required_flows is comma-separated: | diff --git a/src/ad_seller/clients/ssp_base.py b/src/ad_seller/clients/ssp_base.py index cffb5123..c91465ec 100644 --- a/src/ad_seller/clients/ssp_base.py +++ b/src/ad_seller/clients/ssp_base.py @@ -25,6 +25,11 @@ from pydantic import BaseModel + +class SSPDealOutcomeUnknownError(Exception): + """create_deal may or may not have created the deal and this could not be confirmed.""" + + # ============================================================================= # SSP result models # ============================================================================= @@ -76,7 +81,9 @@ class SSPDeal(BaseModel): priority: Optional[str] = None # SSP-specific (e.g., PubMatic P11-P15) start_date: Optional[str] = None end_date: Optional[str] = None - targeting: Optional[dict[str, Any]] = None + # Shape varies by SSP: a flat dict (e.g. PubMatic) or a structured array of + # targeting objects (e.g. Index Exchange's {targetingType, keyName, sets}). + targeting: Optional[Any] = None impressions_goal: Optional[int] = None ssp_type: SSPType = SSPType.CUSTOM ssp_name: Optional[str] = None # Human-readable SSP name @@ -107,11 +114,17 @@ class SSPDealCreateRequest(BaseModel): priority: Optional[str] = None start_date: Optional[str] = None end_date: Optional[str] = None - targeting: Optional[dict[str, Any]] = None + # Shape varies by SSP: a flat dict (e.g. PubMatic) or a structured array of + # targeting objects (e.g. Index Exchange's {targetingType, keyName, sets}). + targeting: Optional[dict[str, Any] | list[dict[str, Any]]] = None impressions_goal: Optional[int] = None buyer_seat_ids: list[str] = [] # For clone operations source_deal_id: Optional[str] = None + # SSP-assigned identifiers some SSPs (e.g. Index Exchange) require at creation time + external_deal_id: Optional[str] = None + account_id: Optional[int] = None + dsp_id: Optional[int] = None # ============================================================================= diff --git a/src/ad_seller/clients/ssp_index_exchange.py b/src/ad_seller/clients/ssp_index_exchange.py index ec672749..02dd9f29 100644 --- a/src/ad_seller/clients/ssp_index_exchange.py +++ b/src/ad_seller/clients/ssp_index_exchange.py @@ -3,29 +3,65 @@ """Index Exchange SSP client (REST API). -Index Exchange exposes a REST API for deal management at api.indexexchange.com. -No MCP server yet (as of March 2026), though they're part of the IAB Tech Lab -Agentic RTB Framework (ARTF) coalition. - -Key endpoints: - - POST /api/deals — Create PMP deal - - GET /api/deals — List/get deals - - GET /api/cal/v1/mappings/deals — Deal mapping (name, dates, IDs) - - Reporting API for supply, demand, deals, brand data - -Docs: https://api.indexexchange.com/reference - https://kb.indexexchange.com/publishers/pmp/deals_api.htm - -TODO(index-exchange-auth): Full API docs require IX account for detailed -request/response schemas and auth mechanism. +Index Exchange's REST API lives at the FQDN https://app.indexexchange.com. +Deal-management endpoints (this connector's concern) sit under a shared +/api/deals prefix, with /v3/deals beneath that for deal CRUD operations. +INDEX_EXCHANGE_API_URL is configured as just the FQDN — the /api/deals +prefix is applied in code (_API_BASE_PATH below) rather than baked into the +configured base URL, since /api/deals is only one of several path prefixes +under this host we expect to call as seller-agent's agentic workflows +expand. No MCP server yet (as of March 2026), though they're part of the +IAB Tech Lab Agentic RTB Framework (ARTF) coalition. + +Key endpoints (FQDN + path): + - POST /api/deals/v3/deals — Create a deal + - GET /api/deals/v3/deals — List deals + - GET /api/deals/v3/deals/{id} — Get a single deal (id = internalDealID) + - PATCH /api/deals/v3/deals/{id} — Update a deal (requires If-Match ETag) + - DELETE /api/deals/v3/deals/{id} — Soft-delete a deal + - GET /api/deals/v3/deals/reports — Deals data export + - GET /api/deals/v1/dsps/-/seats — Resolve dspID(s) from seat IDs + (see resolve_dsp_ids_for_seat_ids() below) + +There is no clone/copy endpoint — clone_deal() below implements cloning by +composing get_deal() + create_deal() instead. + +Field names and behavior here are derived directly from the Deals API (the +service implementing /v3/deals), not from public docs. + +Known limitations: + - dspID (directConfigurations.dspID for Direct deal types, or + marketplaceConfigurations.dspID for Marketplace Package) is required for + every deal type this connector supports, not just Direct. create_deal() + will raise ValueError until request.dsp_id is supplied. Callers with + only a buyer_seat_id (no dsp_id) should call + resolve_dsp_ids_for_seat_ids() first — see distribute_deal_via_ssp() in + interfaces/api/main.py for how ambiguous (>1 dspID) resolutions are + gated behind human approval before create_deal() is called. + - account.accountID is likewise not sourced anywhere upstream today — + it's deliberately not a fixed deployment-wide setting, since a single + seller-agent instance may need to create deals under more than one + Index account, so it must be supplied per-request via + SSPDealCreateRequest.account_id until callers are wired up to provide it. + - No Keycloak JWT refresh: INDEX_EXCHANGE_API_KEY is treated as a + long-lived token; production tokens expire and must be refreshed via + the client-credentials grant, which is not implemented here. + - The new Deals API `floorCurrency` field (create-only, feature-flagged) + is not supported; all deals are assumed USD. """ +import asyncio import logging +import uuid from typing import Any, Optional +import httpx +from pydantic import BaseModel + from .ssp_base import ( SSPDeal, SSPDealCreateRequest, + SSPDealOutcomeUnknownError, SSPDealStatus, SSPDealType, SSPTroubleshootResult, @@ -35,35 +71,116 @@ logger = logging.getLogger(__name__) +# Shared prefix for all deal-management endpoints under the configured FQDN. +# Kept out of the configured base URL so other Index path prefixes (e.g. a +# planned /v1/seats DSP-resolution route) can be added without requiring a +# reconfigured base URL per prefix. +_API_BASE_PATH = "/api/deals" + +# Deal CRUD path, relative to _API_BASE_PATH. +_DEALS_PATH = "/v3/deals" + +# externalDealID: 3-64 chars, letters/numbers/dashes/underscores/periods, and +# per the Deals API's validation cannot start with "0". +_EXTERNAL_DEAL_ID_MAX_LEN = 64 + + +def _generate_clone_external_deal_id(source_external_deal_id: str) -> str: + """Derive a new externalDealID for a clone, guaranteed to fit Index's 64-char limit. + + Appends a random suffix to the source deal's externalDealID (which is + already known-valid — it belongs to an existing deal) so the clone gets + a distinct ID, truncating the base as needed so the combined length + never exceeds Index's max. The source ID's leading character is preserved, + so the "cannot start with 0" rule stays satisfied. + """ + suffix = f"-clone-{uuid.uuid4().hex[:8]}" + base = source_external_deal_id[: _EXTERNAL_DEAL_ID_MAX_LEN - len(suffix)] + return f"{base}{suffix}" + + +# Bounds for the post-failure externalDealID lookup; `search` is a substring match. +_LOOKUP_PAGE_SIZE = 100 +_LOOKUP_MAX_PAGES = 5 + + +def _may_have_reached_index(exc: Exception) -> bool: + """True if a failed POST may have been processed by Index (so a retry could duplicate).""" + if isinstance(exc, httpx.HTTPStatusError): + return exc.response.status_code >= 500 + if isinstance(exc, (httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout)): + return False + return isinstance(exc, (httpx.TimeoutException, httpx.TransportError)) + + +def _ix_deal_config(deal_type: SSPDealType) -> tuple[int, str, bool]: + """Map an SSPDealType to Index Exchange's classID/auctionType/programmaticGuaranteed. + + Index classID values: 1=Direct Deal, 3=Inventory Package, 4=Marketplace + Package, 5=Deal with Marketplaces. Only 1 and 4 are reachable from the + deal types seller-agent currently models. + """ + if deal_type == SSPDealType.PG: + return 1, "fixed", True + if deal_type == SSPDealType.PMP: + return 1, "first", False + if deal_type == SSPDealType.PREFERRED: + return 1, "fixed", False + if deal_type == SSPDealType.AUCTION_PACKAGE: + return 4, "first", False + raise ValueError(f"Unsupported Index Exchange deal type: {deal_type}") -# Index Exchange deal type mapping -_IX_DEAL_TYPE_MAP = { - SSPDealType.PMP: "pmp", - SSPDealType.PG: "programmatic_guaranteed", - SSPDealType.PREFERRED: "preferred", -} _IX_STATUS_MAP = { "active": SSPDealStatus.ACTIVE, "paused": SSPDealStatus.PAUSED, "expired": SSPDealStatus.EXPIRED, - "archived": SSPDealStatus.ARCHIVED, - "pending": SSPDealStatus.CREATED, + "auto-paused": SSPDealStatus.PAUSED, +} + +# GET /v3/deals `status` filter values — Index has no equivalent for CREATED or +# ARCHIVED, so list_deals() rejects those instead of returning an unfiltered list. +_IX_LIST_STATUS_MAP = { + SSPDealStatus.ACTIVE: "active", + SSPDealStatus.PAUSED: "paused", + SSPDealStatus.EXPIRED: "expired", } +class DspSeatMatch(BaseModel): + """A single seat/DSP match from resolve_dsp_ids_for_seat_ids(). + + A given buyer_seat_id (DSP-facing, alphanumeric) can resolve to more + than one dspID — collisions are rare but not impossible, confirmed + empirically against real seat data — so callers must handle zero, one, + or multiple matches explicitly rather than assuming a 1:1 mapping. + """ + + dsp_id: int + requested_seat_id: str = "" + extended_seat_id: str = "" + buyer_name: str = "" + status: str = "" + + class IndexExchangeSSPClient(RESTSSPClient): """Index Exchange SSP client using their REST API. Extends RESTSSPClient with Index Exchange-specific: - - API path structure (/api/deals, /api/cal/v1/mappings/deals) - - Request format (JSON body with IX-specific field names) - - Response parsing (IX deal objects → normalized SSPDeal) - - Deal type mapping + - API path structure (/api/deals/v3/deals under the configured FQDN) + - Request format (JSON body with Index's actual /v3/deals field names) + - Response parsing (Index deal objects → normalized SSPDeal) + - classID/auctionType/programmaticGuaranteed deal type mapping Config: - INDEX_EXCHANGE_API_URL=https://api.indexexchange.com - INDEX_EXCHANGE_API_KEY=your-api-key + INDEX_EXCHANGE_API_URL=https://app.indexexchange.com (FQDN only — + see _API_BASE_PATH for why the /api/deals prefix isn't baked in here) + INDEX_EXCHANGE_API_KEY= + + account.accountID is not a fixed deployment-wide setting — a single + seller-agent instance may need to create deals under more than one + Index account — so it's only ever taken from SSPDealCreateRequest.account_id, + per-request, same as dsp_id. """ def __init__( @@ -81,61 +198,229 @@ def __init__( auth_prefix="Bearer", ) - # --- Override deal operations with IX-specific paths --- + # --- Override deal operations with Index-specific paths --- async def create_deal(self, request: SSPDealCreateRequest) -> SSPDeal: - """Create a PMP deal on Index Exchange.""" + """Create a deal on Index Exchange via POST /v3/deals.""" http = self._ensure_connected() + if not request.external_deal_id: + raise ValueError( + "Index Exchange requires external_deal_id (SSPDealCreateRequest.external_deal_id) " + "to create a deal" + ) + if request.account_id is None: + raise ValueError( + "Index Exchange requires account_id (SSPDealCreateRequest.account_id) " + "to create a deal" + ) + + class_id, auction_type, programmatic_guaranteed = _ix_deal_config(request.deal_type) + if request.dsp_id is None: + raise ValueError( + f"Index Exchange requires dsp_id (SSPDealCreateRequest.dsp_id) for " + f"deal_type={request.deal_type.value}" + ) + body: dict[str, Any] = { - "deal_type": _IX_DEAL_TYPE_MAP.get(request.deal_type, "pmp"), + "classID": class_id, + "name": request.name, + "externalDealID": request.external_deal_id, + "account": {"accountID": request.account_id}, + "auctionType": auction_type, + "floor": request.cpm, } - if request.name: - body["deal_name"] = request.name - if request.advertiser: - body["advertiser_name"] = request.advertiser - if request.cpm: - body["floor_price"] = request.cpm - if request.currency: - body["currency"] = request.currency if request.start_date: - body["start_date"] = request.start_date + body["startDate"] = request.start_date if request.end_date: - body["end_date"] = request.end_date - if request.buyer_seat_ids: - body["buyer_seat_ids"] = request.buyer_seat_ids - if request.impressions_goal: - body["impression_goal"] = request.impressions_goal + body["endDate"] = request.end_date + + if class_id == 1: + direct_config: dict[str, Any] = { + "dspID": request.dsp_id, + "programmaticGuaranteed": programmatic_guaranteed, + } + if request.buyer_seat_ids: + direct_config["seatIDs"] = request.buyer_seat_ids + if request.impressions_goal: + direct_config["impressionGoal"] = request.impressions_goal + body["directConfigurations"] = direct_config + else: + # classID 4 (Marketplace Package) requires marketplaceConfigurations + # instead of directConfigurations; only dspID is confirmed required. + body["marketplaceConfigurations"] = {"dspID": request.dsp_id} + + if request.advertiser: + body["labels"] = {"advertiser": request.advertiser} if request.targeting: body["targeting"] = request.targeting - resp = await http.post("/api/deals", json=body) - resp.raise_for_status() + try: + resp = await http.post(f"{_API_BASE_PATH}{_DEALS_PATH}", json=body) + resp.raise_for_status() + except Exception as exc: + if not _may_have_reached_index(exc): + raise + try: + existing = await self.find_deal_by_external_id( + request.external_deal_id, request.account_id + ) + except SSPDealOutcomeUnknownError as lookup_exc: + raise SSPDealOutcomeUnknownError( + f"Index Exchange create_deal for externalDealID " + f"{request.external_deal_id} failed ({exc}); {lookup_exc}" + ) from exc + if existing is None: + raise + logger.warning( + "Index Exchange create_deal failed (%s) but deal %s exists; returning it", + exc, + request.external_deal_id, + ) + return existing return self._parse_deal(resp.json()) + async def find_deal_by_external_id( + self, external_deal_id: str, account_id: int + ) -> Optional[SSPDeal]: + """Look up a deal by exact externalDealID, or None if it does not exist. + + GET /v3/deals `search` matches loosely, so results are filtered to an exact + externalDealID match and paged. Raises SSPDealOutcomeUnknownError if the + lookup fails or finds no exact match within the page cap. + """ + http = self._ensure_connected() + try: + for page_offset in range(_LOOKUP_MAX_PAGES): + resp = await http.get( + f"{_API_BASE_PATH}{_DEALS_PATH}", + params={ + "search": external_deal_id, + "accountIDs": account_id, + "pageOffset": page_offset, + "pageSize": _LOOKUP_PAGE_SIZE, + }, + ) + resp.raise_for_status() + raw_deals = resp.json().get("deals", []) + for raw in raw_deals: + if raw.get("externalDealID") == external_deal_id: + return self._parse_deal(raw) + if len(raw_deals) < _LOOKUP_PAGE_SIZE: + return None + except Exception as lookup_exc: + raise SSPDealOutcomeUnknownError( + f"the lookup for externalDealID {external_deal_id} failed ({lookup_exc}); " + "check Index Exchange before retrying" + ) from lookup_exc + raise SSPDealOutcomeUnknownError( + f"the lookup for externalDealID {external_deal_id} found no exact match within " + f"{_LOOKUP_MAX_PAGES} pages; check Index Exchange before retrying" + ) + + async def resolve_dsp_ids_for_seat_ids(self, seat_ids: list[str]) -> list[DspSeatMatch]: + """Resolve dspID(s) for the given buyer seat IDs. + + Calls GET /v1/dsps/-/seats?seatIDs=... — the Deals API's wildcard + dspID lookup, which searches across all DSPs for seats matching the + given (DSP-facing, alphanumeric) seat IDs. Only active seats are + returned; a deleted/paused trading desk is never a valid dsp_id to + create a new deal against. + + Issues one request per seat ID and tags each match with the seat ID + it was queried for (requested_seat_id), so callers never have to + infer the owning seat from the response's extendedSeatID. + + Returns an empty list if none matched, or more than one DspSeatMatch + if the seat ID(s) collide across multiple DSPs — callers must decide + how to handle that (e.g. gate on human approval) rather than + assuming the first match is correct. + """ + if not seat_ids: + return [] + + http = self._ensure_connected() + + async def _lookup(seat_id: str) -> list[DspSeatMatch]: + resp = await http.get( + f"{_API_BASE_PATH}/v1/dsps/-/seats", + params={"seatIDs": seat_id}, + ) + resp.raise_for_status() + return [ + DspSeatMatch( + dsp_id=seat["dspID"], + requested_seat_id=seat_id, + extended_seat_id=seat.get("extendedSeatID", ""), + buyer_name=seat.get("name", ""), + status=seat.get("status", ""), + ) + for seat in resp.json().get("seats", []) + if seat.get("status") == "A" + ] + + per_seat = await asyncio.gather(*(_lookup(s) for s in seat_ids)) + return [match for matches in per_seat for match in matches] + async def clone_deal( self, source_deal_id: str, overrides: Optional[dict[str, Any]] = None, ) -> SSPDeal: - """Clone a deal on Index Exchange.""" - http = self._ensure_connected() - - body = {"source_deal_id": source_deal_id} + """Clone a deal by composing get_deal() + create_deal(). + + Index Exchange has no `/{id}/copy` (or equivalent) route on + /v3/deals, so this retrieves the source deal — source_deal_id must + be the internalDealID, same convention as get_deal()/update_deal() + — and creates a new one with a freshly generated external_deal_id, + copying over its configuration. account_id and dsp_id are read back + out of the source deal's raw response, since SSPDeal doesn't carry + them as first-class fields. `overrides` may override any + SSPDealCreateRequest field, e.g. a specific external_deal_id or a + different account_id/dsp_id. + """ + source = await self.get_deal(source_deal_id) + raw = source.raw or {} + account = raw.get("account") or {} + direct_config = raw.get("directConfigurations") or {} + marketplace_config = raw.get("marketplaceConfigurations") or {} + + create_request = SSPDealCreateRequest( + deal_type=source.deal_type, + name=source.name, + advertiser=source.advertiser, + cpm=source.cpm, + start_date=source.start_date, + end_date=source.end_date, + targeting=source.targeting, + impressions_goal=source.impressions_goal, + buyer_seat_ids=direct_config.get("seatIDs", []), + external_deal_id=_generate_clone_external_deal_id( + source.external_deal_id or source.deal_id + ), + account_id=account.get("accountID"), + dsp_id=direct_config.get("dspID") or marketplace_config.get("dspID"), + ) if overrides: - body.update(overrides) + create_request = create_request.model_copy(update=overrides) - resp = await http.post(f"/api/deals/{source_deal_id}/copy", json=body) - resp.raise_for_status() - return self._parse_deal(resp.json()) + return await self.create_deal(create_request) async def get_deal(self, deal_id: str) -> SSPDeal: - """Get deal details from Index Exchange.""" + """Get deal details from Index Exchange. + + deal_id must be the internalDealID (primary key for deals at Index), not + the externalDealID. + """ http = self._ensure_connected() - resp = await http.get(f"/api/deals/{deal_id}") + resp = await http.get(f"{_API_BASE_PATH}{_DEALS_PATH}/{deal_id}") resp.raise_for_status() - return self._parse_deal(resp.json()) + deal = self._parse_deal(resp.json()) + etag = resp.headers.get("ETag") + if etag and deal.raw is not None: + deal.raw["_etag"] = etag + return deal async def list_deals( self, @@ -143,59 +428,123 @@ async def list_deals( status: Optional[SSPDealStatus] = None, limit: int = 100, ) -> list[SSPDeal]: - """List deals from Index Exchange.""" + """List deals from Index Exchange via GET /v3/deals.""" http = self._ensure_connected() - params: dict[str, Any] = {"limit": limit} + params: dict[str, Any] = {"pageOffset": 0, "pageSize": limit} if status: - params["status"] = status.value - - resp = await http.get("/api/deals", params=params) + ix_status = _IX_LIST_STATUS_MAP.get(status) + if ix_status is None: + raise ValueError( + f"Index Exchange has no '{status.value}' deal status to filter on; " + f"supported: {sorted(s.value for s in _IX_LIST_STATUS_MAP)}" + ) + params["status"] = ix_status + + resp = await http.get(f"{_API_BASE_PATH}{_DEALS_PATH}", params=params) resp.raise_for_status() data = resp.json() - items = data if isinstance(data, list) else data.get("deals", data.get("data", [])) - return [self._parse_deal(d) for d in items] + return [self._parse_deal(d) for d in data.get("deals", [])] async def update_deal( self, deal_id: str, updates: dict[str, Any], + etag: Optional[str] = None, ) -> SSPDeal: - """Update a deal on Index Exchange.""" + """Update a deal on Index Exchange via PATCH /v3/deals/{internalDealID}. + + deal_id must be the internalDealID (not externalDealID). Index requires + an If-Match header carrying the deal's ETag. Pass the `etag` captured + when the caller read the deal (get_deal() stores it in `raw["_etag"]`) + to get lost-update protection; without one, the current ETag is + fetched immediately before the PATCH, which cannot detect concurrent + edits. `updates["status"]`, if present, must be "active" or "paused" + (not "ACTIVE"/"PAUSED") — expired/auto-paused are system-set and + cannot be patched. + """ http = self._ensure_connected() - resp = await http.post(f"/api/deals/{deal_id}", json=updates) + if not etag: + get_resp = await http.get(f"{_API_BASE_PATH}{_DEALS_PATH}/{deal_id}") + get_resp.raise_for_status() + etag = get_resp.headers.get("ETag") + if not etag: + raise ValueError( + f"Index Exchange did not return an ETag for deal {deal_id}; cannot PATCH" + ) + + resp = await http.patch( + f"{_API_BASE_PATH}{_DEALS_PATH}/{deal_id}", + json=updates, + headers={"If-Match": etag}, + ) resp.raise_for_status() return self._parse_deal(resp.json()) async def troubleshoot_deal(self, deal_id: str) -> SSPTroubleshootResult: - """Troubleshoot a deal on Index Exchange. - - Index Exchange doesn't have a dedicated troubleshooting endpoint - (unlike PubMatic's MCP). We use the reporting API to pull deal - performance data and flag issues. + """Diagnose a deal via GET /v3/deals/{internalDealID}. - TODO: Integrate with IX Reporting API for real diagnostics. + Index Exchange has no dedicated troubleshooting endpoint (unlike + PubMatic's MCP), so health is derived from the deal's own status and + configuration rather than a reporting API. """ http = self._ensure_connected() - # Get deal details as a baseline try: - resp = await http.get(f"/api/deals/{deal_id}") + resp = await http.get(f"{_API_BASE_PATH}{_DEALS_PATH}/{deal_id}") resp.raise_for_status() deal_data = resp.json() - except Exception: - deal_data = {} + except Exception as exc: + return SSPTroubleshootResult( + deal_id=deal_id, + health_score=0, + status="unreachable", + primary_issues=[f"Failed to fetch deal from Index Exchange: {exc}"], + ssp_type=self.ssp_type, + ) + + status_str = str(deal_data.get("status", "")).lower() + issues: list[str] = [] + recommendations: list[dict[str, str]] = [] + + if status_str == "active": + health_score: Optional[int] = 90 + elif status_str == "paused": + health_score = 50 + issues.append("Deal is paused") + recommendations.append( + {"action": "Resume the deal via PATCH if pausing was unintentional"} + ) + elif status_str == "auto-paused": + health_score = 20 + issues.append("Deal was automatically paused by Index Exchange due to inactivity") + recommendations.append( + { + "action": "Review deal configuration (floor, targeting, budget) " + "that may have triggered the system pause" + } + ) + elif status_str == "expired": + health_score = 0 + issues.append("Deal has expired") + else: + health_score = None + + class_id = deal_data.get("classID") + direct_config = deal_data.get("directConfigurations") or {} + if class_id == 1 and not direct_config.get("seatIDs"): + issues.append("No buyer seat IDs are configured") + if not deal_data.get("floor"): + issues.append("Floor price is not set") return SSPTroubleshootResult( - deal_id=deal_id, - status=deal_data.get("status", "unknown"), - primary_issues=[], - root_causes=[], - recommendations=[ - {"action": "Check IX reporting dashboard for detailed deal diagnostics"}, - ], + deal_id=str(deal_data.get("externalDealID", deal_id)), + health_score=health_score, + status=status_str or "unknown", + primary_issues=issues, + recommendations=recommendations, ssp_type=self.ssp_type, raw=deal_data, ) @@ -203,31 +552,36 @@ async def troubleshoot_deal(self, deal_id: str) -> SSPTroubleshootResult: # --- Index Exchange-specific response parsing --- def _parse_deal(self, raw: dict[str, Any]) -> SSPDeal: - """Parse Index Exchange deal response to normalized SSPDeal.""" - status_str = str(raw.get("status", "pending")).lower() + """Parse an Index Exchange /v3/deals response into a normalized SSPDeal.""" + status_str = str(raw.get("status", "")).lower() + class_id = raw.get("classID") + direct_config = raw.get("directConfigurations") or {} + labels = raw.get("labels") or {} - # IX may use different field names - deal_type_raw = raw.get("deal_type", raw.get("type", "pmp")) deal_type = SSPDealType.PMP - if deal_type_raw in ("pmp", "PMP"): - deal_type = SSPDealType.PMP - elif deal_type_raw in ("programmatic_guaranteed", "pg", "PG"): - deal_type = SSPDealType.PG - elif deal_type_raw in ("preferred", "preferred_deal"): - deal_type = SSPDealType.PREFERRED + if class_id == 1: + if direct_config.get("programmaticGuaranteed"): + deal_type = SSPDealType.PG + elif raw.get("auctionType") == "fixed": + deal_type = SSPDealType.PREFERRED + else: + deal_type = SSPDealType.PMP + elif class_id == 4: + deal_type = SSPDealType.AUCTION_PACKAGE return SSPDeal( - deal_id=str(raw.get("deal_id", raw.get("id", "unknown"))), - name=raw.get("deal_name", raw.get("name")), + deal_id=str(raw.get("internalDealID", raw.get("externalDealID", "unknown"))), + external_deal_id=raw.get("externalDealID"), + name=raw.get("name"), deal_type=deal_type, - status=_IX_STATUS_MAP.get(status_str, SSPDealStatus.CREATED), - advertiser=raw.get("advertiser_name", raw.get("advertiser")), - cpm=raw.get("floor_price", raw.get("cpm")), - currency=raw.get("currency", "USD"), - start_date=raw.get("start_date"), - end_date=raw.get("end_date"), + status=_IX_STATUS_MAP.get(status_str, SSPDealStatus.ERROR), + advertiser=labels.get("advertiser"), + cpm=raw.get("floor"), + currency="USD", + start_date=raw.get("startDate"), + end_date=raw.get("endDate"), targeting=raw.get("targeting"), - impressions_goal=raw.get("impression_goal", raw.get("impressions_goal")), + impressions_goal=direct_config.get("impressionGoal"), ssp_type=SSPType.INDEX_EXCHANGE, ssp_name="Index Exchange", raw=raw, diff --git a/src/ad_seller/config/settings.py b/src/ad_seller/config/settings.py index ffe3f328..310405c4 100644 --- a/src/ad_seller/config/settings.py +++ b/src/ad_seller/config/settings.py @@ -199,6 +199,8 @@ class Settings(BaseSettings): magnite_api_key: Optional[str] = None # Index Exchange SSP (REST API) index_exchange_api_url: Optional[str] = None + # Keycloak JWT bearer token — not a static API key. Tokens expire and must + # be refreshed via the client-credentials grant (not yet automated here). index_exchange_api_key: Optional[str] = None # Deal Sync Connectors (external deal-sync services, peer of SSP connectors) diff --git a/src/ad_seller/events/models.py b/src/ad_seller/events/models.py index b6018e37..ec5867cb 100644 --- a/src/ad_seller/events/models.py +++ b/src/ad_seller/events/models.py @@ -103,6 +103,10 @@ class ApprovalStatus(str, Enum): APPROVED = "approved" REJECTED = "rejected" TIMED_OUT = "timed_out" + CONSUMED = "consumed" # approved and already acted on (terminal) + # Approved, but whether the SSP created the deal could not be confirmed; + # resuming re-checks the SSP and never creates a second deal. + OUTCOME_UNKNOWN = "outcome_unknown" class ApprovalRequest(BaseModel): diff --git a/src/ad_seller/flows/execution_activation_flow.py b/src/ad_seller/flows/execution_activation_flow.py index 0e9ac441..d61d0dfc 100644 --- a/src/ad_seller/flows/execution_activation_flow.py +++ b/src/ad_seller/flows/execution_activation_flow.py @@ -275,6 +275,7 @@ async def distribute_to_ssps(self) -> None: deal_type=ssp_deal_type, name=f"Deal {deal.deal_id}", cpm=deal.price, + external_deal_id=deal.deal_id, ) # SSP block diff --git a/src/ad_seller/interfaces/api/routers/approvals.py b/src/ad_seller/interfaces/api/routers/approvals.py index c6694055..cd56b29b 100644 --- a/src/ad_seller/interfaces/api/routers/approvals.py +++ b/src/ad_seller/interfaces/api/routers/approvals.py @@ -5,16 +5,18 @@ These are privileged control-plane endpoints: submitting a decision here finalizes (or rejects/counters) a gated deal. They therefore REQUIRE a -valid API-key credential (EP-4.5). The approver identity written to the -audit trail is the VERIFIED principal derived from that credential — -``body.decided_by`` is retained only as a free-text display label. +valid OPERATOR API-key credential (EP-4.5): approval requests snapshot +deal pricing and resume can create the deal, so buyer keys are rejected. +The approver identity written to the audit trail is the VERIFIED principal +derived from that credential — ``body.decided_by`` is retained only as a +free-text display label. """ from fastapi import APIRouter, Depends from ....auth.dependencies import ( principal_from_api_key, - require_api_key_record, + require_operator_key, ) from ....models.api_key import ApiKeyRecord from ....services import approval_service @@ -25,20 +27,20 @@ @router.get("/approvals", tags=["Approvals"]) async def list_pending_approvals( - _principal: ApiKeyRecord = Depends(require_api_key_record), + _principal: ApiKeyRecord = Depends(require_operator_key), ): - """List all pending approval requests. Requires authentication.""" + """List all pending approval requests. Requires an operator credential.""" return await approval_service.list_pending_approvals() @router.get("/approvals/{approval_id}", tags=["Approvals"]) async def get_approval( approval_id: str, - _principal: ApiKeyRecord = Depends(require_api_key_record), + _principal: ApiKeyRecord = Depends(require_operator_key), ): """Get a specific approval request and its response (if any). - Requires authentication. + Requires an operator credential. """ return await approval_service.get_approval(approval_id) @@ -47,9 +49,9 @@ async def get_approval( async def decide_approval( approval_id: str, body: ApprovalDecisionRequest, - principal: ApiKeyRecord = Depends(require_api_key_record), + principal: ApiKeyRecord = Depends(require_operator_key), ): - """Submit a human decision for a pending approval. Requires authentication. + """Submit a human decision for a pending approval. Requires an operator credential. The verified principal from the authenticated API key is stamped into the audit record; ``body.decided_by`` is kept as a display label only. @@ -67,11 +69,11 @@ async def decide_approval( @router.post("/approvals/{approval_id}/resume", tags=["Approvals"]) async def resume_flow( approval_id: str, - _principal: ApiKeyRecord = Depends(require_api_key_record), + _principal: ApiKeyRecord = Depends(require_operator_key), ): """Resume a flow after an approval decision has been submitted. - Requires authentication. Loads the flow state snapshot, applies the + Requires an operator credential. Loads the flow state snapshot, applies the decision, and returns the final result without re-running expensive crew evaluations. """ diff --git a/src/ad_seller/interfaces/api/schemas.py b/src/ad_seller/interfaces/api/schemas.py index cea4302a..cf4a5fb9 100644 --- a/src/ad_seller/interfaces/api/schemas.py +++ b/src/ad_seller/interfaces/api/schemas.py @@ -632,9 +632,10 @@ class SSPDealDistributeRequest(BaseModel): advertiser: Optional[str] = None cpm: Optional[float] = None buyer_seat_ids: Optional[list[str]] = None + account_id: Optional[int] = None start_date: Optional[str] = None end_date: Optional[str] = None - targeting: Optional[dict[str, Any]] = None + targeting: Optional[dict[str, Any] | list[dict[str, Any]]] = None # Routing hint — if set, routes to this SSP. Otherwise uses routing rules. ssp_name: Optional[str] = None inventory_type: Optional[str] = None # for routing: ctv, display, video, etc. diff --git a/src/ad_seller/interfaces/mcp_server.py b/src/ad_seller/interfaces/mcp_server.py index 0754cef1..61440cba 100644 --- a/src/ad_seller/interfaces/mcp_server.py +++ b/src/ad_seller/interfaces/mcp_server.py @@ -791,9 +791,15 @@ async def distribute_deal_via_ssp( cpm: float = 0, ssp_name: str = "", inventory_type: str = "", + name: str = "", + account_id: int = 0, + targeting: dict[str, Any] | list[dict[str, Any]] | None = None, ) -> str: """Distribute a deal through configured SSP(s). - Routes based on ssp_name or inventory_type routing rules.""" + Routes based on ssp_name or inventory_type routing rules. + Some SSPs (e.g. Index Exchange) require name, account_id, cpm, and/or + targeting to actually create the deal. targeting example: + [{"targetingType":"standard","keyName":"domain","sets":[{"values":[{"value":"example.com"}],"operator":"ANY_OF"}]}].""" denied = await _deny_unless_operator() if denied: return denied @@ -805,13 +811,14 @@ async def distribute_deal_via_ssp( request = SimpleNamespace( deal_id=deal_id, deal_type=deal_type, - name=None, + name=name or None, advertiser=None, cpm=cpm or None, buyer_seat_ids=None, + account_id=account_id or None, start_date=None, end_date=None, - targeting=None, + targeting=targeting, ssp_name=ssp_name or None, inventory_type=inventory_type or None, ) @@ -993,14 +1000,29 @@ async def transition_order(order_id: str, new_status: str, reason: str = "") -> @mcp.tool() async def list_pending_approvals() -> str: """List pending approval requests waiting for human decision.""" + denied = await _deny_unless_operator() + if denied: + return denied + from ..services import approval_service return await _service_json(approval_service.list_pending_approvals()) @mcp.tool() -async def approve_or_reject(approval_id: str, decision: str, reason: str = "") -> str: - """Submit an approval decision. decision: 'approve', 'reject', or 'counter'.""" +async def approve_or_reject( + approval_id: str, + decision: str, + reason: str = "", + modifications: dict[str, Any] | None = None, +) -> str: + """Submit an approval decision. decision: 'approve', 'reject', or 'counter'. + Some gates require modifications when approving — e.g. a dsp_resolution + gate (seat ID matched multiple DSPs) needs a selected_dsp_id, e.g. + {"selected_dsp_id": 85}. For an SSP-distribution approval, you can also + include any other field the SSP needs that wasn't set when the gate + fired (e.g. "account_id", "name", "cpm", "targeting") in the same + object — resume_approval applies all of them, not just selected_dsp_id.""" denied = await _deny_unless_operator() if denied: return denied @@ -1012,10 +1034,27 @@ async def approve_or_reject(approval_id: str, decision: str, reason: str = "") - approval_id=approval_id, decision=decision, reason=reason, + modifications=modifications, ) ) +@mcp.tool() +async def resume_approval(approval_id: str) -> str: + """Resume a flow that was paused for approval, after a decision has been + submitted via approve_or_reject. For an approved dsp_resolution gate, + this is the step that actually calls the SSP's create_deal(). For a + rejected approval, this returns a rejection result without creating + anything.""" + denied = await _deny_unless_operator() + if denied: + return denied + + from ..services import approval_service + + return await _service_json(approval_service.resume_flow(approval_id)) + + @mcp.tool() async def set_approval_gates( enabled: bool, required_flows: str = "", timeout_hours: int = 24 @@ -1139,8 +1178,12 @@ async def create_curated_deal( product_id: str = "", max_cpm: float = 0, impressions: int = 0, + buyer_seat_ids: str = "", ) -> str: - """Create a deal with curator overlay. The curator's fee is added on top.""" + """Create a deal with curator overlay. The curator's fee is added on top. + Pass buyer_seat_ids as a comma-separated list to attach DSP-facing seat + IDs at creation time — needed for SSPs that resolve dspID from a seat + (e.g. Index Exchange) later, via distribute_deal_via_ssp.""" from types import SimpleNamespace from ..services import deal_service @@ -1153,7 +1196,7 @@ async def create_curated_deal( impressions=impressions or None, flight_start=None, flight_end=None, - buyer_seat_ids=[], + buyer_seat_ids=[s.strip() for s in buyer_seat_ids.split(",") if s.strip()], audience_segments=[], content_categories=[], ) diff --git a/src/ad_seller/services/approval_service.py b/src/ad_seller/services/approval_service.py index 53f35f2e..c0bc2e8c 100644 --- a/src/ad_seller/services/approval_service.py +++ b/src/ad_seller/services/approval_service.py @@ -11,6 +11,7 @@ from typing import Any from fastapi import HTTPException +from pydantic import ValidationError logger = logging.getLogger(__name__) @@ -63,6 +64,19 @@ async def decide_approval( storage = await get_storage() gate = ApprovalGate(storage) + + # Reject a bad dsp_resolution decision while the approval is still pending, so it + # can be corrected instead of surfacing as a 400 on resume. + pending_request = await gate.get_request(approval_id) + if ( + pending_request + and pending_request.status.value == "pending" + and decision == "approve" + and pending_request.flow_type == "ssp_distribution" + and pending_request.gate_name == "dsp_resolution" + ): + _build_dsp_resolution_create_request(pending_request, modifications or {}) + try: response = await gate.submit_decision( approval_id=approval_id, @@ -107,12 +121,169 @@ async def resume_flow(approval_id: str) -> dict[str, Any]: if request.flow_type == "proposal_handling" and request.gate_name == "proposal_decision": return await _resume_proposal_flow(request, response) + if request.flow_type == "ssp_distribution" and request.gate_name == "dsp_resolution": + return await _resume_ssp_distribution(request, response) + raise HTTPException( status_code=400, detail=f"Unknown flow_type/gate_name: {request.flow_type}/{request.gate_name}", ) +async def _set_approval_status(request, status) -> None: + from ..storage.factory import get_storage + + request.status = status + storage = await get_storage() + await storage.set(f"approval:{request.approval_id}", request.model_dump(mode="json")) + + +_DECISION_TIME_FIELDS = frozenset( + { + "name", + "cpm", + "account_id", + "advertiser", + "targeting", + "start_date", + "end_date", + "impressions_goal", + } +) + + +def _build_dsp_resolution_create_request(request, modifications: dict[str, Any]): + """Validate a dsp_resolution decision's modifications and build the create request. + + Raises HTTPException(400) for a missing/unknown selected_dsp_id, unknown or + disallowed keys, or values that fail SSPDealCreateRequest validation. + """ + from ..clients.ssp_base import SSPDealCreateRequest + + selected_dsp_id = modifications.get("selected_dsp_id") + if selected_dsp_id is None: + raise HTTPException( + status_code=400, + detail='Approving a dsp_resolution gate requires modifications={"selected_dsp_id": }', + ) + + unknown_keys = sorted(set(modifications) - _DECISION_TIME_FIELDS - {"selected_dsp_id"}) + if unknown_keys: + raise HTTPException( + status_code=400, + detail=( + f"Unsupported modifications {unknown_keys}; allowed: " + f"{sorted(_DECISION_TIME_FIELDS | {'selected_dsp_id'})}" + ), + ) + + candidate_dsp_ids = {str(c.get("dsp_id")) for c in request.context.get("candidates", [])} + if str(selected_dsp_id) not in candidate_dsp_ids: + raise HTTPException( + status_code=400, + detail=( + f"selected_dsp_id {selected_dsp_id} is not one of the candidate DSPs " + f"for this approval: {sorted(candidate_dsp_ids)}" + ), + ) + + supplied = {k: v for k, v in modifications.items() if k in _DECISION_TIME_FIELDS} + try: + return SSPDealCreateRequest.model_validate( + { + **request.flow_state_snapshot["create_request"], + **supplied, + "dsp_id": selected_dsp_id, + } + ) + except ValidationError as exc: + raise HTTPException( + status_code=400, detail=f"Invalid modifications: {exc.errors(include_url=False)}" + ) from exc + + +async def _resume_ssp_distribution(request, response): + """Resume an SSP distribution after a dsp_resolution approval decision. + + Unlike _resume_proposal_flow, there's no CrewAI flow/state machine to + re-hydrate here — distribute_deal_via_ssp is a plain service call, so the + snapshot is just the SSP name and the SSPDealCreateRequest that was + waiting on a dsp_id. On approval, the human's selected_dsp_id + (response.modifications) is applied and create_deal() finally runs. + """ + from ..clients.ssp_base import SSPDealOutcomeUnknownError + from ..clients.ssp_factory import build_ssp_registry + from ..events.models import ApprovalStatus + + if request.status == ApprovalStatus.CONSUMED: + raise HTTPException( + status_code=409, + detail=f"Approval {request.approval_id} has already been resumed", + ) + + if response.decision != "approve": + return { + "deal_id": request.deal_id, + "status": response.decision, + "resumed_from_approval": request.approval_id, + } + + create_request = _build_dsp_resolution_create_request(request, response.modifications) + + registry = build_ssp_registry() + ssp = registry.get_client(request.flow_state_snapshot["ssp_name"]) + + def _success(result) -> dict[str, Any]: + return { + "deal_id": result.deal_id, + "external_deal_id": result.external_deal_id, + "ssp": result.ssp_name, + "ssp_type": result.ssp_type.value, + "status": result.status.value, + "deal": result.model_dump(exclude={"raw"}), + "resumed_from_approval": request.approval_id, + } + + if request.status == ApprovalStatus.OUTCOME_UNKNOWN: + # A previous create may or may not have landed. Only look the deal up here; never + # POST, so this cannot create a second deal. + try: + async with ssp: + existing = await ssp.find_deal_by_external_id( + create_request.external_deal_id, create_request.account_id + ) + except SSPDealOutcomeUnknownError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + if existing is not None: + await _set_approval_status(request, ApprovalStatus.CONSUMED) + return _success(existing) + await _set_approval_status(request, ApprovalStatus.APPROVED) + raise HTTPException( + status_code=409, + detail=( + f"No deal with externalDealID {create_request.external_deal_id} exists in the " + "SSP; the approval was restored to 'approved'. Resume again to create it." + ), + ) + + # Mark the approval consumed before creating the deal so a repeated resume + # cannot create it twice; restore it if creation fails so the call can be retried. + # Storage has no compare-and-set, so two truly simultaneous resumes can still race. + await _set_approval_status(request, ApprovalStatus.CONSUMED) + try: + async with ssp: + result = await ssp.create_deal(create_request) + except SSPDealOutcomeUnknownError as exc: + # The deal may exist in the SSP; resuming again re-checks it instead of re-creating. + await _set_approval_status(request, ApprovalStatus.OUTCOME_UNKNOWN) + raise HTTPException(status_code=502, detail=str(exc)) from exc + except Exception: + await _set_approval_status(request, ApprovalStatus.APPROVED) + raise + + return _success(result) + + async def _resume_proposal_flow(request, response): """Resume a proposal handling flow after approval decision.""" from datetime import datetime diff --git a/src/ad_seller/services/deal_service.py b/src/ad_seller/services/deal_service.py index 05be02f6..2fd53602 100644 --- a/src/ad_seller/services/deal_service.py +++ b/src/ad_seller/services/deal_service.py @@ -1086,11 +1086,106 @@ async def get_deal_buyer_status(deal_id: str, buyer_url: str) -> dict[str, Any]: # ============================================================================= +async def _resolve_dsp_id_or_pending_approval( + ssp: Any, + create_request: Any, + deal_id: str, +) -> Optional[dict[str, Any]]: + """Resolve create_request.dsp_id from buyer_seat_ids if not already set. + + Mutates create_request.dsp_id in place when exactly one DSP owns every + seat ID. Raises a 400 if any seat has no active match or the seats share + no DSP (a deal can target only one). If more than one DSP owns every + seat (a seat ID collision), creates a pending ApprovalGate request instead of guessing which one is + correct, and returns the response dict the caller should return + immediately — the actual create_deal() call happens later, via the + ssp_distribution/dsp_resolution resume path, once a human picks one. + + Returns None if no gating was needed: dsp_id was already set, no seat + IDs were given, the SSP doesn't support seat resolution at all, or + resolution found exactly one candidate. + """ + if create_request.dsp_id is not None or not create_request.buyer_seat_ids: + return None + + resolver = getattr(ssp, "resolve_dsp_ids_for_seat_ids", None) + if resolver is None: + return None + + seat_ids = create_request.buyer_seat_ids + matches = await resolver(seat_ids) + + # A deal carries a single dspID, so it must be valid for every seat. + dsps_by_seat: dict[str, set[int]] = {s: set() for s in seat_ids} + for m in matches: + dsps_by_seat.setdefault(m.requested_seat_id, set()).add(m.dsp_id) + + unmatched = [s for s in seat_ids if not dsps_by_seat[s]] + if unmatched: + raise HTTPException( + status_code=400, + detail={ + "error": "dsp_resolution_failed", + "message": f"No active DSP found for seat ID(s): {unmatched}", + }, + ) + + distinct_dsp_ids = sorted(set.intersection(*(dsps_by_seat[s] for s in seat_ids))) + if not distinct_dsp_ids: + mapping = {s: sorted(dsps_by_seat[s]) for s in seat_ids} + raise HTTPException( + status_code=400, + detail={ + "error": "dsp_resolution_failed", + "message": ( + f"Seat IDs do not belong to a single DSP (seat -> DSP IDs: {mapping}). " + "A deal can target only one DSP; create a separate deal per DSP." + ), + }, + ) + if len(distinct_dsp_ids) == 1: + create_request.dsp_id = distinct_dsp_ids[0] + return None + + candidates = [m for m in matches if m.dsp_id in distinct_dsp_ids] + + from ..events.approval import ApprovalGate + from ..storage.factory import get_storage + + storage = await get_storage() + gate = ApprovalGate(storage) + approval_req = await gate.request_approval( + flow_id=f"ssp-distribute-{deal_id}", + flow_type="ssp_distribution", + gate_name="dsp_resolution", + context={ + "seat_ids": create_request.buyer_seat_ids, + "candidates": [m.model_dump() for m in candidates], + }, + flow_state_snapshot={ + "ssp_name": ssp.ssp_type.value, + "create_request": create_request.model_dump(mode="json"), + }, + deal_id=deal_id, + ) + return { + "status": "pending_approval", + "approval_id": approval_req.approval_id, + "message": ( + f"Seat ID(s) {create_request.buyer_seat_ids} matched multiple DSPs " + f"{distinct_dsp_ids}; a human must select the correct one via " + f"POST /approvals/{approval_req.approval_id}/decide " + '(modifications={"selected_dsp_id": }) before this deal can be created.' + ), + } + + async def distribute_deal_via_ssp(request: Any) -> dict[str, Any]: """Distribute a deal through configured SSP(s) or deal-sync providers.""" from ..clients.deal_sync_factory import build_deal_sync_registry - from ..clients.ssp_base import SSPDealCreateRequest, SSPDealType + from ..clients.ssp_base import SSPDealCreateRequest, SSPDealOutcomeUnknownError, SSPDealType from ..clients.ssp_factory import build_ssp_registry + from ..storage.factory import get_storage ssp_registry = build_ssp_registry() sync_registry = build_deal_sync_registry() @@ -1147,22 +1242,39 @@ async def distribute_deal_via_ssp(request: Any) -> dict[str, Any]: "preferred": SSPDealType.PREFERRED, } + # Same fallback push_deal_to_buyers already uses: no MCP tool lets the + # LLM set buyer_seat_ids on this route, so without this a stored deal's + # seat IDs (set at creation time) would never reach DSP resolution. + storage = await get_storage() + stored_deal = await storage.get_deal(request.deal_id) + create_request = SSPDealCreateRequest( deal_type=deal_type_map.get(request.deal_type or "PMP", SSPDealType.PMP), name=request.name, advertiser=request.advertiser, cpm=request.cpm, - buyer_seat_ids=request.buyer_seat_ids or [], - start_date=request.start_date, - end_date=request.end_date, + buyer_seat_ids=request.buyer_seat_ids or (stored_deal or {}).get("buyer_seat_ids", []), + account_id=getattr(request, "account_id", None) or (stored_deal or {}).get("account_id"), + start_date=request.start_date or (stored_deal or {}).get("flight_start"), + end_date=request.end_date or (stored_deal or {}).get("flight_end"), targeting=request.targeting, + external_deal_id=request.deal_id, ) try: async with client: + if not is_deal_sync: + pending = await _resolve_dsp_id_or_pending_approval( + client, create_request, request.deal_id + ) + if pending is not None: + return pending + result = await client.create_deal(create_request) except ValueError as e: raise HTTPException(status_code=400, detail={"error": "invalid_request", "message": str(e)}) + except SSPDealOutcomeUnknownError as e: + raise HTTPException(status_code=502, detail={"error": "outcome_unknown", "message": str(e)}) if is_deal_sync: return { diff --git a/tests/integration/fixtures/index_exchange_deal_response.json b/tests/integration/fixtures/index_exchange_deal_response.json index 57ccbc6e..4c0c35cf 100644 --- a/tests/integration/fixtures/index_exchange_deal_response.json +++ b/tests/integration/fixtures/index_exchange_deal_response.json @@ -1,15 +1,30 @@ { - "deal_id": "IX-DEAL-20260326-001", - "ssp_name": "index_exchange", - "status": "active", - "deal_type": "PREFERRED", + "internalDealID": 987654, + "externalDealID": "IX-DEAL-20260326-001", + "classID": 1, "name": "Test Preferred Deal via Index Exchange", - "cpm": 22.50, - "currency": "USD", - "buyer_seats": ["seat-dv360-001"], - "start_date": "2026-03-26", - "end_date": "2026-05-26", - "created_at": "2026-03-26T10:00:00Z", - "publisher_id": "pub-12345", - "inventory_source": "video" + "account": { "accountID": 12345 }, + "startDate": "2026-03-26", + "endDate": "2026-05-26", + "modifiedDate": "2026-03-26 10:00:00", + "auctionType": "fixed", + "floor": 22.50, + "status": "active", + "directConfigurations": { + "dspID": 5551, + "seatIDs": ["seat-dv360-001"], + "programmaticGuaranteed": false + }, + "labels": { + "advertiser": "Acme Corp" + }, + "targeting": [ + { + "targetingType": "standard", + "keyName": "Country", + "sets": [ + { "values": [{ "value": "US", "label": "United States" }], "operator": "ANY_OF" } + ] + } + ] } diff --git a/tests/integration/test_deal_flow_e2e.py b/tests/integration/test_deal_flow_e2e.py index e9f1128f..5684f3e2 100644 --- a/tests/integration/test_deal_flow_e2e.py +++ b/tests/integration/test_deal_flow_e2e.py @@ -346,9 +346,10 @@ def test_pubmatic_response_structure(self, pubmatic_deal_response): assert pubmatic_deal_response["cpm"] == 15.0 def test_ix_response_structure(self, ix_deal_response): - assert ix_deal_response["ssp_name"] == "index_exchange" + assert ix_deal_response["externalDealID"] == "IX-DEAL-20260326-001" assert ix_deal_response["status"] == "active" - assert ix_deal_response["cpm"] == 22.5 + assert ix_deal_response["floor"] == 22.5 + assert ix_deal_response["classID"] == 1 def test_products_can_become_product_definitions(self, freewheel_products): """Verify fixture data can be loaded into ProductDefinition models.""" diff --git a/tests/unit/test_approval_gates.py b/tests/unit/test_approval_gates.py index 19936b15..8820be7c 100644 --- a/tests/unit/test_approval_gates.py +++ b/tests/unit/test_approval_gates.py @@ -43,6 +43,7 @@ from ad_seller.models.api_key import ( # noqa: E402 API_KEY_STORAGE_PREFIX, ApiKeyRecord, + ApiKeyRole, generate_api_key, hash_api_key, ) @@ -70,8 +71,10 @@ def client(): return httpx.AsyncClient(transport=transport, base_url="http://test") -def _seed_key(store, *, key_id="key-approver", agency_id="agency-approve"): - """Seed a valid API key record; return the raw key.""" +def _seed_key( + store, *, key_id="key-approver", agency_id="agency-approve", role=ApiKeyRole.OPERATOR +): + """Seed a valid API key record (operator by default); return the raw key.""" raw_key = generate_api_key() key_hash = hash_api_key(raw_key) record = ApiKeyRecord( @@ -80,6 +83,7 @@ def _seed_key(store, *, key_id="key-approver", agency_id="agency-approve"): key_prefix_hint=raw_key[:12] + "...", identity=BuyerIdentity(agency_id=agency_id, agency_name="Acme Agency"), label="Approval test key", + role=role, ) store[f"{API_KEY_STORAGE_PREFIX}{key_hash}"] = record.model_dump(mode="json") return raw_key @@ -137,6 +141,36 @@ async def test_unauthenticated_list_and_resume_rejected(self, client, mock_stora assert list_resp.status_code == 401 assert resume_resp.status_code == 401 + async def test_buyer_key_rejected_on_every_approval_route(self, client, mock_storage): + """Approvals snapshot deal pricing and resume creates the deal, so a + buyer-role key gets 403 on list, get, decide and resume.""" + buyer_key = _seed_key(mock_storage._store, key_id="key-buyer", role=ApiKeyRole.BUYER) + approval_id = _seed_pending_approval(mock_storage._store) + headers = {"X-Api-Key": buyer_key} + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + responses = [ + await client.get("/approvals", headers=headers), + await client.get(f"/approvals/{approval_id}", headers=headers), + await client.post( + f"/approvals/{approval_id}/decide", + json={"decision": "approve"}, + headers=headers, + ), + await client.post(f"/approvals/{approval_id}/resume", headers=headers), + ] + assert [r.status_code for r in responses] == [403, 403, 403, 403] + assert mock_storage._store[f"approval:{approval_id}"]["status"] == "pending" + + async def test_operator_key_can_list_and_get_approvals(self, client, mock_storage): + operator_key = _seed_key(mock_storage._store, key_id="key-op") + approval_id = _seed_pending_approval(mock_storage._store) + headers = {"X-Api-Key": operator_key} + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + list_resp = await client.get("/approvals", headers=headers) + get_resp = await client.get(f"/approvals/{approval_id}", headers=headers) + assert list_resp.status_code == 200 + assert get_resp.status_code == 200 + async def test_authenticated_resume_after_decide_returns_200(self, client, mock_storage): """Resume after a recorded decision must not 500 on CrewAI's read-only state.""" from ad_seller.events.models import ApprovalResponse diff --git a/tests/unit/test_dsp_resolution_gate.py b/tests/unit/test_dsp_resolution_gate.py new file mode 100644 index 00000000..6456d65c --- /dev/null +++ b/tests/unit/test_dsp_resolution_gate.py @@ -0,0 +1,658 @@ +"""Unit tests for DSP-resolution approval gating in deal_service/approval_service. + +Covers: +- _resolve_dsp_id_or_pending_approval: dsp_id passthrough, no-seat-ids + short-circuit, single-match resolution, zero-match error, and + multi-match approval gating +- _resume_ssp_distribution: reject short-circuits, missing selected_dsp_id + raises, approve applies the human's selection and creates the deal +""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import HTTPException + +from ad_seller.clients.ssp_base import SSPDealCreateRequest, SSPDealType, SSPType +from ad_seller.clients.ssp_index_exchange import DspSeatMatch +from ad_seller.events.models import ApprovalRequest, ApprovalResponse, ApprovalStatus +from ad_seller.services.approval_service import _resume_ssp_distribution +from ad_seller.services.deal_service import _resolve_dsp_id_or_pending_approval + + +def _create_request(**overrides) -> SSPDealCreateRequest: + defaults = dict( + deal_type=SSPDealType.PMP, + name="Test Deal", + cpm=15.0, + buyer_seat_ids=["acc-100522"], + external_deal_id="EXT-DEAL-001", + account_id=12345, + ) + defaults.update(overrides) + return SSPDealCreateRequest(**defaults) + + +class TestResolveDspIdOrPendingApproval: + @pytest.mark.asyncio + async def test_returns_none_when_dsp_id_already_set(self): + ssp = MagicMock() + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock() + request = _create_request(dsp_id=17) + + result = await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert result is None + ssp.resolve_dsp_ids_for_seat_ids.assert_not_called() + + @pytest.mark.asyncio + async def test_returns_none_when_no_buyer_seat_ids(self): + ssp = MagicMock() + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock() + request = _create_request(buyer_seat_ids=[]) + + result = await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert result is None + ssp.resolve_dsp_ids_for_seat_ids.assert_not_called() + + @pytest.mark.asyncio + async def test_returns_none_when_ssp_lacks_resolver(self): + ssp = MagicMock(spec=["ssp_type", "create_deal"]) + request = _create_request() + + result = await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert result is None + assert request.dsp_id is None + + @pytest.mark.asyncio + async def test_single_match_sets_dsp_id(self): + ssp = MagicMock() + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock( + return_value=[DspSeatMatch(dsp_id=17, requested_seat_id="acc-100522", status="A")] + ) + request = _create_request() + + result = await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert result is None + assert request.dsp_id == 17 + + @pytest.mark.asyncio + async def test_no_match_raises_400(self): + ssp = MagicMock() + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock(return_value=[]) + request = _create_request() + + with pytest.raises(HTTPException) as exc_info: + await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert exc_info.value.status_code == 400 + assert exc_info.value.detail["error"] == "dsp_resolution_failed" + + @pytest.mark.asyncio + async def test_seats_on_different_dsps_raises_400(self): + ssp = MagicMock() + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock( + return_value=[ + DspSeatMatch(dsp_id=17, requested_seat_id="seat-a", status="A"), + DspSeatMatch(dsp_id=52, requested_seat_id="seat-b", status="A"), + ] + ) + request = _create_request(buyer_seat_ids=["seat-a", "seat-b"]) + + with pytest.raises(HTTPException) as exc_info: + await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert exc_info.value.status_code == 400 + assert exc_info.value.detail["error"] == "dsp_resolution_failed" + assert "separate deal per DSP" in exc_info.value.detail["message"] + assert request.dsp_id is None + + @pytest.mark.asyncio + async def test_one_seat_unmatched_raises_400(self): + ssp = MagicMock() + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock( + return_value=[DspSeatMatch(dsp_id=17, requested_seat_id="seat-a", status="A")] + ) + request = _create_request(buyer_seat_ids=["seat-a", "seat-b"]) + + with pytest.raises(HTTPException) as exc_info: + await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert exc_info.value.status_code == 400 + assert "seat-b" in exc_info.value.detail["message"] + assert "seat-a" not in exc_info.value.detail["message"] + + @pytest.mark.asyncio + async def test_overlapping_seats_resolve_to_shared_dsp(self): + ssp = MagicMock() + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock( + return_value=[ + DspSeatMatch(dsp_id=17, requested_seat_id="seat-a", status="A"), + DspSeatMatch(dsp_id=52, requested_seat_id="seat-a", status="A"), + DspSeatMatch(dsp_id=17, requested_seat_id="seat-b", status="A"), + ] + ) + request = _create_request(buyer_seat_ids=["seat-a", "seat-b"]) + + result = await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert result is None + assert request.dsp_id == 17 + + @pytest.mark.asyncio + async def test_match_resolves_when_extended_seat_id_differs_from_requested(self): + ssp = MagicMock() + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock( + return_value=[ + DspSeatMatch( + dsp_id=17, + requested_seat_id="acc-100522", + extended_seat_id="ACC-100522-NORMALIZED", + status="A", + ) + ] + ) + request = _create_request(buyer_seat_ids=["acc-100522"]) + + await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert request.dsp_id == 17 + + @pytest.mark.asyncio + async def test_gate_candidates_limited_to_dsps_shared_by_all_seats(self): + ssp = MagicMock() + ssp.ssp_type = SSPType.INDEX_EXCHANGE + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock( + return_value=[ + DspSeatMatch(dsp_id=17, requested_seat_id="seat-a", status="A"), + DspSeatMatch(dsp_id=52, requested_seat_id="seat-a", status="A"), + DspSeatMatch(dsp_id=99, requested_seat_id="seat-a", status="A"), + DspSeatMatch(dsp_id=17, requested_seat_id="seat-b", status="A"), + DspSeatMatch(dsp_id=52, requested_seat_id="seat-b", status="A"), + ] + ) + request = _create_request(buyer_seat_ids=["seat-a", "seat-b"]) + mock_gate_instance = MagicMock() + mock_gate_instance.request_approval = AsyncMock(return_value=MagicMock(approval_id="a")) + + with ( + patch( + "ad_seller.storage.factory.get_storage", + new_callable=AsyncMock, + return_value=MagicMock(), + ), + patch("ad_seller.events.approval.ApprovalGate", return_value=mock_gate_instance), + ): + result = await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert result["status"] == "pending_approval" + candidates = mock_gate_instance.request_approval.call_args.kwargs["context"]["candidates"] + assert {c["dsp_id"] for c in candidates} == {17, 52} + + @pytest.mark.asyncio + async def test_multiple_matches_requests_approval_and_returns_pending(self): + ssp = MagicMock() + ssp.ssp_type = SSPType.INDEX_EXCHANGE + ssp.resolve_dsp_ids_for_seat_ids = AsyncMock( + return_value=[ + DspSeatMatch(dsp_id=17, requested_seat_id="acc-100522", status="A"), + DspSeatMatch(dsp_id=52, requested_seat_id="acc-100522", status="A"), + ] + ) + request = _create_request() + + mock_approval_request = MagicMock(approval_id="apr-999") + mock_gate_instance = MagicMock() + mock_gate_instance.request_approval = AsyncMock(return_value=mock_approval_request) + + with ( + patch( + "ad_seller.storage.factory.get_storage", + new_callable=AsyncMock, + return_value=MagicMock(), + ), + patch( + "ad_seller.events.approval.ApprovalGate", + return_value=mock_gate_instance, + ), + ): + result = await _resolve_dsp_id_or_pending_approval(ssp, request, "deal-1") + + assert result == { + "status": "pending_approval", + "approval_id": "apr-999", + "message": ( + "Seat ID(s) ['acc-100522'] matched multiple DSPs [17, 52]; a human must " + "select the correct one via POST /approvals/apr-999/decide " + '(modifications={"selected_dsp_id": }) before this deal can be created.' + ), + } + assert request.dsp_id is None # left unresolved — resume path sets it + + call_kwargs = mock_gate_instance.request_approval.call_args.kwargs + assert call_kwargs["flow_type"] == "ssp_distribution" + assert call_kwargs["gate_name"] == "dsp_resolution" + assert call_kwargs["deal_id"] == "deal-1" + assert call_kwargs["context"]["seat_ids"] == ["acc-100522"] + assert {c["dsp_id"] for c in call_kwargs["context"]["candidates"]} == {17, 52} + assert call_kwargs["flow_state_snapshot"]["ssp_name"] == "index_exchange" + assert call_kwargs["flow_state_snapshot"]["create_request"]["external_deal_id"] == ( + "EXT-DEAL-001" + ) + + +def _storage(): + store: dict = {} + storage = AsyncMock() + storage.get = AsyncMock(side_effect=lambda k: store.get(k)) + storage.set = AsyncMock(side_effect=lambda k, v, ttl=None: store.__setitem__(k, v)) + storage._store = store + return storage + + +class TestResumeSSPDistribution: + def _approval_request(self, **overrides): + defaults = dict( + approval_id="apr-999", + event_id="evt-1", + flow_id="ssp-distribute-deal-1", + flow_type="ssp_distribution", + gate_name="dsp_resolution", + deal_id="deal-1", + context={"candidates": [{"dsp_id": 52}, {"dsp_id": 85}]}, + flow_state_snapshot={ + "ssp_name": "index_exchange", + "create_request": _create_request().model_dump(mode="json"), + }, + ) + defaults.update(overrides) + return ApprovalRequest(**defaults) + + def _mock_registry(self, created_deal=None, create_side_effect=None): + mock_ssp = AsyncMock() + mock_ssp.create_deal = AsyncMock(return_value=created_deal, side_effect=create_side_effect) + mock_ssp.__aenter__ = AsyncMock(return_value=mock_ssp) + mock_ssp.__aexit__ = AsyncMock(return_value=False) + mock_registry = MagicMock() + mock_registry.get_client.return_value = mock_ssp + return mock_registry, mock_ssp + + def _created_deal(self): + created_deal = MagicMock() + created_deal.deal_id = "EXT-DEAL-001" + created_deal.ssp_name = "Index Exchange" + created_deal.ssp_type = SSPType.INDEX_EXCHANGE + created_deal.status = MagicMock(value="active") + created_deal.model_dump.return_value = {"deal_id": "EXT-DEAL-001"} + return created_deal + + @pytest.mark.asyncio + async def test_reject_returns_without_creating_deal(self): + request = self._approval_request() + response = ApprovalResponse(approval_id="apr-999", decision="reject") + + with patch("ad_seller.clients.ssp_factory.build_ssp_registry") as mock_build: + result = await _resume_ssp_distribution(request, response) + + mock_build.assert_not_called() + assert result == { + "deal_id": "deal-1", + "status": "reject", + "resumed_from_approval": "apr-999", + } + + @pytest.mark.asyncio + async def test_approve_missing_selected_dsp_id_raises_400(self): + request = self._approval_request() + response = ApprovalResponse(approval_id="apr-999", decision="approve", modifications={}) + + with pytest.raises(HTTPException) as exc_info: + await _resume_ssp_distribution(request, response) + + assert exc_info.value.status_code == 400 + + @pytest.mark.asyncio + async def test_approve_with_selected_dsp_id_creates_deal(self): + request = self._approval_request() + response = ApprovalResponse( + approval_id="apr-999", + decision="approve", + modifications={"selected_dsp_id": 52}, + ) + + created_deal = MagicMock() + created_deal.deal_id = "EXT-DEAL-001" + created_deal.ssp_name = "Index Exchange" + created_deal.ssp_type = SSPType.INDEX_EXCHANGE + created_deal.status = MagicMock(value="active") + created_deal.model_dump.return_value = {"deal_id": "EXT-DEAL-001"} + + mock_ssp = AsyncMock() + mock_ssp.create_deal = AsyncMock(return_value=created_deal) + mock_ssp.__aenter__ = AsyncMock(return_value=mock_ssp) + mock_ssp.__aexit__ = AsyncMock(return_value=False) + + mock_registry = MagicMock() + mock_registry.get_client.return_value = mock_ssp + storage = _storage() + + with ( + patch( + "ad_seller.clients.ssp_factory.build_ssp_registry", + return_value=mock_registry, + ), + patch("ad_seller.storage.factory.get_storage", return_value=storage), + ): + result = await _resume_ssp_distribution(request, response) + + mock_registry.get_client.assert_called_once_with("index_exchange") + create_request_arg = mock_ssp.create_deal.call_args.args[0] + assert create_request_arg.dsp_id == 52 + assert create_request_arg.external_deal_id == "EXT-DEAL-001" + assert result["resumed_from_approval"] == "apr-999" + assert result["deal_id"] == "EXT-DEAL-001" + + @pytest.mark.asyncio + async def test_approve_with_selected_dsp_id_outside_candidates_raises_400(self): + request = self._approval_request() + response = ApprovalResponse( + approval_id="apr-999", decision="approve", modifications={"selected_dsp_id": 999} + ) + registry, ssp = self._mock_registry() + + with patch("ad_seller.clients.ssp_factory.build_ssp_registry", return_value=registry): + with pytest.raises(HTTPException) as exc_info: + await _resume_ssp_distribution(request, response) + + assert exc_info.value.status_code == 400 + assert "candidate" in exc_info.value.detail + ssp.create_deal.assert_not_called() + + @pytest.mark.asyncio + async def test_approve_marks_approval_consumed(self): + request = self._approval_request(status=ApprovalStatus.APPROVED) + response = ApprovalResponse( + approval_id="apr-999", decision="approve", modifications={"selected_dsp_id": 52} + ) + registry, _ = self._mock_registry(self._created_deal()) + storage = _storage() + + with ( + patch("ad_seller.clients.ssp_factory.build_ssp_registry", return_value=registry), + patch("ad_seller.storage.factory.get_storage", return_value=storage), + ): + await _resume_ssp_distribution(request, response) + + assert storage._store["approval:apr-999"]["status"] == "consumed" + + @pytest.mark.asyncio + async def test_second_resume_returns_409_and_does_not_create_deal_again(self): + request = self._approval_request(status=ApprovalStatus.APPROVED) + response = ApprovalResponse( + approval_id="apr-999", decision="approve", modifications={"selected_dsp_id": 52} + ) + registry, ssp = self._mock_registry(self._created_deal()) + storage = _storage() + + with ( + patch("ad_seller.clients.ssp_factory.build_ssp_registry", return_value=registry), + patch("ad_seller.storage.factory.get_storage", return_value=storage), + ): + await _resume_ssp_distribution(request, response) + consumed = ApprovalRequest(**storage._store["approval:apr-999"]) + with pytest.raises(HTTPException) as exc_info: + await _resume_ssp_distribution(consumed, response) + + assert exc_info.value.status_code == 409 + assert ssp.create_deal.call_count == 1 + + @pytest.mark.asyncio + async def test_failed_create_deal_restores_approved_status_for_retry(self): + request = self._approval_request(status=ApprovalStatus.APPROVED) + response = ApprovalResponse( + approval_id="apr-999", decision="approve", modifications={"selected_dsp_id": 52} + ) + registry, _ = self._mock_registry(create_side_effect=RuntimeError("IX down")) + storage = _storage() + + with ( + patch("ad_seller.clients.ssp_factory.build_ssp_registry", return_value=registry), + patch("ad_seller.storage.factory.get_storage", return_value=storage), + ): + with pytest.raises(RuntimeError): + await _resume_ssp_distribution(request, response) + + assert storage._store["approval:apr-999"]["status"] == "approved" + + +class TestResumeSSPDistributionHardening: + _approval_request = TestResumeSSPDistribution._approval_request + _mock_registry = TestResumeSSPDistribution._mock_registry + _created_deal = TestResumeSSPDistribution._created_deal + + async def _resume(self, modifications, create_side_effect=None): + request = self._approval_request(status=ApprovalStatus.APPROVED) + response = ApprovalResponse( + approval_id="apr-999", decision="approve", modifications=modifications + ) + registry, ssp = self._mock_registry( + self._created_deal(), create_side_effect=create_side_effect + ) + storage = _storage() + with ( + patch("ad_seller.clients.ssp_factory.build_ssp_registry", return_value=registry), + patch("ad_seller.storage.factory.get_storage", return_value=storage), + ): + try: + result = await _resume_ssp_distribution(request, response) + finally: + self.storage = storage + return result, ssp + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "extra", + [ + {"dsp_id": 999}, + {"external_deal_id": "OTHER"}, + {"cmp": 99}, + {"buyer_seat_ids": ["seat-solo"]}, + ], + ) + async def test_disallowed_or_unknown_modification_keys_return_400(self, extra): + request = self._approval_request(status=ApprovalStatus.APPROVED) + response = ApprovalResponse( + approval_id="apr-999", + decision="approve", + modifications={"selected_dsp_id": 52, **extra}, + ) + registry, ssp = self._mock_registry(self._created_deal()) + with patch("ad_seller.clients.ssp_factory.build_ssp_registry", return_value=registry): + with pytest.raises(HTTPException) as exc_info: + await _resume_ssp_distribution(request, response) + assert exc_info.value.status_code == 400 + assert next(iter(extra)) in exc_info.value.detail + ssp.create_deal.assert_not_called() + + @pytest.mark.asyncio + async def test_allowed_modifications_are_applied(self): + _, ssp = await self._resume({"selected_dsp_id": 52, "cpm": 7.5, "account_id": 4242}) + sent = ssp.create_deal.call_args.args[0] + assert (sent.cpm, sent.account_id) == (7.5, 4242) + + @pytest.mark.asyncio + async def test_invalid_modification_value_returns_400_without_creating(self): + request = self._approval_request(status=ApprovalStatus.APPROVED) + response = ApprovalResponse( + approval_id="apr-999", + decision="approve", + modifications={"selected_dsp_id": 52, "cpm": "not-a-number"}, + ) + registry, ssp = self._mock_registry(self._created_deal()) + with patch("ad_seller.clients.ssp_factory.build_ssp_registry", return_value=registry): + with pytest.raises(HTTPException) as exc_info: + await _resume_ssp_distribution(request, response) + assert exc_info.value.status_code == 400 + ssp.create_deal.assert_not_called() + + @pytest.mark.asyncio + async def test_outcome_unknown_leaves_approval_consumed(self): + from ad_seller.clients.ssp_base import SSPDealOutcomeUnknownError + + with pytest.raises(HTTPException) as exc_info: + await self._resume( + {"selected_dsp_id": 52}, + create_side_effect=SSPDealOutcomeUnknownError("check Index"), + ) + assert exc_info.value.status_code == 502 + assert self.storage._store["approval:apr-999"]["status"] == "outcome_unknown" + + async def _resume_unknown(self, find_result=None, find_side_effect=None): + request = self._approval_request(status=ApprovalStatus.OUTCOME_UNKNOWN) + response = ApprovalResponse( + approval_id="apr-999", decision="approve", modifications={"selected_dsp_id": 52} + ) + registry, ssp = self._mock_registry(self._created_deal()) + ssp.find_deal_by_external_id = AsyncMock( + return_value=find_result, side_effect=find_side_effect + ) + storage = _storage() + self.storage = storage + with ( + patch("ad_seller.clients.ssp_factory.build_ssp_registry", return_value=registry), + patch("ad_seller.storage.factory.get_storage", return_value=storage), + ): + result = await _resume_ssp_distribution(request, response) + return result, ssp + + @pytest.mark.asyncio + async def test_resuming_outcome_unknown_returns_existing_deal_without_posting(self): + result, ssp = await self._resume_unknown(find_result=self._created_deal()) + + assert result["deal_id"] == "EXT-DEAL-001" + ssp.create_deal.assert_not_called() + ssp.find_deal_by_external_id.assert_awaited_once_with("EXT-DEAL-001", 12345) + assert self.storage._store["approval:apr-999"]["status"] == "consumed" + + @pytest.mark.asyncio + async def test_resuming_outcome_unknown_with_absent_deal_restores_approved(self): + with pytest.raises(HTTPException) as exc_info: + await self._resume_unknown(find_result=None) + + assert exc_info.value.status_code == 409 + assert "Resume again" in exc_info.value.detail + assert self.storage._store["approval:apr-999"]["status"] == "approved" + + @pytest.mark.asyncio + async def test_resuming_outcome_unknown_with_failed_lookup_stays_unknown(self): + from ad_seller.clients.ssp_base import SSPDealOutcomeUnknownError + + with pytest.raises(HTTPException) as exc_info: + await self._resume_unknown(find_side_effect=SSPDealOutcomeUnknownError("still down")) + + assert exc_info.value.status_code == 502 + assert "approval:apr-999" not in self.storage._store + + +class TestDecideTimeValidation: + """A bad dsp_resolution decision is rejected while the approval is still pending.""" + + def _stored_request(self, storage): + request = TestResumeSSPDistribution._approval_request(None, status=ApprovalStatus.PENDING) + storage._store["approval:apr-999"] = request.model_dump(mode="json") + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "modifications", + [ + {"selected_dsp_id": 999}, + {"selected_dsp_id": 52, "cpm": "not-a-number"}, + {"selected_dsp_id": 52, "cmp": 99}, + {}, + ], + ) + async def test_bad_decision_returns_400_and_stays_pending(self, modifications): + from ad_seller.services.approval_service import decide_approval + + storage = _storage() + self._stored_request(storage) + with patch("ad_seller.storage.factory.get_storage", return_value=storage): + with pytest.raises(HTTPException) as exc_info: + await decide_approval("apr-999", "approve", modifications=modifications) + + assert exc_info.value.status_code == 400 + assert storage._store["approval:apr-999"]["status"] == "pending" + + @pytest.mark.asyncio + async def test_valid_decision_is_accepted(self): + from ad_seller.services.approval_service import decide_approval + + storage = _storage() + self._stored_request(storage) + with patch("ad_seller.storage.factory.get_storage", return_value=storage): + result = await decide_approval( + "apr-999", "approve", modifications={"selected_dsp_id": 52, "cpm": 9.5} + ) + + assert result["decision"] == "approve" + + +class TestDirectDistributeOutcomeUnknown: + @pytest.mark.asyncio + async def test_outcome_unknown_maps_to_502(self): + from types import SimpleNamespace + + from ad_seller.clients.ssp_base import SSPDealOutcomeUnknownError + from ad_seller.services import deal_service + + client = AsyncMock() + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=False) + client.create_deal = AsyncMock(side_effect=SSPDealOutcomeUnknownError("check Index")) + ssp_registry = MagicMock() + ssp_registry.list_ssps.return_value = ["index_exchange"] + ssp_registry.get_client.return_value = client + storage = AsyncMock() + storage.get_deal = AsyncMock(return_value=None) + request = SimpleNamespace( + deal_id="deal-1", + deal_type="PMP", + name="n", + advertiser=None, + cpm=1.0, + buyer_seat_ids=None, + account_id=1, + start_date=None, + end_date=None, + targeting={"geo": ["US"]}, + ssp_name="index_exchange", + inventory_type=None, + ) + + with ( + patch("ad_seller.clients.ssp_factory.build_ssp_registry", return_value=ssp_registry), + patch("ad_seller.clients.deal_sync_factory.build_deal_sync_registry"), + patch("ad_seller.storage.factory.get_storage", return_value=storage), + patch.object( + deal_service, "_resolve_dsp_id_or_pending_approval", AsyncMock(return_value=None) + ), + ): + with pytest.raises(HTTPException) as exc_info: + await deal_service.distribute_deal_via_ssp(request) + + assert exc_info.value.status_code == 502 + assert exc_info.value.detail["error"] == "outcome_unknown" + assert client.create_deal.call_args.args[0].targeting == {"geo": ["US"]} + + +def test_distribute_request_schema_accepts_dict_and_list_targeting(): + from ad_seller.interfaces.api.schemas import SSPDealDistributeRequest + + assert SSPDealDistributeRequest(deal_id="d", targeting={"geo": ["US"]}).targeting == { + "geo": ["US"] + } + listed = [{"targetingType": "standard", "keyName": "domain", "sets": []}] + assert SSPDealDistributeRequest(deal_id="d", targeting=listed).targeting == listed diff --git a/tests/unit/test_operator_auth.py b/tests/unit/test_operator_auth.py index d7914d88..9358d33f 100644 --- a/tests/unit/test_operator_auth.py +++ b/tests/unit/test_operator_auth.py @@ -1003,6 +1003,42 @@ async def test_http_with_operator_key_is_allowed(self, mock_storage): assert "operator_required" not in result +class TestMcpListPendingApprovalsGated: + """Pending approvals snapshot deal pricing, so listing them is operator-only.""" + + async def test_http_without_credentials_is_denied(self, mock_storage): + from ad_seller.interfaces import mcp_server + + ctx = _fake_mcp_context(headers={}) + with ( + patch.object(mcp_server.mcp, "get_context", return_value=ctx), + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch( + "ad_seller.services.approval_service.list_pending_approvals", + new=AsyncMock(return_value={"approvals": []}), + ) as listing, + ): + result = await mcp_server.list_pending_approvals() + assert "authentication_required" in result + listing.assert_not_called() + + async def test_http_with_operator_key_is_allowed(self, mock_storage): + from ad_seller.interfaces import mcp_server + + op_key = _seed_key(mock_storage._store, role=ApiKeyRole.OPERATOR) + ctx = _fake_mcp_context(headers={"x-api-key": op_key}) + with ( + patch.object(mcp_server.mcp, "get_context", return_value=ctx), + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch( + "ad_seller.services.approval_service.list_pending_approvals", + new=AsyncMock(return_value={"approvals": []}), + ), + ): + result = await mcp_server.list_pending_approvals() + assert "operator_required" not in result + + class TestMcpCreateDealFromTemplateGated: """The MCP twin of POST /api/v1/deals/from-template — unlike the REST route (which accepts any authenticated buyer key), the MCP tool diff --git a/tests/unit/test_ssp_connectors.py b/tests/unit/test_ssp_connectors.py index f16aba60..3d67b709 100644 --- a/tests/unit/test_ssp_connectors.py +++ b/tests/unit/test_ssp_connectors.py @@ -20,6 +20,7 @@ SSPClient, SSPDeal, SSPDealCreateRequest, + SSPDealOutcomeUnknownError, SSPDealStatus, SSPDealType, SSPRegistry, @@ -57,7 +58,7 @@ def test_creates_index_exchange_client(self): settings = _make_settings( ssp_connectors="index_exchange", - index_exchange_api_url="https://api.indexexchange.com", + index_exchange_api_url="https://app.indexexchange.com", index_exchange_api_key="test-key", ) registry = build_ssp_registry(settings) @@ -107,7 +108,7 @@ def test_routing_rules_applied(self): settings = _make_settings( ssp_connectors="index_exchange", - index_exchange_api_url="https://api.indexexchange.com", + index_exchange_api_url="https://app.indexexchange.com", index_exchange_api_key="key", ssp_routing_rules="ctv:index_exchange,display:index_exchange", ) @@ -121,73 +122,755 @@ def test_routing_rules_applied(self): # ========================================================================= +# ========================================================================= +# Index Exchange REST client +# ========================================================================= + + +def _ix_response(**overrides) -> dict: + """A realistic /v3/deals response shape, per the Deals API's schema.""" + base = { + "internalDealID": 987654, + "externalDealID": "IX-DEAL-001", + "classID": 1, + "name": "Test Deal", + "account": {"accountID": 12345}, + "startDate": "2026-01-01", + "endDate": "2026-12-31", + "auctionType": "first", + "floor": 15.0, + "status": "active", + "directConfigurations": { + "dspID": 5551, + "seatIDs": ["seat-100"], + "programmaticGuaranteed": False, + "impressionGoal": 1_000_000, + }, + "labels": {"advertiser": "Test Advertiser"}, + "targeting": [ + { + "targetingType": "standard", + "keyName": "Country", + "sets": [{"values": [{"value": "US"}], "operator": "ANY_OF"}], + } + ], + } + base.update(overrides) + return base + + +def _mock_response(json_data=None, headers=None, status_error=None): + resp = MagicMock() + resp.json.return_value = json_data or {} + resp.headers = headers or {} + if status_error: + resp.raise_for_status.side_effect = status_error + else: + resp.raise_for_status = MagicMock() + return resp + + +def _valid_create_request(**overrides) -> SSPDealCreateRequest: + defaults = dict( + deal_type=SSPDealType.PMP, + name="Test PMP Deal", + advertiser="Test Advertiser", + cpm=15.0, + start_date="2026-01-01", + end_date="2026-12-31", + buyer_seat_ids=["seat-100"], + impressions_goal=1_000_000, + external_deal_id="EXT-DEAL-001", + account_id=12345, + dsp_id=5551, + ) + defaults.update(overrides) + return SSPDealCreateRequest(**defaults) + + class TestIndexExchangeClient: - """Index Exchange REST client formats deal payload correctly.""" + """Index Exchange REST client formats /v3/deals requests/responses correctly.""" + + # --- create_deal() --- @pytest.mark.asyncio - async def test_create_deal_payload_format(self): + async def test_create_deal_pmp_payload_shape(self): client = IndexExchangeSSPClient( - base_url="https://api.indexexchange.com", - api_key="test-key", + base_url="https://app.indexexchange.com", api_key="test-key" ) - # Manually set the http client to a mock - mock_response = MagicMock() - mock_response.json.return_value = { - "deal_id": "ix-deal-001", - "deal_name": "Test Deal", - "deal_type": "pmp", - "status": "active", - "floor_price": 15.0, - "currency": "USD", - } - mock_response.raise_for_status = MagicMock() - mock_http = AsyncMock() - mock_http.post.return_value = mock_response + mock_http.post.return_value = _mock_response(_ix_response()) client._http = mock_http - request = SSPDealCreateRequest( - deal_type=SSPDealType.PMP, - name="Test PMP Deal", - advertiser="Test Advertiser", - cpm=15.0, - currency="USD", - buyer_seat_ids=["seat-100"], - impressions_goal=1_000_000, - ) - + request = _valid_create_request(deal_type=SSPDealType.PMP) result = await client.create_deal(request) - # Verify the call was made with correct IX-specific field names call_args = mock_http.post.call_args - assert call_args[0][0] == "/api/deals" + assert call_args[0][0] == "/api/deals/v3/deals" body = call_args[1]["json"] - assert body["deal_type"] == "pmp" - assert body["deal_name"] == "Test PMP Deal" - assert body["advertiser_name"] == "Test Advertiser" - assert body["floor_price"] == 15.0 - assert body["buyer_seat_ids"] == ["seat-100"] - assert body["impression_goal"] == 1_000_000 - - # Verify parsed result - assert result.deal_id == "ix-deal-001" + + assert body["classID"] == 1 + assert body["name"] == "Test PMP Deal" + assert body["externalDealID"] == "EXT-DEAL-001" + assert body["account"] == {"accountID": 12345} + assert body["startDate"] == "2026-01-01" + assert body["endDate"] == "2026-12-31" + assert body["auctionType"] == "first" + assert body["floor"] == 15.0 + assert body["directConfigurations"] == { + "dspID": 5551, + "programmaticGuaranteed": False, + "seatIDs": ["seat-100"], + "impressionGoal": 1_000_000, + } + assert body["labels"] == {"advertiser": "Test Advertiser"} + + # Old invented field names must be gone + for old_field in ( + "deal_type", + "deal_name", + "advertiser_name", + "floor_price", + "currency", + "start_date", + "end_date", + "buyer_seat_ids", + "impression_goal", + ): + assert old_field not in body + assert result.ssp_type == SSPType.INDEX_EXCHANGE assert result.status == SSPDealStatus.ACTIVE - def test_parse_deal_maps_status(self): - client = IndexExchangeSSPClient( - base_url="https://api.indexexchange.com", + @pytest.mark.asyncio + async def test_create_deal_pg_sets_fixed_auction_and_programmatic_guaranteed(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.post.return_value = _mock_response(_ix_response()) + client._http = mock_http + + await client.create_deal(_valid_create_request(deal_type=SSPDealType.PG)) + + body = mock_http.post.call_args[1]["json"] + assert body["auctionType"] == "fixed" + assert body["directConfigurations"]["programmaticGuaranteed"] is True + + @pytest.mark.asyncio + async def test_create_deal_preferred_sets_fixed_auction_not_guaranteed(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.post.return_value = _mock_response(_ix_response()) + client._http = mock_http + + await client.create_deal(_valid_create_request(deal_type=SSPDealType.PREFERRED)) + + body = mock_http.post.call_args[1]["json"] + assert body["auctionType"] == "fixed" + assert body["directConfigurations"]["programmaticGuaranteed"] is False + + @pytest.mark.asyncio + async def test_create_deal_auction_package_uses_marketplace_configurations(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.post.return_value = _mock_response(_ix_response(classID=4)) + client._http = mock_http + + await client.create_deal(_valid_create_request(deal_type=SSPDealType.AUCTION_PACKAGE)) + + body = mock_http.post.call_args[1]["json"] + assert body["classID"] == 4 + assert body["marketplaceConfigurations"] == {"dspID": 5551} + assert "directConfigurations" not in body + + @pytest.mark.asyncio + async def test_create_deal_missing_external_deal_id_raises(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + client._http = AsyncMock() + + with pytest.raises(ValueError, match="external_deal_id"): + await client.create_deal(_valid_create_request(external_deal_id=None)) + + @pytest.mark.asyncio + async def test_create_deal_missing_account_id_raises(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + client._http = AsyncMock() + + with pytest.raises(ValueError, match="account_id"): + await client.create_deal(_valid_create_request(account_id=None)) + + @pytest.mark.asyncio + async def test_create_deal_missing_dsp_id_raises(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + client._http = AsyncMock() + + with pytest.raises(ValueError, match="dsp_id"): + await client.create_deal(_valid_create_request(dsp_id=None)) + + # --- resolve_dsp_ids_for_seat_ids() --- + + @pytest.mark.asyncio + async def test_resolve_dsp_ids_single_match(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response( + { + "seats": [ + { + "seatID": 6508079, + "name": "Acme", + "buyerID": 10, + "extendedSeatID": "acc-100522", + "status": "A", + "dspID": 17, + } + ] + } ) - deal = client._parse_deal({"deal_id": "d1", "status": "paused"}) - assert deal.status == SSPDealStatus.PAUSED + client._http = mock_http + + matches = await client.resolve_dsp_ids_for_seat_ids(["acc-100522"]) + + call_args = mock_http.get.call_args + assert call_args[0][0] == "/api/deals/v1/dsps/-/seats" + assert call_args[1]["params"] == {"seatIDs": "acc-100522"} + assert len(matches) == 1 + assert matches[0].dsp_id == 17 + assert matches[0].requested_seat_id == "acc-100522" + assert matches[0].extended_seat_id == "acc-100522" + assert matches[0].status == "A" + + @pytest.mark.asyncio + async def test_resolve_dsp_ids_multiple_distinct_dsps(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response( + { + "seats": [ + {"seatID": 1, "extendedSeatID": "acc-dupe", "status": "A", "dspID": 17}, + {"seatID": 2, "extendedSeatID": "acc-dupe", "status": "A", "dspID": 52}, + ] + } + ) + client._http = mock_http + + matches = await client.resolve_dsp_ids_for_seat_ids(["acc-dupe"]) + + assert {m.dsp_id for m in matches} == {17, 52} + + @pytest.mark.asyncio + async def test_resolve_dsp_ids_filters_out_inactive(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response( + { + "seats": [ + {"seatID": 1, "extendedSeatID": "acc-1", "status": "A", "dspID": 17}, + {"seatID": 2, "extendedSeatID": "acc-1", "status": "D", "dspID": 99}, + ] + } + ) + client._http = mock_http + + matches = await client.resolve_dsp_ids_for_seat_ids(["acc-1"]) + + assert len(matches) == 1 + assert matches[0].dsp_id == 17 + + @pytest.mark.asyncio + async def test_resolve_dsp_ids_no_match_returns_empty(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response({"seats": []}) + client._http = mock_http + + matches = await client.resolve_dsp_ids_for_seat_ids(["unknown-seat"]) + + assert matches == [] + + @pytest.mark.asyncio + async def test_resolve_dsp_ids_empty_input_short_circuits(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + client._http = mock_http + + matches = await client.resolve_dsp_ids_for_seat_ids([]) + + assert matches == [] + mock_http.get.assert_not_called() + + @pytest.mark.asyncio + async def test_resolve_dsp_ids_queries_each_seat_separately(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + + async def _get(_path, params): + dsp = {"acc-1": 17, "acc-2": 52}[params["seatIDs"]] + return _mock_response( + { + "seats": [ + { + "seatID": 1, + "extendedSeatID": params["seatIDs"].upper(), + "status": "A", + "dspID": dsp, + } + ] + } + ) + + mock_http.get.side_effect = _get + client._http = mock_http + + matches = await client.resolve_dsp_ids_for_seat_ids(["acc-1", "acc-2"]) + + assert [c[1]["params"] for c in mock_http.get.call_args_list] == [ + {"seatIDs": "acc-1"}, + {"seatIDs": "acc-2"}, + ] + assert {(m.requested_seat_id, m.dsp_id) for m in matches} == { + ("acc-1", 17), + ("acc-2", 52), + } + + # --- clone_deal() --- + + @pytest.mark.asyncio + async def test_clone_deal_composes_get_and_create(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response(_ix_response()) + mock_http.post.return_value = _mock_response(_ix_response()) + client._http = mock_http + + await client.clone_deal("987654") + + assert mock_http.get.call_args[0][0] == "/api/deals/v3/deals/987654" + assert mock_http.post.call_args[0][0] == "/api/deals/v3/deals" + body = mock_http.post.call_args[1]["json"] + assert body["externalDealID"] != "IX-DEAL-001" + assert body["externalDealID"].startswith("IX-DEAL-001-clone-") + assert body["account"] == {"accountID": 12345} + assert body["directConfigurations"]["dspID"] == 5551 + + @pytest.mark.asyncio + async def test_clone_deal_marketplace_package_reads_dsp_from_marketplace_configurations(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response( + _ix_response( + classID=4, + directConfigurations={}, + marketplaceConfigurations={"dspID": 777}, + ) + ) + mock_http.post.return_value = _mock_response(_ix_response(classID=4)) + client._http = mock_http + + await client.clone_deal("987654") + + body = mock_http.post.call_args[1]["json"] + assert body["classID"] == 4 + assert body["marketplaceConfigurations"] == {"dspID": 777} + + @pytest.mark.asyncio + async def test_clone_deal_applies_overrides(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response(_ix_response()) + mock_http.post.return_value = _mock_response(_ix_response()) + client._http = mock_http + + await client.clone_deal( + "987654", overrides={"external_deal_id": "MY-CUSTOM-ID", "name": "Custom Name"} + ) + + body = mock_http.post.call_args[1]["json"] + assert body["externalDealID"] == "MY-CUSTOM-ID" + assert body["name"] == "Custom Name" + + @pytest.mark.asyncio + async def test_clone_deal_generated_id_never_exceeds_64_chars(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + long_source_id = "a" * 64 # already at IX's max length + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response(_ix_response(externalDealID=long_source_id)) + mock_http.post.return_value = _mock_response(_ix_response()) + client._http = mock_http + + await client.clone_deal("987654") + + body = mock_http.post.call_args[1]["json"] + assert len(body["externalDealID"]) <= 64 + assert body["externalDealID"] != long_source_id + assert "-clone-" in body["externalDealID"] + + # --- get_deal() --- + + @pytest.mark.asyncio + async def test_get_deal_uses_v3_path(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response(_ix_response()) + client._http = mock_http + + await client.get_deal("987654") + + assert mock_http.get.call_args[0][0] == "/api/deals/v3/deals/987654" + + # --- list_deals() --- + + @pytest.mark.asyncio + async def test_list_deals_uses_pageSize_pageOffset(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response({"totalCount": 1, "deals": [_ix_response()]}) + client._http = mock_http + + deals = await client.list_deals(limit=50) + + assert mock_http.get.call_args[0][0] == "/api/deals/v3/deals" + params = mock_http.get.call_args[1]["params"] + assert params["pageSize"] == 50 + assert params["pageOffset"] == 0 + assert "limit" not in params + assert len(deals) == 1 + + @pytest.mark.parametrize( + "status,expected_param", + [ + (SSPDealStatus.ACTIVE, "active"), + (SSPDealStatus.PAUSED, "paused"), + (SSPDealStatus.EXPIRED, "expired"), + ], + ) + @pytest.mark.asyncio + async def test_list_deals_status_mapping(self, status, expected_param): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response({"totalCount": 0, "deals": []}) + client._http = mock_http + + await client.list_deals(status=status) + + assert mock_http.get.call_args[1]["params"]["status"] == expected_param + + @pytest.mark.parametrize("status", [SSPDealStatus.CREATED, SSPDealStatus.ARCHIVED]) + @pytest.mark.asyncio + async def test_list_deals_rejects_status_index_cannot_filter_on(self, status): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + client._http = mock_http + + with pytest.raises(ValueError, match="no '.*' deal status"): + await client.list_deals(status=status) + + mock_http.get.assert_not_called() + + # --- update_deal() --- + + @pytest.mark.asyncio + async def test_update_deal_uses_patch_with_if_match_etag(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response(_ix_response(), headers={"ETag": '"etag-123"'}) + mock_http.patch.return_value = _mock_response(_ix_response(status="paused")) + client._http = mock_http + + await client.update_deal("987654", {"status": "paused"}) + + assert mock_http.get.call_args[0][0] == "/api/deals/v3/deals/987654" + assert mock_http.patch.call_args[0][0] == "/api/deals/v3/deals/987654" + assert mock_http.patch.call_args[1]["json"] == {"status": "paused"} + assert mock_http.patch.call_args[1]["headers"] == {"If-Match": '"etag-123"'} + + @pytest.mark.asyncio + async def test_update_deal_uses_caller_held_etag_without_refetching(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.patch.return_value = _mock_response(_ix_response(status="paused")) + client._http = mock_http + + await client.update_deal("987654", {"status": "paused"}, etag='"held-etag"') + + mock_http.get.assert_not_called() + assert mock_http.patch.call_args[1]["headers"] == {"If-Match": '"held-etag"'} + + @pytest.mark.asyncio + async def test_get_deal_exposes_etag_for_later_update(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response(_ix_response(), headers={"ETag": '"etag-9"'}) + client._http = mock_http + + deal = await client.get_deal("987654") + + assert deal.raw["_etag"] == '"etag-9"' + + @pytest.mark.asyncio + async def test_update_deal_raises_without_etag(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response(_ix_response(), headers={}) + client._http = mock_http + + with pytest.raises(ValueError, match="ETag"): + await client.update_deal("987654", {"status": "paused"}) + + @pytest.mark.asyncio + async def test_update_deal_412_propagates(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response(_ix_response(), headers={"ETag": '"stale"'}) + mock_http.patch.return_value = _mock_response( + status_error=httpx.HTTPStatusError( + "Precondition Failed", request=MagicMock(), response=MagicMock(status_code=412) + ) + ) + client._http = mock_http + + with pytest.raises(httpx.HTTPStatusError): + await client.update_deal("987654", {"status": "paused"}) + + # --- _parse_deal() --- + + def test_parse_deal_maps_fields(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + deal = client._parse_deal(_ix_response()) + + assert deal.deal_id == "987654" + assert deal.external_deal_id == "IX-DEAL-001" + assert deal.cpm == 15.0 + assert deal.start_date == "2026-01-01" + assert deal.end_date == "2026-12-31" + assert deal.impressions_goal == 1_000_000 + assert deal.advertiser == "Test Advertiser" + assert deal.currency == "USD" + assert deal.targeting == _ix_response()["targeting"] + + @pytest.mark.parametrize( + "status_str,expected", + [ + ("active", SSPDealStatus.ACTIVE), + ("paused", SSPDealStatus.PAUSED), + ("expired", SSPDealStatus.EXPIRED), + ("auto-paused", SSPDealStatus.PAUSED), + ], + ) + def test_parse_deal_status_map(self, status_str, expected): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + deal = client._parse_deal(_ix_response(status=status_str)) + assert deal.status == expected + + def test_parse_deal_unknown_status_maps_to_error(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + deal = client._parse_deal(_ix_response(status="archived")) + assert deal.status == SSPDealStatus.ERROR + + def test_status_map_has_no_archived_or_pending(self): + from ad_seller.clients.ssp_index_exchange import _IX_STATUS_MAP + + assert "archived" not in _IX_STATUS_MAP + assert "pending" not in _IX_STATUS_MAP def test_parse_deal_maps_pg_type(self): - client = IndexExchangeSSPClient( - base_url="https://api.indexexchange.com", + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + deal = client._parse_deal( + _ix_response( + classID=1, + auctionType="fixed", + directConfigurations={"programmaticGuaranteed": True}, + ) ) - deal = client._parse_deal({"deal_id": "d1", "deal_type": "programmatic_guaranteed"}) assert deal.deal_type == SSPDealType.PG + def test_parse_deal_maps_pmp_type(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + deal = client._parse_deal( + _ix_response( + classID=1, + auctionType="first", + directConfigurations={"programmaticGuaranteed": False}, + ) + ) + assert deal.deal_type == SSPDealType.PMP + + def test_parse_deal_maps_preferred_type(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + deal = client._parse_deal( + _ix_response( + classID=1, + auctionType="fixed", + directConfigurations={"programmaticGuaranteed": False}, + ) + ) + assert deal.deal_type == SSPDealType.PREFERRED + + def test_parse_deal_maps_auction_package_type(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + deal = client._parse_deal(_ix_response(classID=4, directConfigurations={})) + assert deal.deal_type == SSPDealType.AUCTION_PACKAGE + + # --- troubleshoot_deal() --- + + @pytest.mark.parametrize( + "status_str,expected_score", + [("active", 90), ("paused", 50), ("auto-paused", 20), ("expired", 0)], + ) + @pytest.mark.asyncio + async def test_troubleshoot_deal_health_score_by_status(self, status_str, expected_score): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response(_ix_response(status=status_str)) + client._http = mock_http + + result = await client.troubleshoot_deal("987654") + + assert result.health_score == expected_score + assert result.deal_id == "IX-DEAL-001" + + @pytest.mark.asyncio + async def test_troubleshoot_deal_flags_missing_seat_ids(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.return_value = _mock_response( + _ix_response(directConfigurations={"dspID": 5551, "seatIDs": []}) + ) + client._http = mock_http + + result = await client.troubleshoot_deal("987654") + + assert any("seat" in issue.lower() for issue in result.primary_issues) + + @pytest.mark.asyncio + async def test_troubleshoot_deal_api_failure_returns_unreachable(self): + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com") + mock_http = AsyncMock() + mock_http.get.side_effect = httpx.ConnectError("connection refused") + client._http = mock_http + + result = await client.troubleshoot_deal("987654") + + assert result.status == "unreachable" + assert result.health_score == 0 + + +class TestIndexExchangeCreateDealAmbiguousFailure: + """A POST that may have reached Index must not be blindly retried.""" + + @staticmethod + def _client(handler) -> IndexExchangeSSPClient: + client = IndexExchangeSSPClient(base_url="https://app.indexexchange.com", api_key="k") + client._http = httpx.AsyncClient( + base_url="https://app.indexexchange.com", transport=httpx.MockTransport(handler) + ) + return client + + @pytest.mark.asyncio + async def test_timeout_after_post_returns_deal_already_created(self): + calls = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append(request) + if request.method == "POST": + raise httpx.ReadTimeout("lost response", request=request) + return httpx.Response( + 200, + json={ + "deals": [ + _ix_response(externalDealID="EXT-DEAL-001-OTHER"), + _ix_response(externalDealID="EXT-DEAL-001"), + ] + }, + ) + + client = self._client(handler) + result = await client.create_deal(_valid_create_request()) + + assert result.external_deal_id == "EXT-DEAL-001" + assert [c.method for c in calls] == ["POST", "GET"] + assert calls[1].url.params["search"] == "EXT-DEAL-001" + assert calls[1].url.params["accountIDs"] == "12345" + + @pytest.mark.asyncio + async def test_lookup_pages_past_substring_matches_to_find_exact_deal(self): + pages = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST": + raise httpx.ReadTimeout("lost response", request=request) + offset = int(request.url.params["pageOffset"]) + pages.append(offset) + if offset == 0: + decoys = [_ix_response(externalDealID=f"EXT-DEAL-001-{i}") for i in range(100)] + return httpx.Response(200, json={"deals": decoys}) + return httpx.Response( + 200, json={"deals": [_ix_response(externalDealID="EXT-DEAL-001")]} + ) + + result = await self._client(handler).create_deal(_valid_create_request()) + + assert result.external_deal_id == "EXT-DEAL-001" + assert pages == [0, 1] + + @pytest.mark.asyncio + async def test_lookup_without_exact_match_within_page_cap_is_outcome_unknown(self): + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST": + raise httpx.ReadTimeout("lost response", request=request) + decoys = [_ix_response(externalDealID=f"EXT-DEAL-001-{i}") for i in range(100)] + return httpx.Response(200, json={"deals": decoys}) + + with pytest.raises(SSPDealOutcomeUnknownError): + await self._client(handler).create_deal(_valid_create_request()) + + @pytest.mark.asyncio + async def test_timeout_after_post_reraises_when_deal_absent(self): + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST": + raise httpx.ReadTimeout("lost response", request=request) + return httpx.Response(200, json={"deals": [_ix_response(externalDealID="NOPE")]}) + + with pytest.raises(httpx.ReadTimeout): + await self._client(handler).create_deal(_valid_create_request()) + + @pytest.mark.asyncio + async def test_5xx_then_failed_lookup_raises_outcome_unknown(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(503, json={}) + + with pytest.raises(SSPDealOutcomeUnknownError): + await self._client(handler).create_deal(_valid_create_request()) + + @pytest.mark.asyncio + @pytest.mark.parametrize("status", [400, 409, 422]) + async def test_4xx_does_not_look_up_or_swallow(self, status): + calls = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append(request.method) + return httpx.Response(status, json={}) + + with pytest.raises(httpx.HTTPStatusError): + await self._client(handler).create_deal(_valid_create_request()) + assert calls == ["POST"] + + @pytest.mark.asyncio + async def test_connect_error_does_not_look_up(self): + calls = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append(request.method) + raise httpx.ConnectError("refused", request=request) + + with pytest.raises(httpx.ConnectError): + await self._client(handler).create_deal(_valid_create_request()) + assert calls == ["POST"] + + def test_create_request_accepts_dict_and_list_targeting(self): + assert _valid_create_request(targeting={"geo": ["US"]}).targeting == {"geo": ["US"]} + listed = [{"targetingType": "standard", "keyName": "domain", "sets": []}] + assert _valid_create_request(targeting=listed).targeting == listed + # ========================================================================= # SSP Error handling @@ -217,24 +900,27 @@ async def test_connect_without_url_raises(self): @pytest.mark.asyncio async def test_http_error_propagates(self): client = IndexExchangeSSPClient( - base_url="https://api.indexexchange.com", + base_url="https://app.indexexchange.com", api_key="test-key", ) mock_response = MagicMock() mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( - "Server Error", request=MagicMock(), response=MagicMock(status_code=500) + "Bad Request", request=MagicMock(), response=MagicMock(status_code=400) ) mock_http = AsyncMock() mock_http.post.return_value = mock_response client._http = mock_http + request = SSPDealCreateRequest( + deal_type=SSPDealType.PMP, + name="Test Deal", + cpm=15.0, + external_deal_id="EXT-DEAL-001", + account_id=12345, + dsp_id=5551, + ) with pytest.raises(httpx.HTTPStatusError): - await client.create_deal(SSPDealCreateRequest()) - - -# ========================================================================= -# SSP Registry (deal distribution) -# ========================================================================= + await client.create_deal(request) class _MockSSPClient(SSPClient): @@ -327,3 +1013,9 @@ def test_get_nonexistent_client_raises(self): registry = SSPRegistry() with pytest.raises(KeyError): registry.get_client("nonexistent") + + +def test_index_exchange_api_url_has_no_default(): + from ad_seller.config.settings import Settings + + assert Settings.model_fields["index_exchange_api_url"].default is None