From ce5e773e9c146121067069b03f867d0099440ab4 Mon Sep 17 00:00:00 2001 From: Daniel Cox Date: Wed, 22 Apr 2026 16:01:31 -0600 Subject: [PATCH 1/2] Add SafeGuard Privacy vendor approval gate for deal requests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces an optional integration with the SafeGuard Privacy (SGP) platform that lets buyers gate Deal ID generation on the `iabBuyerAgentApproval` flag they maintain for each vendor in their SGP tenant. The integration is off by default — when SGP_API_KEY is empty the buyer agent behaves exactly as before. New components - SGPClient: async httpx client for the SGP buyer-agent-approval endpoint. Domain normalization, dedupe, batching to 10 per request, TTL cache, and full HTTP status handling. Transport errors are wrapped as SGPClientError so the deal-request gate fails closed. - ApprovalRecord pydantic model mirroring IabBuyerAgentResource. - SGPVendorApprovalTool: CrewAI tool wired into the Buyer Deal Specialist so the agent can consult approval status during product selection, not just at Deal ID generation. Class is SGP-prefixed to leave room for future vendor-approval sources. Integration points - DiscoverInventoryTool annotates each product row with APPROVED / NOT APPROVED / UNKNOWN when an SGPClient is provided. Discovery fails open on SGP transport errors so outages do not break browsing. - RequestDealTool adds a pre-flight gate. When SGP_ENFORCE_ON_DEAL_REQUEST is true and an SGPClient is configured, Deal IDs are not issued for unapproved vendors. Unknown-vendor behavior is governed by SGP_UNKNOWN_VENDOR_POLICY (block | warn | allow; default block). - BuyerDealFlow auto-instantiates the client from settings and logs a warning if enforcement is configured without an API key. Configuration - SGP_API_KEY, SGP_BASE_URL, SGP_ENFORCE_ON_DEAL_REQUEST, SGP_UNKNOWN_VENDOR_POLICY, SGP_CACHE_TTL_SECONDS. Defaults point at the production SGP endpoint; staging is noted inline. Docs - docs/integration/safeguard-privacy.md covers endpoint contract, config, behavior matrix, and troubleshooting. Added to mkdocs nav, the configuration guide, and the README Key Features section. Tests - 31 new unit tests covering the client (normalization, batching, all HTTP statuses, transport errors, caching), the gate (approved, denied, unknown policies, fail-closed on errors, missing seller domain), and flow-level tool wiring. --- .env.example | 9 + README.md | 6 + docs/guides/configuration.md | 16 + docs/integration/safeguard-privacy.md | 160 ++++++++++ mkdocs.yml | 1 + src/ad_buyer/clients/__init__.py | 5 + src/ad_buyer/clients/sgp_client.py | 244 ++++++++++++++ src/ad_buyer/config/settings.py | 14 + src/ad_buyer/flows/buyer_deal_flow.py | 43 ++- src/ad_buyer/models/sgp.py | 31 ++ .../tools/buyer_deals/discover_inventory.py | 67 +++- .../tools/buyer_deals/request_deal.py | 118 ++++++- src/ad_buyer/tools/research/__init__.py | 3 +- .../tools/research/sgp_vendor_approval.py | 96 ++++++ tests/unit/test_sgp_client.py | 273 ++++++++++++++++ tests/unit/test_sgp_gate.py | 298 ++++++++++++++++++ 16 files changed, 1376 insertions(+), 8 deletions(-) create mode 100644 docs/integration/safeguard-privacy.md create mode 100644 src/ad_buyer/clients/sgp_client.py create mode 100644 src/ad_buyer/models/sgp.py create mode 100644 src/ad_buyer/tools/research/sgp_vendor_approval.py create mode 100644 tests/unit/test_sgp_client.py create mode 100644 tests/unit/test_sgp_gate.py diff --git a/.env.example b/.env.example index 155c3d7..3c10d5b 100644 --- a/.env.example +++ b/.env.example @@ -24,3 +24,12 @@ REDIS_URL= # Environment ENVIRONMENT=development LOG_LEVEL=INFO + +# SafeGuard Privacy — IAB buyer-agent approval (optional; inert when SGP_API_KEY is empty) +SGP_API_KEY= +# Staging Environment: https://api.safeguardprivacy-demo.com +SGP_BASE_URL=https://api.safeguardprivacy.com +SGP_ENFORCE_ON_DEAL_REQUEST=false +# block | warn | allow +SGP_UNKNOWN_VENDOR_POLICY=block +SGP_CACHE_TTL_SECONDS=900 diff --git a/README.md b/README.md index 885a081..7930f71 100644 --- a/README.md +++ b/README.md @@ -118,6 +118,12 @@ Reveal buyer identity progressively to unlock better pricing from sellers: → [Authentication Guide](https://iabtechlab.github.io/buyer-agent/api/authentication/) +### Vendor Approval Gating (optional) + +Plug in a [SafeGuard Privacy](https://safeguardprivacy.com) tenant to block Deal IDs for sellers the buyer has not approved in their SGP vendor portfolio. Consults the `iabBuyerAgentApproval` flag via SGP's integration API; discovery annotates each product with APPROVED / NOT APPROVED / UNKNOWN, and `RequestDealTool` refuses to generate a Deal ID for unapproved vendors. Off by default — inert when `SGP_API_KEY` is empty. + +→ [IAB Buyer-Agent Approval](https://iabtechlab.github.io/buyer-agent/integration/safeguard-privacy/) + ## Quick Start ### Install diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md index 2f908bb..01cae51 100644 --- a/docs/guides/configuration.md +++ b/docs/guides/configuration.md @@ -175,6 +175,22 @@ CORS_ALLOWED_ORIGINS=https://dashboard.example.com,https://app.example.com --- +### SafeGuard Privacy (IAB Buyer-Agent Approval) + +Optional integration that gates deal requests against the buyer's [SafeGuard Privacy](https://safeguardprivacy.com) vendor portfolio. Inert when `SGP_API_KEY` is empty. + +| Variable | Type | Default | Description | +|----------|------|---------|-------------| +| `SGP_API_KEY` | `str` | `""` | API key with the `iab:buyerAgent` scope. Empty = integration disabled. | +| `SGP_BASE_URL` | `str` | `https://api.safeguardprivacy.com` | SGP base URL. Staging: `https://api.safeguardprivacy-demo.com`. | +| `SGP_ENFORCE_ON_DEAL_REQUEST` | `bool` | `False` | When `True`, `RequestDealTool` blocks Deal ID generation for unapproved vendors. | +| `SGP_UNKNOWN_VENDOR_POLICY` | `str` | `block` | Behavior when the vendor is not in the buyer's SGP portfolio (HTTP 404). One of `block`, `warn`, `allow`. | +| `SGP_CACHE_TTL_SECONDS` | `int` | `900` | Per-domain cache lifetime for approval lookups. | + +See the [IAB Buyer-Agent Approval](../integration/safeguard-privacy.md) integration guide for endpoint contract, behavior matrix, and troubleshooting. + +--- + ### Environment | Variable | Type | Default | Description | diff --git a/docs/integration/safeguard-privacy.md b/docs/integration/safeguard-privacy.md new file mode 100644 index 0000000..328b4d4 --- /dev/null +++ b/docs/integration/safeguard-privacy.md @@ -0,0 +1,160 @@ +# IAB Buyer-Agent Approval (via SafeGuard Privacy) + +The buyer agent can verify, before issuing a Deal ID, that the buyer has explicitly approved a seller's vendor record for IAB buyer-agent purchases. Approvals are stored in the buyer's [SafeGuard Privacy](https://safeguardprivacy.com) tenant; the buyer agent consults them through SGP's integration API. + +This integration is **optional and off by default**. When `SGP_API_KEY` is empty the feature is fully inert — the buyer agent behaves exactly as it did before this page existed. Once configured, it acts as a privacy rail in front of the existing deal workflow. + +## Who should enable this + +SafeGuard Privacy customers who treat vendor onboarding and approval as a compliance prerequisite for programmatic buying. If your team already maintains a vendor inventory in SGP with IAB buyer-agent approval flags, this integration enforces that workflow inside the buyer agent itself. + +## Endpoint contract + +The client calls a single endpoint on the SafeGuard Privacy platform: + +``` +GET /api/v1/integrations/iab/buyer-agent-approval?domain=a.com,b.com +``` + +| Property | Value | +|----------|-------| +| Auth | `api-key` header | +| Scope | `iab:buyerAgent` | +| Batch size | Up to 10 domains per request | +| Tenant scope | Results are scoped to the caller's SGP `companyId` | + +The response contains one `IabBuyerAgentResource` per matched vendor: + +```json +{ + "status": "success", + "code": 200, + "data": [ + { + "vendorId": 123, + "vendorCompanyId": 456, + "companyName": "Example Publisher", + "domain": "example.com", + "internalId": "", + "iabBuyerAgentApproval": true, + "iabBuyerAgentApprovedAt": "2026-03-14T12:00:00Z" + } + ] +} +``` + +Three response states matter to the buyer agent: + +| State | Meaning | How the gate treats it | +|-------|---------|------------------------| +| `iabBuyerAgentApproval: true` | Buyer has approved this vendor | ✅ Deal proceeds | +| `iabBuyerAgentApproval: false` | Vendor exists but is not approved | ❌ Deal blocked | +| HTTP 404 | Vendor is not in the buyer's SGP portfolio | Governed by `SGP_UNKNOWN_VENDOR_POLICY` | + +## Configuration + +| Variable | Type | Default | Description | +|----------|------|---------|-------------| +| `SGP_API_KEY` | `str` | `""` | API key with the `iab:buyerAgent` scope. Empty = integration disabled. | +| `SGP_BASE_URL` | `str` | `https://api.safeguardprivacy.com` | Production endpoint. The staging environment is `https://api.safeguardprivacy-demo.com`. | +| `SGP_ENFORCE_ON_DEAL_REQUEST` | `bool` | `False` | When `True`, `RequestDealTool` blocks Deal ID generation unless the seller's vendor is approved. | +| `SGP_UNKNOWN_VENDOR_POLICY` | `str` | `"block"` | Behavior for domains not in the SGP portfolio (HTTP 404). One of `block`, `warn`, `allow`. | +| `SGP_CACHE_TTL_SECONDS` | `int` | `900` | Per-domain cache lifetime. Discovery→pricing→booking reuse a single SGP call within the TTL. | + +!!! warning "Enforcement without a key is a no-op" + If `SGP_ENFORCE_ON_DEAL_REQUEST=true` but `SGP_API_KEY` is empty, the gate cannot be evaluated and is silently bypassed. The buyer agent logs a warning at flow construction time so this misconfiguration is visible. + +## Where the gate runs + +The integration plugs into two existing buyer-agent tools: + +### Inventory discovery annotations + +`DiscoverInventoryTool` accepts an optional `SGPClient`. When provided, it extracts the seller domain from each returned product (checking `seller_url`, `publisher_domain`, then `publisherId`/`publisher` if they contain a `.`), batches distinct domains into groups of 10, and annotates each product row in the formatted output: + +``` +1. Premium CTV - Sports + Product ID: ctv-premium-sports + Publisher: premium-pub-001 + CPM: $28.26 (was $35.00) + SGP Approval: ✓ APPROVED — seller.example.com +``` + +Discovery **fails open** on SGP transport errors — the tool logs and continues without annotations, so a SafeGuard outage never breaks inventory browsing. The actual enforcement is always at the deal-request step. + +### Deal-request gate + +`RequestDealTool` checks the seller's vendor approval after fetching product details and before generating a Deal ID. The gate runs only when an `SGPClient` is wired in and `sgp_enforce=True`: + +```python +# Injected automatically by BuyerDealFlow from settings +RequestDealTool( + client=unified_client, + buyer_context=ctx, + sgp_client=sgp_client, + sgp_enforce=settings.sgp_enforce_on_deal_request, + sgp_unknown_policy=settings.sgp_unknown_vendor_policy, +) +``` + +A successful gate prepends a banner to the Deal ID response: + +``` +SGP: ✓ Example Publisher approved for IAB buyer-agent purchases (since 2026-03-14T12:00:00Z). + +============================================================ +DEAL CREATED SUCCESSFULLY +============================================================ +... +``` + +A failed gate returns a blocking message and does **not** generate a Deal ID. + +## Behavior matrix + +With enforcement on (`SGP_ENFORCE_ON_DEAL_REQUEST=true`, `SGP_API_KEY` set): + +| SGP response | `block` policy | `warn` policy | `allow` policy | +|---|---|---|---| +| `iabBuyerAgentApproval: true` | ✅ deal proceeds + banner | same | same | +| `iabBuyerAgentApproval: false` | ❌ blocked | ❌ blocked | ❌ blocked | +| 404 (not onboarded in SGP) | ❌ blocked | ✅ proceeds + warning banner | ✅ proceeds silently | +| Transport error | ❌ fails closed | ❌ fails closed | ❌ fails closed | +| Product has no seller domain field | ❌ blocked (cannot evaluate) | ❌ | ❌ | + +The `iabBuyerAgentApproval: false` row is intentionally the same across all three unknown-vendor policies — an explicit non-approval is always fatal. The policies only govern the "unknown to SGP" case. + +## Agent tool + +For CrewAI agents that want to consult approvals outside the automatic gate, a tool is provided: + +```python +from ad_buyer.clients import SGPClient +from ad_buyer.tools.research import SGPVendorApprovalTool + +sgp = SGPClient(api_key=settings.sgp_api_key, base_url=settings.sgp_base_url) +tool = SGPVendorApprovalTool(client=sgp) + +# Agent calls it with a list of domains (any number; client chunks to 10) +# Returns a formatted APPROVED / NOT APPROVED / UNKNOWN summary. +``` + +`BuyerDealFlow` injects this tool into the Buyer Deal Specialist automatically when an SGP client is configured, so the agent can consult approval status during product selection (before commitment), not only at Deal ID generation time. + +The class is prefixed `SGP` so future vendor-approval integrations can coexist under their own class names and CrewAI `name` attributes without colliding. + +## Troubleshooting + +| Symptom | Likely cause | +|---------|-------------| +| `SafeGuard Privacy rejected the api-key` (401) | The key is missing, revoked, or lacks the `iab:buyerAgent` scope. Issue a new key in SGP with that scope. | +| `Deal blocked: is not in your SafeGuard Privacy portfolio` | The vendor is not onboarded in SGP. Add and approve the vendor in SGP, or switch `SGP_UNKNOWN_VENDOR_POLICY` to `warn` for soft-fail behavior. | +| `Deal blocked: does not carry the IAB buyer-agent approval flag` | The vendor is onboarded but not marked approved for IAB buyer-agent purchases. Toggle the approval in SGP. | +| `Deal blocked: SafeGuard Privacy lookup failed` | SGP was unreachable or returned a transient error. Enforcement fails closed; retry once the service is reachable. | +| Gate seems to do nothing | Either `SGP_API_KEY` is empty or `SGP_ENFORCE_ON_DEAL_REQUEST=false`. Check startup logs for the bypass warning. | + +## Related + +- [Configuration reference](../guides/configuration.md) — all env vars including SGP +- [Buyer Deal Flow](../architecture/buyer-deal-flow.md) — the flow the gate plugs into +- [Seller Agent Integration](seller-agent.md) — the seller side of the deal request diff --git a/mkdocs.yml b/mkdocs.yml index e34225b..df70a0a 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -104,6 +104,7 @@ nav: - Integration: - Seller Agent Guide: integration/seller-agent.md - OpenDirect Protocol: integration/opendirect.md + - IAB Buyer-Agent Approval: integration/safeguard-privacy.md - AI Assistant Setup: - Claude (Desktop & Web): claude-desktop-setup.md - ChatGPT, Codex & AI IDEs: multi-client-setup.md diff --git a/src/ad_buyer/clients/__init__.py b/src/ad_buyer/clients/__init__.py index 1c79355..2029639 100644 --- a/src/ad_buyer/clients/__init__.py +++ b/src/ad_buyer/clients/__init__.py @@ -7,6 +7,7 @@ from .deals_client import DealsClient, DealsClientError from .mcp_client import IABMCPClient, MCPClientError, MCPToolResult from .opendirect_client import OpenDirectClient +from .sgp_client import SGPAuthError, SGPClient, SGPClientError from .ucp_client import UCPClient, UCPExchangeResult from .unified_client import Protocol, UnifiedClient, UnifiedResult @@ -31,4 +32,8 @@ # IAB Deals API v1.0 client (quote-then-book flow) "DealsClient", "DealsClientError", + # SafeGuard Privacy (SGP) approval gate + "SGPClient", + "SGPClientError", + "SGPAuthError", ] diff --git a/src/ad_buyer/clients/sgp_client.py b/src/ad_buyer/clients/sgp_client.py new file mode 100644 index 0000000..25e83bf --- /dev/null +++ b/src/ad_buyer/clients/sgp_client.py @@ -0,0 +1,244 @@ +# Author: SafeGuard Privacy +# Donated to IAB Tech Lab + +"""SafeGuard Privacy (SGP) platform client. + +Async HTTP client for the SafeGuard Privacy integration API. Currently +exposes a single capability: checking whether a vendor has the IAB +buyer-agent approval flag set on the buyer's SGP tenant. + +Endpoint: + GET /api/v1/integrations/iab/buyer-agent-approval?domain=a.com,b.com + +Auth: api-key header, scope `iab:buyerAgent`. +Limit: up to 10 domains/internalIds per call (SGP-enforced). +""" + +from __future__ import annotations + +import logging +import time +from typing import Optional +from urllib.parse import urlparse + +import httpx + +from ..models.sgp import ApprovalRecord + +logger = logging.getLogger(__name__) + +_DEFAULT_TIMEOUT = 15.0 +_MAX_BATCH = 10 +_RETRYABLE_STATUS_CODES = {502, 503, 504} +_ENDPOINT = "/api/v1/integrations/iab/buyer-agent-approval" + +# Ordered list of product-dict keys to probe when deriving a seller domain +# for an SGP approval lookup. Explicit domain fields come first; publisher +# identifiers are used only when they look like a hostname. +_DOMAIN_KEYS = ("seller_url", "sellerUrl", "publisher_domain", "publisherDomain") +_PUBLISHER_KEYS = ("publisherId", "publisher") + + +def extract_product_domain(product: dict) -> Optional[str]: + """Best-guess seller domain from a product dict for an SGP lookup. + + Checks explicit domain/URL fields first, then falls back to + ``publisherId`` / ``publisher`` when those values contain a ``.`` + (i.e. look like a hostname rather than an opaque ID). Returns the + raw value; ``SGPClient.normalize_domain`` handles cleanup. + """ + for key in _DOMAIN_KEYS: + value = product.get(key) + if isinstance(value, str) and value: + return value + for key in _PUBLISHER_KEYS: + value = product.get(key) + if isinstance(value, str) and "." in value: + return value + return None + + +class SGPClientError(Exception): + """Error raised by SGPClient for API or transport failures.""" + + def __init__(self, message: str, status_code: int = 0) -> None: + super().__init__(message) + self.status_code = status_code + + +class SGPAuthError(SGPClientError): + """Raised on 401 — api-key missing, invalid, or lacks required scope.""" + + +class SGPClient: + """Async client for SafeGuard Privacy buyer-agent approval checks. + + Normalizes domains (strips scheme, www, port, lowercases), dedupes, + chunks into groups of 10, and caches per-domain results for + ``cache_ttl_seconds``. Returns a dict keyed by normalized domain; a + value of ``None`` means the vendor is unknown to SGP (HTTP 404 or + absent from the batch response). + + Args: + api_key: SGP API key with ``iab:buyerAgent`` scope. + base_url: SGP base URL. Defaults to production + (``https://api.safeguardprivacy.com``). The demo environment + is at ``https://api.safeguardprivacy-demo.com``. + timeout: Request timeout in seconds. + cache_ttl_seconds: How long to cache per-domain results. + """ + + def __init__( + self, + api_key: str, + base_url: str = "https://api.safeguardprivacy.com", + timeout: float = _DEFAULT_TIMEOUT, + cache_ttl_seconds: int = 900, + ) -> None: + self._api_key = api_key + self._base_url = base_url.rstrip("/") + self._timeout = timeout + self._cache_ttl = cache_ttl_seconds + self._cache: dict[str, tuple[float, Optional[ApprovalRecord]]] = {} + self._http = httpx.AsyncClient( + base_url=self._base_url, + headers={"api-key": api_key}, + timeout=timeout, + ) + + async def aclose(self) -> None: + """Close the underlying httpx client.""" + await self._http.aclose() + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + @staticmethod + def normalize_domain(value: str) -> str: + """Reduce a seller URL or raw domain to the form SGP accepts. + + Strips scheme, ``www.``, path, query, and port; lowercases. + Returns an empty string for inputs that yield no host. + """ + if not value: + return "" + raw = value.strip() + # urlparse needs a scheme to extract netloc reliably + if "://" not in raw: + raw = "http://" + raw + host = urlparse(raw).hostname or "" + host = host.lower() + if host.startswith("www."): + host = host[4:] + return host + + async def check_approvals( + self, domains: list[str] + ) -> dict[str, Optional[ApprovalRecord]]: + """Look up IAB buyer-agent approval for a list of domains. + + Args: + domains: Raw seller URLs or domains. Duplicates and invalid + entries are silently dropped. + + Returns: + Dict keyed by normalized domain. ``None`` value means the + vendor is unknown to SGP (not onboarded on the buyer's tenant). + """ + normalized = [self.normalize_domain(d) for d in domains] + normalized = [d for d in normalized if d] + if not normalized: + return {} + + now = time.monotonic() + result: dict[str, Optional[ApprovalRecord]] = {} + to_fetch: list[str] = [] + + seen: set[str] = set() + for d in normalized: + if d in seen: + continue + seen.add(d) + cached = self._cache.get(d) + if cached and (now - cached[0]) < self._cache_ttl: + result[d] = cached[1] + else: + to_fetch.append(d) + + for i in range(0, len(to_fetch), _MAX_BATCH): + chunk = to_fetch[i : i + _MAX_BATCH] + chunk_result = await self._fetch_chunk(chunk) + stamp = time.monotonic() + for d in chunk: + record = chunk_result.get(d) + self._cache[d] = (stamp, record) + result[d] = record + + return result + + # ------------------------------------------------------------------ + # HTTP + # ------------------------------------------------------------------ + + async def _fetch_chunk( + self, domains: list[str] + ) -> dict[str, Optional[ApprovalRecord]]: + """Fetch approvals for up to 10 domains in a single HTTP call.""" + params = {"domain": ",".join(domains)} + try: + resp = await self._http.get(_ENDPOINT, params=params) + except httpx.RequestError as exc: + # Connection refused, timeout, DNS, read errors, etc. — surface + # as SGPClientError so callers catch it on a single type and + # the deal-request gate can fail closed. + raise SGPClientError( + f"SafeGuard Privacy request failed: {exc.__class__.__name__}: {exc}" + ) from exc + + if resp.status_code == 404: + # Entire batch unknown to SGP. + return {d: None for d in domains} + + if resp.status_code == 401: + raise SGPAuthError( + "SafeGuard Privacy rejected the api-key " + "(missing or lacks iab:buyerAgent scope)", + status_code=401, + ) + + if resp.status_code == 400: + raise SGPClientError( + f"SafeGuard Privacy rejected the request as malformed: {resp.text}", + status_code=400, + ) + + if resp.status_code in _RETRYABLE_STATUS_CODES or resp.status_code >= 500: + raise SGPClientError( + f"SafeGuard Privacy returned {resp.status_code}: {resp.text}", + status_code=resp.status_code, + ) + + if resp.status_code != 200: + raise SGPClientError( + f"Unexpected SafeGuard Privacy response {resp.status_code}: {resp.text}", + status_code=resp.status_code, + ) + + try: + payload = resp.json() + except ValueError as exc: + raise SGPClientError(f"SGP response was not JSON: {exc}") from None + + raw_records = payload.get("data") or [] + by_domain: dict[str, Optional[ApprovalRecord]] = {d: None for d in domains} + for raw in raw_records: + try: + record = ApprovalRecord.model_validate(raw) + except (ValueError, TypeError): + logger.warning("Skipping malformed SGP record: %r", raw) + continue + domain_key = self.normalize_domain(record.domain) or record.domain.lower() + by_domain[domain_key] = record + + return by_domain \ No newline at end of file diff --git a/src/ad_buyer/config/settings.py b/src/ad_buyer/config/settings.py index 6c8ac08..0b74eff 100644 --- a/src/ad_buyer/config/settings.py +++ b/src/ad_buyer/config/settings.py @@ -35,6 +35,20 @@ class Settings(BaseSettings): opendirect_token: str | None = None opendirect_api_key: str | None = None + # SafeGuard Privacy — vendor approval gate. + # The integration is inert when ``sgp_api_key`` is empty; deal-request + # enforcement only activates once an SGP API key is supplied AND + # ``sgp_enforce_on_deal_request`` is true. + sgp_api_key: str = "" + # Production endpoint. For testing, use the demo environment: + # https://api.safeguardprivacy-demo.com + sgp_base_url: str = "https://api.safeguardprivacy.com" + sgp_enforce_on_deal_request: bool = False + # Behavior when SafeGuard Privacy returns 404 for a seller domain (vendor + # not in the buyer's SGP portfolio). One of: "block", "warn", "allow". + sgp_unknown_vendor_policy: str = "block" + sgp_cache_ttl_seconds: int = 900 + def get_seller_endpoints(self) -> list[str]: """Parse seller endpoints from comma-separated string. diff --git a/src/ad_buyer/flows/buyer_deal_flow.py b/src/ad_buyer/flows/buyer_deal_flow.py index 330e7f5..15810bb 100644 --- a/src/ad_buyer/flows/buyer_deal_flow.py +++ b/src/ad_buyer/flows/buyer_deal_flow.py @@ -14,7 +14,9 @@ from pydantic import BaseModel, Field from ..agents.level2.buyer_deal_specialist_agent import create_buyer_deal_specialist_agent +from ..clients.sgp_client import SGPClient from ..clients.unified_client import UnifiedClient +from ..config.settings import settings from ..models.buyer_identity import ( AccessTier, BuyerContext, @@ -28,6 +30,7 @@ from ..models.state_machine import BuyerDealStatus, DealStateMachine, InvalidTransitionError from ..storage.deal_store import DealStore from ..tools.buyer_deals import DiscoverInventoryTool, GetPricingTool, RequestDealTool +from ..tools.research import SGPVendorApprovalTool logger = logging.getLogger(__name__) @@ -145,6 +148,7 @@ def __init__( client: UnifiedClient, buyer_context: BuyerContext, store: Optional[DealStore] = None, + sgp_client: Optional[SGPClient] = None, ): """Initialize the flow with client, buyer context, and optional persistence. @@ -153,6 +157,8 @@ def __init__( buyer_context: BuyerContext with identity for tiered access store: Optional DealStore for persisting deal state. When None, the flow behaves identically to before (in-memory only). + sgp_client: Optional SafeGuard Privacy client. When omitted, + one is built from settings if ``SGP_API_KEY`` is set. """ super().__init__() self._client = client @@ -160,10 +166,28 @@ def __init__( self._store = store self._store_deal_id: Optional[str] = None + if sgp_client is None and settings.sgp_api_key: + sgp_client = SGPClient( + api_key=settings.sgp_api_key, + base_url=settings.sgp_base_url, + cache_ttl_seconds=settings.sgp_cache_ttl_seconds, + ) + if ( + sgp_client is None + and settings.sgp_enforce_on_deal_request + ): + logger.warning( + "SGP_ENFORCE_ON_DEAL_REQUEST is true but SGP_API_KEY is empty; " + "the SafeGuard Privacy deal-request gate will be bypassed. " + "Set SGP_API_KEY to enable vendor approval enforcement." + ) + self._sgp_client = sgp_client + # Create tools self._discover_tool = DiscoverInventoryTool( client=client, buyer_context=buyer_context, + sgp_client=sgp_client, ) self._pricing_tool = GetPricingTool( client=client, @@ -172,6 +196,13 @@ def __init__( self._deal_tool = RequestDealTool( client=client, buyer_context=buyer_context, + sgp_client=sgp_client, + sgp_enforce=settings.sgp_enforce_on_deal_request, + sgp_unknown_policy=settings.sgp_unknown_vendor_policy, + ) + # Agent-callable vendor approval tool — only useful with an SGP client. + self._vendor_approval_tool: Optional[SGPVendorApprovalTool] = ( + SGPVendorApprovalTool(client=sgp_client) if sgp_client is not None else None ) # ------------------------------------------------------------------ @@ -325,10 +356,14 @@ def evaluate_and_select(self, discovery_result: dict[str, Any]) -> dict[str, Any try: self.state.status = BuyerDealFlowStatus.EVALUATING_PRICING - # Create crew for intelligent selection - deal_agent = create_buyer_deal_specialist_agent( - tools=[self._discover_tool, self._pricing_tool], - ) + # Create crew for intelligent selection. Include the vendor + # approval tool so the agent can check IAB buyer-agent approval + # status for candidate sellers during selection, not just at + # Deal ID generation. + agent_tools: list[Any] = [self._discover_tool, self._pricing_tool] + if self._vendor_approval_tool is not None: + agent_tools.append(self._vendor_approval_tool) + deal_agent = create_buyer_deal_specialist_agent(tools=agent_tools) selection_task = Task( description=f"""Analyze the discovery results and select the best product diff --git a/src/ad_buyer/models/sgp.py b/src/ad_buyer/models/sgp.py new file mode 100644 index 0000000..fbfc1c4 --- /dev/null +++ b/src/ad_buyer/models/sgp.py @@ -0,0 +1,31 @@ +# Author: SafeGuard Privacy +# Donated to IAB Tech Lab + +"""SafeGuard Privacy (SGP) integration models. + +Mirrors the IabBuyerAgentResource returned by + GET /api/v1/integrations/iab/buyer-agent-approval +on the SafeGuard Privacy platform. +""" + +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field + + +class ApprovalRecord(BaseModel): + """A single vendor's IAB buyer-agent approval status from SafeGuard Privacy.""" + + model_config = ConfigDict(populate_by_name=True) + + vendor_id: int = Field(alias="vendorId") + vendor_company_id: int = Field(alias="vendorCompanyId") + company_name: str = Field(alias="companyName", default="") + domain: str = "" + internal_id: str = Field(alias="internalId", default="") + iab_buyer_agent_approval: bool = Field(alias="iabBuyerAgentApproval", default=False) + iab_buyer_agent_approved_at: datetime | None = Field( + alias="iabBuyerAgentApprovedAt", default=None + ) \ No newline at end of file diff --git a/src/ad_buyer/tools/buyer_deals/discover_inventory.py b/src/ad_buyer/tools/buyer_deals/discover_inventory.py index 17537ca..1bd2302 100644 --- a/src/ad_buyer/tools/buyer_deals/discover_inventory.py +++ b/src/ad_buyer/tools/buyer_deals/discover_inventory.py @@ -3,6 +3,7 @@ """Inventory discovery tool for buyer deal workflows.""" +import logging from typing import Any from crewai.tools import BaseTool @@ -10,8 +11,12 @@ from ...async_utils import run_async from ...booking.pricing import PricingCalculator +from ...clients.sgp_client import SGPClient, SGPClientError, extract_product_domain from ...clients.unified_client import UnifiedClient from ...models.buyer_identity import BuyerContext +from ...models.sgp import ApprovalRecord + +logger = logging.getLogger(__name__) class DiscoverInventoryInput(BaseModel): @@ -76,11 +81,13 @@ class DiscoverInventoryTool(BaseTool): args_schema: type[BaseModel] = DiscoverInventoryInput _client: UnifiedClient _buyer_context: BuyerContext + _sgp_client: SGPClient | None def __init__( self, client: UnifiedClient, buyer_context: BuyerContext, + sgp_client: SGPClient | None = None, **kwargs: Any, ): """Initialize with unified client and buyer context. @@ -88,10 +95,14 @@ def __init__( Args: client: UnifiedClient for seller communication buyer_context: BuyerContext with identity for tiered access + sgp_client: Optional SafeGuard Privacy client. When provided, + each returned product is annotated with the seller's + IAB buyer-agent approval status. """ super().__init__(**kwargs) self._client = client self._buyer_context = buyer_context + self._sgp_client = sgp_client def _run( self, @@ -154,15 +165,63 @@ async def _arun( if not result.success: return f"Error discovering inventory: {result.error}" - return self._format_results(result.data, identity_context) + approvals = await self._fetch_approvals(result.data) + return self._format_results(result.data, identity_context, approvals) except (OSError, ValueError, RuntimeError) as e: return f"Error discovering inventory: {e}" + def _approval_line( + self, + product: dict, + approvals: dict[str, ApprovalRecord | None] | None, + ) -> str | None: + """Render the SGP approval annotation for a single product row.""" + if not approvals or self._sgp_client is None: + return None + raw_domain = extract_product_domain(product) + if not raw_domain: + return " SGP Approval: ? UNKNOWN (no seller domain on product)" + normalized = self._sgp_client.normalize_domain(raw_domain) + record = approvals.get(normalized) + if record is None: + return f" SGP Approval: ? UNKNOWN — {normalized} not in SGP portfolio" + if record.iab_buyer_agent_approval: + return f" SGP Approval: ✓ APPROVED — {normalized}" + return f" SGP Approval: ✗ NOT APPROVED — {normalized}" + + async def _fetch_approvals( + self, products: Any + ) -> dict[str, ApprovalRecord | None]: + """Batch-check SGP approvals for the distinct seller domains in the result. + + Returns a dict keyed by normalized domain. Empty dict when no + SGP client is configured or when a transport error occurs + (discovery fails open — the gate lives in RequestDealTool). + """ + if self._sgp_client is None: + return {} + product_list = products if isinstance(products, list) else [products] + raw_domains: list[str] = [] + for product in product_list: + if not isinstance(product, dict): + continue + domain = extract_product_domain(product) + if domain: + raw_domains.append(domain) + if not raw_domains: + return {} + try: + return await self._sgp_client.check_approvals(raw_domains) + except SGPClientError: + logger.warning("SGP approval lookup failed during discovery; continuing without annotations", exc_info=True) + return {} + def _format_results( self, products: Any, identity_context: dict, + approvals: dict[str, ApprovalRecord | None] | None = None, ) -> str: """Format discovery results with tier information.""" if not products: @@ -210,6 +269,8 @@ def _format_results( else str(base_price) ) + approval_line = self._approval_line(product, approvals) + output_lines.extend( [ f"{i}. {name}", @@ -221,9 +282,11 @@ def _format_results( if isinstance(impressions, int) else f" Available: {impressions}", f" Targeting: {', '.join(targeting) if targeting else 'Standard'}", - "", ] ) + if approval_line: + output_lines.append(approval_line) + output_lines.append("") else: output_lines.append(f"{i}. {product}") output_lines.append("") diff --git a/src/ad_buyer/tools/buyer_deals/request_deal.py b/src/ad_buyer/tools/buyer_deals/request_deal.py index cdaccf5..9af2c06 100644 --- a/src/ad_buyer/tools/buyer_deals/request_deal.py +++ b/src/ad_buyer/tools/buyer_deals/request_deal.py @@ -3,6 +3,7 @@ """Deal ID request tool for buyer deal workflows.""" +import logging from datetime import datetime, timedelta, timezone from typing import Any @@ -12,6 +13,7 @@ from ...async_utils import run_async from ...booking.deal_id import generate_deal_id from ...booking.pricing import PricingCalculator +from ...clients.sgp_client import SGPClient, SGPClientError, extract_product_domain from ...clients.unified_client import UnifiedClient from ...models.buyer_identity import ( AccessTier, @@ -19,6 +21,11 @@ DealResponse, DealType, ) +from ...models.sgp import ApprovalRecord + +logger = logging.getLogger(__name__) + +_VALID_UNKNOWN_POLICIES = {"block", "warn", "allow"} class RequestDealInput(BaseModel): @@ -85,11 +92,17 @@ class RequestDealTool(BaseTool): args_schema: type[BaseModel] = RequestDealInput _client: UnifiedClient _buyer_context: BuyerContext + _sgp_client: SGPClient | None + _sgp_enforce: bool + _sgp_unknown_policy: str def __init__( self, client: UnifiedClient, buyer_context: BuyerContext, + sgp_client: SGPClient | None = None, + sgp_enforce: bool = False, + sgp_unknown_policy: str = "block", **kwargs: Any, ): """Initialize with unified client and buyer context. @@ -97,10 +110,26 @@ def __init__( Args: client: UnifiedClient for seller communication buyer_context: BuyerContext with identity for tiered access + sgp_client: Optional SafeGuard Privacy client. When provided + and ``sgp_enforce`` is True, the seller's IAB buyer-agent + approval is verified before a Deal ID is generated. + sgp_enforce: When True, block the deal request unless SGP + returns ``iabBuyerAgentApproval=true`` for the seller. + sgp_unknown_policy: How to treat vendors absent from the + buyer's SGP portfolio (HTTP 404). One of ``block``, + ``warn``, ``allow``. """ super().__init__(**kwargs) self._client = client self._buyer_context = buyer_context + self._sgp_client = sgp_client + self._sgp_enforce = sgp_enforce + if sgp_unknown_policy not in _VALID_UNKNOWN_POLICIES: + raise ValueError( + f"Invalid sgp_unknown_policy '{sgp_unknown_policy}'. " + f"Must be one of: {', '.join(sorted(_VALID_UNKNOWN_POLICIES))}" + ) + self._sgp_unknown_policy = sgp_unknown_policy def _run( self, @@ -160,6 +189,11 @@ async def _arun( if not product: return f"Product {product_id} not found." + # SafeGuard Privacy approval gate — must pass before a Deal ID is issued. + gate_error, approval_banner = await self._check_sgp_approval(product) + if gate_error: + return gate_error + # Calculate pricing deal_response = self._create_deal_response( product=product, @@ -170,11 +204,93 @@ async def _arun( target_cpm=target_cpm, ) - return self._format_deal_response(deal_response) + formatted = self._format_deal_response(deal_response) + if approval_banner: + formatted = f"{approval_banner}\n{formatted}" + return formatted except (OSError, ValueError, RuntimeError) as e: return f"Error requesting deal: {e}" + async def _check_sgp_approval( + self, product: dict + ) -> tuple[str | None, str | None]: + """Gate a deal request against SafeGuard Privacy approval. + + Returns ``(error_message, banner)``: + * ``error_message`` is non-None when the deal must be refused. + * ``banner`` is a one-line note prepended to a successful deal + response (e.g. "warn" policy, unknown vendor proceeding). + + When ``sgp_client`` is None or ``sgp_enforce`` is False, the gate + is skipped entirely. + """ + if self._sgp_client is None or not self._sgp_enforce: + return None, None + + raw_domain = extract_product_domain(product) + if not raw_domain: + return ( + "Deal blocked: cannot determine seller domain for SafeGuard " + "Privacy approval check. Add a seller_url / publisher_domain " + "field to the product, or disable SGP_ENFORCE_ON_DEAL_REQUEST.", + None, + ) + + domain = self._sgp_client.normalize_domain(raw_domain) or raw_domain + + try: + approvals = await self._sgp_client.check_approvals([raw_domain]) + except SGPClientError as exc: + logger.warning( + "SafeGuard Privacy lookup failed for %s during deal request", domain, + exc_info=True, + ) + # Fail closed — enforcement is on, so we must not issue a Deal ID + # when the privacy gate cannot be evaluated. + return ( + f"Deal blocked: SafeGuard Privacy lookup failed for {domain} " + f"({exc}). Retry once the SGP service is reachable.", + None, + ) + + record: ApprovalRecord | None = approvals.get(domain) + + if record is None: + if self._sgp_unknown_policy == "allow": + return None, f"SGP: {domain} not in SGP portfolio — allowed by policy." + if self._sgp_unknown_policy == "warn": + return None, ( + f"SGP WARNING: {domain} is not in your SGP portfolio. " + f"Onboard and approve this vendor in SafeGuard Privacy " + f"to suppress this warning." + ) + return ( + f"Deal blocked: {domain} is not in your SafeGuard Privacy " + f"portfolio. Onboard and approve the vendor in SGP before " + f"requesting a Deal ID.", + None, + ) + + if not record.iab_buyer_agent_approval: + return ( + f"Deal blocked: {record.company_name or domain} does not carry " + f"the IAB buyer-agent approval flag in SafeGuard Privacy. " + f"Update the vendor's approval in SGP and retry.", + None, + ) + + approved_at = ( + record.iab_buyer_agent_approved_at.isoformat() + if record.iab_buyer_agent_approved_at + else "date unknown" + ) + banner = ( + f"SGP: ✓ {record.company_name or domain} approved for IAB " + f"buyer-agent purchases (since {approved_at})." + ) + return None, banner + def _create_deal_response( self, product: dict, diff --git a/src/ad_buyer/tools/research/__init__.py b/src/ad_buyer/tools/research/__init__.py index 5e7c982..b596c0b 100644 --- a/src/ad_buyer/tools/research/__init__.py +++ b/src/ad_buyer/tools/research/__init__.py @@ -5,5 +5,6 @@ from .avails_check import AvailsCheckTool from .product_search import ProductSearchTool +from .sgp_vendor_approval import SGPVendorApprovalTool -__all__ = ["ProductSearchTool", "AvailsCheckTool"] +__all__ = ["ProductSearchTool", "AvailsCheckTool", "SGPVendorApprovalTool"] diff --git a/src/ad_buyer/tools/research/sgp_vendor_approval.py b/src/ad_buyer/tools/research/sgp_vendor_approval.py new file mode 100644 index 0000000..0a89709 --- /dev/null +++ b/src/ad_buyer/tools/research/sgp_vendor_approval.py @@ -0,0 +1,96 @@ +# Author: SafeGuard Privacy +# Donated to IAB Tech Lab + +"""CrewAI tool: check IAB buyer-agent approval via SafeGuard Privacy. + +The class is intentionally prefixed ``SGP`` so that future vendor-approval +integrations (e.g. OneTrust, an IAB Tech Lab registry) can coexist under +distinct class names and distinct CrewAI tool ``name`` attributes. +""" + +from __future__ import annotations + +from typing import Any + +from crewai.tools import BaseTool +from pydantic import BaseModel, Field + +from ...async_utils import run_async +from ...clients.sgp_client import SGPClient, SGPClientError + + +class SGPVendorApprovalInput(BaseModel): + """Input schema for the SafeGuard Privacy vendor approval check tool.""" + + domains: list[str] = Field( + ..., + description=( + "Seller domains or full seller URLs to check. Scheme, www, and " + "port are stripped automatically. Up to 10 are checked per call; " + "larger lists are batched." + ), + ) + + +class SGPVendorApprovalTool(BaseTool): + """Check whether seller vendors carry the IAB buyer-agent approval flag. + + Consults the SafeGuard Privacy `iab/buyer-agent-approval` endpoint, + which returns the ``iabBuyerAgentApproval`` boolean per vendor on the + buyer's SGP tenant. Vendors absent from the tenant come back as + ``UNKNOWN``. + """ + + name: str = "check_sgp_vendor_approval" + description: str = ( + "Check IAB buyer-agent approval for seller domains via SafeGuard " + "Privacy. Returns APPROVED / NOT APPROVED / UNKNOWN per domain, " + "along with the approval date when available. Use before " + "requesting a Deal ID from a seller." + ) + args_schema: type[BaseModel] = SGPVendorApprovalInput + _client: SGPClient + + def __init__(self, client: SGPClient, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._client = client + + def _run(self, domains: list[str]) -> str: + return run_async(self._arun(domains=domains)) + + async def _arun(self, domains: list[str]) -> str: + try: + results = await self._client.check_approvals(domains) + except SGPClientError as exc: + return f"SafeGuard Privacy lookup failed: {exc}" + + if not results: + return "No valid domains were provided." + + lines = [ + "SafeGuard Privacy — IAB Buyer-Agent Approval", + "-" * 50, + ] + for domain in sorted(results): + record = results[domain] + if record is None: + lines.append(f"? {domain}: UNKNOWN (not in SGP portfolio)") + continue + if record.iab_buyer_agent_approval: + approved_at = ( + record.iab_buyer_agent_approved_at.isoformat() + if record.iab_buyer_agent_approved_at + else "date unknown" + ) + lines.append( + f"✓ {domain}: APPROVED " + f"({record.company_name or 'company name unavailable'}, " + f"since {approved_at})" + ) + else: + lines.append( + f"✗ {domain}: NOT APPROVED " + f"({record.company_name or 'company name unavailable'})" + ) + + return "\n".join(lines) \ No newline at end of file diff --git a/tests/unit/test_sgp_client.py b/tests/unit/test_sgp_client.py new file mode 100644 index 0000000..af6de70 --- /dev/null +++ b/tests/unit/test_sgp_client.py @@ -0,0 +1,273 @@ +# Author: SafeGuard Privacy +# Donated to IAB Tech Lab + +"""Tests for the SafeGuard Privacy (SGP) client. + +Covers domain normalization, batch chunking to 10, HTTP status handling +(200 / 400 / 401 / 404 / 5xx), response parsing, TTL cache, and the +api-key header. +""" + +from __future__ import annotations + +import httpx +import pytest + +from ad_buyer.clients.sgp_client import ( + SGPAuthError, + SGPClient, + SGPClientError, +) + +BASE_URL = "https://sgp.test" + + +def _make_client(handler, *, cache_ttl_seconds: int = 900) -> SGPClient: + """Build an SGPClient whose internal httpx client uses MockTransport.""" + c = SGPClient( + api_key="test-key", + base_url=BASE_URL, + cache_ttl_seconds=cache_ttl_seconds, + timeout=5.0, + ) + transport = httpx.MockTransport(handler) + c._http = httpx.AsyncClient( + transport=transport, + base_url=BASE_URL, + headers=dict(c._http.headers), + timeout=5.0, + ) + return c + + +def _success_body(records: list[dict]) -> dict: + return { + "status": "success", + "code": 200, + "message": "", + "data": records, + "pagination": {}, + } + + +def _record(domain: str, approved: bool, approved_at: str | None = "2026-03-14T12:00:00Z") -> dict: + return { + "vendorId": hash(domain) & 0xFFFF, + "vendorCompanyId": (hash(domain) + 1) & 0xFFFF, + "companyName": domain.split(".")[0].title() + " Inc.", + "domain": domain, + "internalId": "", + "iabBuyerAgentApproval": approved, + "iabBuyerAgentApprovedAt": approved_at, + } + + +# --------------------------------------------------------------------------- +# Domain normalization +# --------------------------------------------------------------------------- + + +class TestNormalizeDomain: + @pytest.mark.parametrize( + "raw, expected", + [ + ("example.com", "example.com"), + ("Example.COM", "example.com"), + ("www.example.com", "example.com"), + ("http://example.com", "example.com"), + ("https://www.example.com/path?q=1", "example.com"), + ("http://seller.example.com:8001", "seller.example.com"), + ("", ""), + (" ", ""), + ], + ) + def test_normalizes(self, raw: str, expected: str) -> None: + assert SGPClient.normalize_domain(raw) == expected + + +# --------------------------------------------------------------------------- +# Successful lookups +# --------------------------------------------------------------------------- + + +class TestCheckApprovalsSuccess: + @pytest.mark.asyncio + async def test_single_approved_vendor(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/api/v1/integrations/iab/buyer-agent-approval" + assert request.url.params["domain"] == "example.com" + assert request.headers["api-key"] == "test-key" + return httpx.Response(200, json=_success_body([_record("example.com", True)])) + + client = _make_client(handler) + results = await client.check_approvals(["https://example.com/foo"]) + assert set(results) == {"example.com"} + record = results["example.com"] + assert record is not None + assert record.iab_buyer_agent_approval is True + assert record.iab_buyer_agent_approved_at is not None + + @pytest.mark.asyncio + async def test_multiple_domains_single_call(self) -> None: + seen_params: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_params.append(request.url.params["domain"]) + return httpx.Response( + 200, + json=_success_body([ + _record("a.com", True), + _record("b.com", False), + ]), + ) + + client = _make_client(handler) + results = await client.check_approvals(["a.com", "b.com"]) + assert seen_params == ["a.com,b.com"] + assert results["a.com"].iab_buyer_agent_approval is True + assert results["b.com"].iab_buyer_agent_approval is False + + @pytest.mark.asyncio + async def test_batches_more_than_ten_domains(self) -> None: + captured: list[list[str]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + domains = request.url.params["domain"].split(",") + captured.append(domains) + records = [_record(d, True) for d in domains] + return httpx.Response(200, json=_success_body(records)) + + client = _make_client(handler) + domains = [f"d{i}.com" for i in range(25)] + results = await client.check_approvals(domains) + + assert [len(c) for c in captured] == [10, 10, 5] + assert len(results) == 25 + assert all(r is not None and r.iab_buyer_agent_approval for r in results.values()) + + @pytest.mark.asyncio + async def test_dedupes_input(self) -> None: + captured_domains: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured_domains.extend(request.url.params["domain"].split(",")) + return httpx.Response( + 200, json=_success_body([_record("example.com", True)]) + ) + + client = _make_client(handler) + await client.check_approvals(["example.com", "www.example.com", "EXAMPLE.COM"]) + assert captured_domains == ["example.com"] + + +# --------------------------------------------------------------------------- +# Not-found / unknown vendor +# --------------------------------------------------------------------------- + + +class TestUnknownVendor: + @pytest.mark.asyncio + async def test_404_marks_all_batch_domains_unknown(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"status": "error", "code": 404, "data": None}) + + client = _make_client(handler) + results = await client.check_approvals(["unknown1.com", "unknown2.com"]) + assert results == {"unknown1.com": None, "unknown2.com": None} + + @pytest.mark.asyncio + async def test_partial_batch_response_marks_missing_as_unknown(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + # SGP only returns records for domains it actually knows; the + # unknown ones are simply absent from the data array. + return httpx.Response( + 200, json=_success_body([_record("known.com", True)]) + ) + + client = _make_client(handler) + results = await client.check_approvals(["known.com", "mystery.com"]) + assert results["known.com"] is not None + assert results["mystery.com"] is None + + +# --------------------------------------------------------------------------- +# Error paths +# --------------------------------------------------------------------------- + + +class TestErrorHandling: + @pytest.mark.asyncio + async def test_401_raises_auth_error(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(401, text="unauthorized") + + client = _make_client(handler) + with pytest.raises(SGPAuthError): + await client.check_approvals(["example.com"]) + + @pytest.mark.asyncio + async def test_400_raises_client_error(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(400, text="bad domain") + + client = _make_client(handler) + with pytest.raises(SGPClientError) as exc_info: + await client.check_approvals(["example.com"]) + assert exc_info.value.status_code == 400 + + @pytest.mark.asyncio + async def test_5xx_raises_client_error(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(503, text="maintenance") + + client = _make_client(handler) + with pytest.raises(SGPClientError) as exc_info: + await client.check_approvals(["example.com"]) + assert exc_info.value.status_code == 503 + + @pytest.mark.asyncio + async def test_transport_error_wrapped_as_client_error(self) -> None: + """Real httpx transport failures (connect/timeout/DNS) surface as SGPClientError.""" + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("connection refused") + + client = _make_client(handler) + with pytest.raises(SGPClientError) as exc_info: + await client.check_approvals(["example.com"]) + assert "ConnectError" in str(exc_info.value) + + +# --------------------------------------------------------------------------- +# Caching +# --------------------------------------------------------------------------- + + +class TestCache: + @pytest.mark.asyncio + async def test_cache_hit_avoids_second_request(self) -> None: + calls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + calls["n"] += 1 + return httpx.Response( + 200, json=_success_body([_record("cached.com", True)]) + ) + + client = _make_client(handler) + first = await client.check_approvals(["cached.com"]) + second = await client.check_approvals(["cached.com"]) + assert calls["n"] == 1 + assert first["cached.com"].vendor_id == second["cached.com"].vendor_id + + @pytest.mark.asyncio + async def test_cache_stores_unknown_result(self) -> None: + calls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + calls["n"] += 1 + return httpx.Response(404, json={"status": "error", "code": 404}) + + client = _make_client(handler) + await client.check_approvals(["mystery.com"]) + await client.check_approvals(["mystery.com"]) + assert calls["n"] == 1 \ No newline at end of file diff --git a/tests/unit/test_sgp_gate.py b/tests/unit/test_sgp_gate.py new file mode 100644 index 0000000..f12b1cc --- /dev/null +++ b/tests/unit/test_sgp_gate.py @@ -0,0 +1,298 @@ +# Author: SafeGuard Privacy +# Donated to IAB Tech Lab + +"""Tests for the SafeGuard Privacy deal-request gate in RequestDealTool.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from ad_buyer.clients.sgp_client import SGPClientError +from ad_buyer.models.buyer_identity import BuyerContext, BuyerIdentity +from ad_buyer.models.sgp import ApprovalRecord +from ad_buyer.tools.buyer_deals import RequestDealTool + + +@pytest.fixture +def agency_context() -> BuyerContext: + identity = BuyerIdentity( + seat_id="ttd-seat-123", + agency_id="omnicom-456", + agency_name="OMD", + ) + return BuyerContext(identity=identity, is_authenticated=True) + + +@pytest.fixture +def mock_client() -> MagicMock: + """UnifiedClient mock that returns a product with a seller_url.""" + client = MagicMock() + client.get_product = AsyncMock( + return_value=MagicMock( + success=True, + data={ + "id": "prod_1", + "name": "Premium CTV", + "basePrice": 20.00, + "seller_url": "http://seller.example.com:8001", + }, + ) + ) + return client + + +def _approved(domain: str) -> ApprovalRecord: + return ApprovalRecord.model_validate({ + "vendorId": 1, + "vendorCompanyId": 10, + "companyName": "Example Seller", + "domain": domain, + "internalId": "", + "iabBuyerAgentApproval": True, + "iabBuyerAgentApprovedAt": "2026-03-01T00:00:00Z", + }) + + +def _denied(domain: str) -> ApprovalRecord: + return ApprovalRecord.model_validate({ + "vendorId": 2, + "vendorCompanyId": 20, + "companyName": "Shady Seller", + "domain": domain, + "internalId": "", + "iabBuyerAgentApproval": False, + "iabBuyerAgentApprovedAt": None, + }) + + +# --------------------------------------------------------------------------- +# Gate off +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_no_sgp_client_bypasses_gate(mock_client, agency_context): + """When no SGP client is wired in, the tool operates as before.""" + tool = RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_client=None, + ) + result = await tool._arun(product_id="prod_1", impressions=100) + assert "DEAL CREATED SUCCESSFULLY" in result + + +@pytest.mark.asyncio +async def test_enforce_false_bypasses_gate(mock_client, agency_context): + """When enforcement is off, the gate does not block.""" + sgp = MagicMock() + sgp.normalize_domain = MagicMock(return_value="seller.example.com") + sgp.check_approvals = AsyncMock( + return_value={"seller.example.com": _denied("seller.example.com")} + ) + tool = RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_client=sgp, + sgp_enforce=False, + ) + result = await tool._arun(product_id="prod_1", impressions=100) + assert "DEAL CREATED SUCCESSFULLY" in result + sgp.check_approvals.assert_not_called() + + +# --------------------------------------------------------------------------- +# Approved +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_approved_vendor_allows_deal(mock_client, agency_context): + sgp = MagicMock() + sgp.normalize_domain = MagicMock(return_value="seller.example.com") + sgp.check_approvals = AsyncMock( + return_value={"seller.example.com": _approved("seller.example.com")} + ) + tool = RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_client=sgp, + sgp_enforce=True, + ) + result = await tool._arun(product_id="prod_1", impressions=100) + assert "DEAL CREATED SUCCESSFULLY" in result + assert "SGP: ✓" in result + assert "approved" in result.lower() + + +# --------------------------------------------------------------------------- +# Denied +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_denied_vendor_blocks_deal(mock_client, agency_context): + sgp = MagicMock() + sgp.normalize_domain = MagicMock(return_value="seller.example.com") + sgp.check_approvals = AsyncMock( + return_value={"seller.example.com": _denied("seller.example.com")} + ) + tool = RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_client=sgp, + sgp_enforce=True, + ) + result = await tool._arun(product_id="prod_1", impressions=100) + assert "Deal blocked" in result + assert "IAB buyer-agent approval" in result + assert "DEAL CREATED SUCCESSFULLY" not in result + + +# --------------------------------------------------------------------------- +# Unknown vendor policies +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_unknown_vendor_blocks_by_default(mock_client, agency_context): + sgp = MagicMock() + sgp.normalize_domain = MagicMock(return_value="seller.example.com") + sgp.check_approvals = AsyncMock(return_value={"seller.example.com": None}) + tool = RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_client=sgp, + sgp_enforce=True, + sgp_unknown_policy="block", + ) + result = await tool._arun(product_id="prod_1", impressions=100) + assert "Deal blocked" in result + assert "not in your SafeGuard Privacy" in result + assert "DEAL CREATED SUCCESSFULLY" not in result + + +@pytest.mark.asyncio +async def test_unknown_vendor_warn_allows_with_banner(mock_client, agency_context): + sgp = MagicMock() + sgp.normalize_domain = MagicMock(return_value="seller.example.com") + sgp.check_approvals = AsyncMock(return_value={"seller.example.com": None}) + tool = RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_client=sgp, + sgp_enforce=True, + sgp_unknown_policy="warn", + ) + result = await tool._arun(product_id="prod_1", impressions=100) + assert "SGP WARNING" in result + assert "DEAL CREATED SUCCESSFULLY" in result + + +@pytest.mark.asyncio +async def test_unknown_vendor_allow_proceeds_silently(mock_client, agency_context): + sgp = MagicMock() + sgp.normalize_domain = MagicMock(return_value="seller.example.com") + sgp.check_approvals = AsyncMock(return_value={"seller.example.com": None}) + tool = RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_client=sgp, + sgp_enforce=True, + sgp_unknown_policy="allow", + ) + result = await tool._arun(product_id="prod_1", impressions=100) + assert "DEAL CREATED SUCCESSFULLY" in result + + +# --------------------------------------------------------------------------- +# Failure modes +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_transport_error_fails_closed_when_enforcing(mock_client, agency_context): + """When SGP is unreachable and enforcement is on, deal must not be issued.""" + sgp = MagicMock() + sgp.normalize_domain = MagicMock(return_value="seller.example.com") + sgp.check_approvals = AsyncMock(side_effect=SGPClientError("upstream 503")) + tool = RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_client=sgp, + sgp_enforce=True, + ) + result = await tool._arun(product_id="prod_1", impressions=100) + assert "Deal blocked" in result + assert "SafeGuard Privacy lookup failed" in result + assert "DEAL CREATED SUCCESSFULLY" not in result + + +@pytest.mark.asyncio +async def test_product_without_domain_blocks_when_enforcing(agency_context): + """A product missing any seller domain field cannot be evaluated, so block.""" + mock_client = MagicMock() + mock_client.get_product = AsyncMock( + return_value=MagicMock( + success=True, + data={"id": "prod_1", "name": "Test", "basePrice": 20.00}, + ) + ) + sgp = MagicMock() + sgp.normalize_domain = MagicMock(return_value="") + sgp.check_approvals = AsyncMock() + tool = RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_client=sgp, + sgp_enforce=True, + ) + result = await tool._arun(product_id="prod_1", impressions=100) + assert "Deal blocked" in result + assert "seller domain" in result + sgp.check_approvals.assert_not_called() + + +def test_invalid_unknown_policy_rejected(mock_client, agency_context): + with pytest.raises(ValueError, match="sgp_unknown_policy"): + RequestDealTool( + client=mock_client, + buyer_context=agency_context, + sgp_unknown_policy="maybe", + ) + + +# --------------------------------------------------------------------------- +# Flow-level wiring of SGPVendorApprovalTool +# --------------------------------------------------------------------------- + + +def test_flow_wires_vendor_approval_tool_when_sgp_configured(agency_context): + """BuyerDealFlow exposes the vendor approval tool to the deal agent.""" + from ad_buyer.clients.sgp_client import SGPClient + from ad_buyer.flows.buyer_deal_flow import BuyerDealFlow + from ad_buyer.tools.research import SGPVendorApprovalTool + + sgp = SGPClient(api_key="k", base_url="https://sgp.test") + flow = BuyerDealFlow( + client=MagicMock(), + buyer_context=agency_context, + sgp_client=sgp, + ) + assert isinstance(flow._vendor_approval_tool, SGPVendorApprovalTool) + + +def test_flow_omits_vendor_approval_tool_without_sgp(agency_context, monkeypatch): + """Without an SGP client (and no SGP_API_KEY env), the tool is not built.""" + from ad_buyer.config.settings import settings + from ad_buyer.flows.buyer_deal_flow import BuyerDealFlow + + monkeypatch.setattr(settings, "sgp_api_key", "") + flow = BuyerDealFlow( + client=MagicMock(), + buyer_context=agency_context, + sgp_client=None, + ) + assert flow._vendor_approval_tool is None \ No newline at end of file From a9a43df084526ce27ae340b264b7d69e2a713bdc Mon Sep 17 00:00:00 2001 From: Daniel Cox Date: Wed, 22 Apr 2026 16:24:29 -0600 Subject: [PATCH 2/2] Lint and format our SGP files to match project ruff config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scope is limited to lines introduced by this PR: - Trailing newlines added to new files (W292) - Optional[X] → X | None on our new type annotations (UP045) - Split an over-long logger.warning line in discover_inventory.py (E501) - Minor whitespace/formatting adjustments from `ruff format` on our five new files only Pre-existing lint issues in files we also modified (unused imports, datetime.timezone.utc → datetime.UTC, E501 on the pre-existing RequestDealInput description, I001 import-group order caused by upstream `events.*` placement) are intentionally left alone — they are out of scope for this PR. Tests: 2704 passed, 41 skipped. --- src/ad_buyer/clients/sgp_client.py | 22 ++++------ src/ad_buyer/flows/buyer_deal_flow.py | 4 +- src/ad_buyer/models/sgp.py | 2 +- .../tools/buyer_deals/discover_inventory.py | 6 ++- .../tools/research/sgp_vendor_approval.py | 2 +- tests/unit/test_sgp_client.py | 25 +++++------ tests/unit/test_sgp_gate.py | 42 ++++++++++--------- 7 files changed, 51 insertions(+), 52 deletions(-) diff --git a/src/ad_buyer/clients/sgp_client.py b/src/ad_buyer/clients/sgp_client.py index 25e83bf..97758d5 100644 --- a/src/ad_buyer/clients/sgp_client.py +++ b/src/ad_buyer/clients/sgp_client.py @@ -18,7 +18,6 @@ import logging import time -from typing import Optional from urllib.parse import urlparse import httpx @@ -39,7 +38,7 @@ _PUBLISHER_KEYS = ("publisherId", "publisher") -def extract_product_domain(product: dict) -> Optional[str]: +def extract_product_domain(product: dict) -> str | None: """Best-guess seller domain from a product dict for an SGP lookup. Checks explicit domain/URL fields first, then falls back to @@ -99,7 +98,7 @@ def __init__( self._base_url = base_url.rstrip("/") self._timeout = timeout self._cache_ttl = cache_ttl_seconds - self._cache: dict[str, tuple[float, Optional[ApprovalRecord]]] = {} + self._cache: dict[str, tuple[float, ApprovalRecord | None]] = {} self._http = httpx.AsyncClient( base_url=self._base_url, headers={"api-key": api_key}, @@ -133,9 +132,7 @@ def normalize_domain(value: str) -> str: host = host[4:] return host - async def check_approvals( - self, domains: list[str] - ) -> dict[str, Optional[ApprovalRecord]]: + async def check_approvals(self, domains: list[str]) -> dict[str, ApprovalRecord | None]: """Look up IAB buyer-agent approval for a list of domains. Args: @@ -152,7 +149,7 @@ async def check_approvals( return {} now = time.monotonic() - result: dict[str, Optional[ApprovalRecord]] = {} + result: dict[str, ApprovalRecord | None] = {} to_fetch: list[str] = [] seen: set[str] = set() @@ -181,9 +178,7 @@ async def check_approvals( # HTTP # ------------------------------------------------------------------ - async def _fetch_chunk( - self, domains: list[str] - ) -> dict[str, Optional[ApprovalRecord]]: + async def _fetch_chunk(self, domains: list[str]) -> dict[str, ApprovalRecord | None]: """Fetch approvals for up to 10 domains in a single HTTP call.""" params = {"domain": ",".join(domains)} try: @@ -202,8 +197,7 @@ async def _fetch_chunk( if resp.status_code == 401: raise SGPAuthError( - "SafeGuard Privacy rejected the api-key " - "(missing or lacks iab:buyerAgent scope)", + "SafeGuard Privacy rejected the api-key (missing or lacks iab:buyerAgent scope)", status_code=401, ) @@ -231,7 +225,7 @@ async def _fetch_chunk( raise SGPClientError(f"SGP response was not JSON: {exc}") from None raw_records = payload.get("data") or [] - by_domain: dict[str, Optional[ApprovalRecord]] = {d: None for d in domains} + by_domain: dict[str, ApprovalRecord | None] = {d: None for d in domains} for raw in raw_records: try: record = ApprovalRecord.model_validate(raw) @@ -241,4 +235,4 @@ async def _fetch_chunk( domain_key = self.normalize_domain(record.domain) or record.domain.lower() by_domain[domain_key] = record - return by_domain \ No newline at end of file + return by_domain diff --git a/src/ad_buyer/flows/buyer_deal_flow.py b/src/ad_buyer/flows/buyer_deal_flow.py index 15810bb..f372151 100644 --- a/src/ad_buyer/flows/buyer_deal_flow.py +++ b/src/ad_buyer/flows/buyer_deal_flow.py @@ -148,7 +148,7 @@ def __init__( client: UnifiedClient, buyer_context: BuyerContext, store: Optional[DealStore] = None, - sgp_client: Optional[SGPClient] = None, + sgp_client: SGPClient | None = None, ): """Initialize the flow with client, buyer context, and optional persistence. @@ -201,7 +201,7 @@ def __init__( sgp_unknown_policy=settings.sgp_unknown_vendor_policy, ) # Agent-callable vendor approval tool — only useful with an SGP client. - self._vendor_approval_tool: Optional[SGPVendorApprovalTool] = ( + self._vendor_approval_tool: SGPVendorApprovalTool | None = ( SGPVendorApprovalTool(client=sgp_client) if sgp_client is not None else None ) diff --git a/src/ad_buyer/models/sgp.py b/src/ad_buyer/models/sgp.py index fbfc1c4..2f66667 100644 --- a/src/ad_buyer/models/sgp.py +++ b/src/ad_buyer/models/sgp.py @@ -28,4 +28,4 @@ class ApprovalRecord(BaseModel): iab_buyer_agent_approval: bool = Field(alias="iabBuyerAgentApproval", default=False) iab_buyer_agent_approved_at: datetime | None = Field( alias="iabBuyerAgentApprovedAt", default=None - ) \ No newline at end of file + ) diff --git a/src/ad_buyer/tools/buyer_deals/discover_inventory.py b/src/ad_buyer/tools/buyer_deals/discover_inventory.py index 1bd2302..2ae1cf1 100644 --- a/src/ad_buyer/tools/buyer_deals/discover_inventory.py +++ b/src/ad_buyer/tools/buyer_deals/discover_inventory.py @@ -214,7 +214,11 @@ async def _fetch_approvals( try: return await self._sgp_client.check_approvals(raw_domains) except SGPClientError: - logger.warning("SGP approval lookup failed during discovery; continuing without annotations", exc_info=True) + logger.warning( + "SGP approval lookup failed during discovery; " + "continuing without annotations", + exc_info=True, + ) return {} def _format_results( diff --git a/src/ad_buyer/tools/research/sgp_vendor_approval.py b/src/ad_buyer/tools/research/sgp_vendor_approval.py index 0a89709..f692bf9 100644 --- a/src/ad_buyer/tools/research/sgp_vendor_approval.py +++ b/src/ad_buyer/tools/research/sgp_vendor_approval.py @@ -93,4 +93,4 @@ async def _arun(self, domains: list[str]) -> str: f"({record.company_name or 'company name unavailable'})" ) - return "\n".join(lines) \ No newline at end of file + return "\n".join(lines) diff --git a/tests/unit/test_sgp_client.py b/tests/unit/test_sgp_client.py index af6de70..4820c1a 100644 --- a/tests/unit/test_sgp_client.py +++ b/tests/unit/test_sgp_client.py @@ -115,10 +115,12 @@ def handler(request: httpx.Request) -> httpx.Response: seen_params.append(request.url.params["domain"]) return httpx.Response( 200, - json=_success_body([ - _record("a.com", True), - _record("b.com", False), - ]), + json=_success_body( + [ + _record("a.com", True), + _record("b.com", False), + ] + ), ) client = _make_client(handler) @@ -151,9 +153,7 @@ async def test_dedupes_input(self) -> None: def handler(request: httpx.Request) -> httpx.Response: captured_domains.extend(request.url.params["domain"].split(",")) - return httpx.Response( - 200, json=_success_body([_record("example.com", True)]) - ) + return httpx.Response(200, json=_success_body([_record("example.com", True)])) client = _make_client(handler) await client.check_approvals(["example.com", "www.example.com", "EXAMPLE.COM"]) @@ -180,9 +180,7 @@ async def test_partial_batch_response_marks_missing_as_unknown(self) -> None: def handler(request: httpx.Request) -> httpx.Response: # SGP only returns records for domains it actually knows; the # unknown ones are simply absent from the data array. - return httpx.Response( - 200, json=_success_body([_record("known.com", True)]) - ) + return httpx.Response(200, json=_success_body([_record("known.com", True)])) client = _make_client(handler) results = await client.check_approvals(["known.com", "mystery.com"]) @@ -228,6 +226,7 @@ def handler(request: httpx.Request) -> httpx.Response: @pytest.mark.asyncio async def test_transport_error_wrapped_as_client_error(self) -> None: """Real httpx transport failures (connect/timeout/DNS) surface as SGPClientError.""" + def handler(request: httpx.Request) -> httpx.Response: raise httpx.ConnectError("connection refused") @@ -249,9 +248,7 @@ async def test_cache_hit_avoids_second_request(self) -> None: def handler(request: httpx.Request) -> httpx.Response: calls["n"] += 1 - return httpx.Response( - 200, json=_success_body([_record("cached.com", True)]) - ) + return httpx.Response(200, json=_success_body([_record("cached.com", True)])) client = _make_client(handler) first = await client.check_approvals(["cached.com"]) @@ -270,4 +267,4 @@ def handler(request: httpx.Request) -> httpx.Response: client = _make_client(handler) await client.check_approvals(["mystery.com"]) await client.check_approvals(["mystery.com"]) - assert calls["n"] == 1 \ No newline at end of file + assert calls["n"] == 1 diff --git a/tests/unit/test_sgp_gate.py b/tests/unit/test_sgp_gate.py index f12b1cc..e6dd34f 100644 --- a/tests/unit/test_sgp_gate.py +++ b/tests/unit/test_sgp_gate.py @@ -44,27 +44,31 @@ def mock_client() -> MagicMock: def _approved(domain: str) -> ApprovalRecord: - return ApprovalRecord.model_validate({ - "vendorId": 1, - "vendorCompanyId": 10, - "companyName": "Example Seller", - "domain": domain, - "internalId": "", - "iabBuyerAgentApproval": True, - "iabBuyerAgentApprovedAt": "2026-03-01T00:00:00Z", - }) + return ApprovalRecord.model_validate( + { + "vendorId": 1, + "vendorCompanyId": 10, + "companyName": "Example Seller", + "domain": domain, + "internalId": "", + "iabBuyerAgentApproval": True, + "iabBuyerAgentApprovedAt": "2026-03-01T00:00:00Z", + } + ) def _denied(domain: str) -> ApprovalRecord: - return ApprovalRecord.model_validate({ - "vendorId": 2, - "vendorCompanyId": 20, - "companyName": "Shady Seller", - "domain": domain, - "internalId": "", - "iabBuyerAgentApproval": False, - "iabBuyerAgentApprovedAt": None, - }) + return ApprovalRecord.model_validate( + { + "vendorId": 2, + "vendorCompanyId": 20, + "companyName": "Shady Seller", + "domain": domain, + "internalId": "", + "iabBuyerAgentApproval": False, + "iabBuyerAgentApprovedAt": None, + } + ) # --------------------------------------------------------------------------- @@ -295,4 +299,4 @@ def test_flow_omits_vendor_approval_tool_without_sgp(agency_context, monkeypatch buyer_context=agency_context, sgp_client=None, ) - assert flow._vendor_approval_tool is None \ No newline at end of file + assert flow._vendor_approval_tool is None