Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 26 additions & 1 deletion src/adcp/compat/legacy/v2_5/_media_buy_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,20 @@

from __future__ import annotations

import logging
from typing import Any
from urllib.parse import urlparse

from adcp.compat.legacy.v2_5._url import extract_brand_domain

_logger = logging.getLogger(__name__)

# Paths that v3 sellers reconstruct correctly — no information is lost.
_STANDARD_BRAND_MANIFEST_PATHS = {"", "/", "/.well-known/brand.json"}

# Per-URL dedup so high-RPS v2.5 buyers don't saturate the log pipeline.
_brand_manifest_path_warned: set[str] = set()


def adapt_brand_manifest_to_brand(payload: dict[str, Any]) -> dict[str, Any]:
"""Rewrite v2.5 ``brand_manifest`` (URL string or inline BrandManifest
Expand All @@ -34,7 +44,22 @@ def adapt_brand_manifest_to_brand(payload: dict[str, Any]) -> dict[str, Any]:
out = dict(payload)
manifest = out.pop("brand_manifest", None)
if isinstance(manifest, str) and manifest and "brand" not in out:
out["brand"] = {"domain": extract_brand_domain(manifest)}
domain = extract_brand_domain(manifest)
parsed = urlparse(manifest.strip())
if (
parsed.hostname is not None
and parsed.path not in _STANDARD_BRAND_MANIFEST_PATHS
and manifest not in _brand_manifest_path_warned
):
_brand_manifest_path_warned.add(manifest)
_logger.warning(
"brand_manifest at %s uses a non-standard path; "
"v3 sellers derive %s/.well-known/brand.json from BrandReference.domain. "
"Manifest fetch may 404.",
manifest,
domain,
)
out["brand"] = {"domain": domain}
elif isinstance(manifest, dict) and "brand" not in out:
url = manifest.get("url")
if isinstance(url, str) and url.strip():
Expand Down
27 changes: 26 additions & 1 deletion src/adcp/compat/legacy/v2_5/get_products.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,12 +36,22 @@

from __future__ import annotations

import logging
from typing import Any
from urllib.parse import urlparse

from adcp.compat.legacy import register_adapter
from adcp.compat.legacy.types import AdapterPair
from adcp.compat.legacy.v2_5._url import extract_brand_domain

_logger = logging.getLogger(__name__)

# Paths that v3 sellers reconstruct correctly — no information is lost.
_STANDARD_BRAND_MANIFEST_PATHS = {"", "/", "/.well-known/brand.json"}

# Per-URL dedup so high-RPS v2.5 buyers don't saturate the log pipeline.
_brand_manifest_path_warned: set[str] = set()

# v2.5 channel buckets to v3 channel slugs. Multi-mapped buckets resolve
# to all listed v3 channels; downstream consumers can narrow further via
# ``filters.channels`` if needed.
Expand Down Expand Up @@ -135,7 +145,22 @@ def adapt_request(payload: dict[str, Any]) -> dict[str, Any]:
# v3 validation handle the missing field.
brand_manifest = out.pop("brand_manifest", None)
if isinstance(brand_manifest, str) and brand_manifest and "brand" not in out:
out["brand"] = {"domain": extract_brand_domain(brand_manifest)}
domain = extract_brand_domain(brand_manifest)
parsed = urlparse(brand_manifest.strip())
if (
parsed.hostname is not None
and parsed.path not in _STANDARD_BRAND_MANIFEST_PATHS
and brand_manifest not in _brand_manifest_path_warned
):
_brand_manifest_path_warned.add(brand_manifest)
_logger.warning(
"brand_manifest at %s uses a non-standard path; "
"v3 sellers derive %s/.well-known/brand.json from BrandReference.domain. "
"Manifest fetch may 404.",
brand_manifest,
domain,
)
out["brand"] = {"domain": domain}
elif isinstance(brand_manifest, dict) and "brand" not in out:
url = brand_manifest.get("url")
if isinstance(url, str) and url.strip():
Expand Down
59 changes: 59 additions & 0 deletions tests/test_legacy_adapter_v2_5_get_products.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from __future__ import annotations

import logging
from typing import Any

from adcp.compat.legacy import get_legacy_adapter
Expand Down Expand Up @@ -139,6 +140,64 @@ def test_brand_manifest_inline_object_brand_wins_when_both_present() -> None:
assert "brand_manifest" not in out


# ---------------------------------------------------------------------------
# Request: brand_manifest non-standard-path warning (issue #683)
# ---------------------------------------------------------------------------

_GP_LOGGER = "adcp.compat.legacy.v2_5.get_products"


def test_brand_manifest_cdn_url_warns(caplog, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""CDN brand_manifest with non-standard path emits a WARNING about the
information loss so operators can debug downstream 404s."""
import adcp.compat.legacy.v2_5.get_products as _gp

monkeypatch.setattr(_gp, "_brand_manifest_path_warned", set())
with caplog.at_level(logging.WARNING, logger=_GP_LOGGER):
out = _adapt({"brand_manifest": "https://cdn.acmecorp.com/brand-manifest.json"})
assert out["brand"] == {"domain": "cdn.acmecorp.com"}
records = [r for r in caplog.records if r.name == _GP_LOGGER]
assert len(records) == 1
msg = records[0].getMessage()
assert "non-standard path" in msg
assert "cdn.acmecorp.com" in msg


def test_brand_manifest_well_known_path_no_warning(caplog, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""Standard /.well-known/brand.json path does NOT warn — v3 derives the
same URL that the buyer originally pointed at."""
import adcp.compat.legacy.v2_5.get_products as _gp

monkeypatch.setattr(_gp, "_brand_manifest_path_warned", set())
with caplog.at_level(logging.WARNING, logger=_GP_LOGGER):
_adapt({"brand_manifest": "https://acme.com/.well-known/brand.json"})
assert not any(r.name == _GP_LOGGER for r in caplog.records)


def test_brand_manifest_bare_domain_no_warning(caplog, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""Bare domain brand_manifest (no scheme) does not false-positive — urlparse
returns hostname=None for schemeless strings so we skip the path check."""
import adcp.compat.legacy.v2_5.get_products as _gp

monkeypatch.setattr(_gp, "_brand_manifest_path_warned", set())
with caplog.at_level(logging.WARNING, logger=_GP_LOGGER):
_adapt({"brand_manifest": "acme.com"})
assert not any(r.name == _GP_LOGGER for r in caplog.records)


def test_brand_manifest_cdn_url_warns_once_dedup(caplog, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""Same CDN URL repeated across requests only logs one warning (dedup)."""
import adcp.compat.legacy.v2_5.get_products as _gp

monkeypatch.setattr(_gp, "_brand_manifest_path_warned", set())
url = "https://cdn.acmecorp.com/brand-manifest.json"
with caplog.at_level(logging.WARNING, logger=_GP_LOGGER):
_adapt({"brand_manifest": url})
_adapt({"brand_manifest": url})
records = [r for r in caplog.records if r.name == _GP_LOGGER]
assert len(records) == 1


# ---------------------------------------------------------------------------
# Request: promoted_offerings → catalog
# ---------------------------------------------------------------------------
Expand Down
61 changes: 61 additions & 0 deletions tests/test_legacy_adapter_v2_5_media_buy.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from __future__ import annotations

import logging
from typing import Any

from adcp.compat.legacy import get_legacy_adapter
Expand Down Expand Up @@ -119,6 +120,66 @@ def test_create_media_buy_brand_manifest_inline_object_brand_wins_when_both_pres
assert "brand_manifest" not in out


# ---------------------------------------------------------------------------
# create_media_buy: brand_manifest non-standard-path warning (issue #683)
# ---------------------------------------------------------------------------

_MB_HELPERS_LOGGER = "adcp.compat.legacy.v2_5._media_buy_helpers"


def test_create_media_buy_cdn_url_warns(caplog, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""CDN brand_manifest with non-standard path emits a WARNING about the
information loss so operators can debug downstream 404s."""
import adcp.compat.legacy.v2_5._media_buy_helpers as _mbh

monkeypatch.setattr(_mbh, "_brand_manifest_path_warned", set())
with caplog.at_level(logging.WARNING, logger=_MB_HELPERS_LOGGER):
out = v2_5_cmb.adapt_request(
{"brand_manifest": "https://cdn.acmecorp.com/brand-manifest.json"}
)
assert out["brand"] == {"domain": "cdn.acmecorp.com"}
records = [r for r in caplog.records if r.name == _MB_HELPERS_LOGGER]
assert len(records) == 1
msg = records[0].getMessage()
assert "non-standard path" in msg
assert "cdn.acmecorp.com" in msg


def test_create_media_buy_well_known_path_no_warning(caplog, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""Standard /.well-known/brand.json path does NOT warn."""
import adcp.compat.legacy.v2_5._media_buy_helpers as _mbh

monkeypatch.setattr(_mbh, "_brand_manifest_path_warned", set())
with caplog.at_level(logging.WARNING, logger=_MB_HELPERS_LOGGER):
v2_5_cmb.adapt_request(
{"brand_manifest": "https://acme.com/.well-known/brand.json"}
)
assert not any(r.name == _MB_HELPERS_LOGGER for r in caplog.records)


def test_create_media_buy_bare_domain_no_warning(caplog, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""Bare domain brand_manifest (no scheme) does not false-positive."""
import adcp.compat.legacy.v2_5._media_buy_helpers as _mbh

monkeypatch.setattr(_mbh, "_brand_manifest_path_warned", set())
with caplog.at_level(logging.WARNING, logger=_MB_HELPERS_LOGGER):
v2_5_cmb.adapt_request({"brand_manifest": "acme.com"})
assert not any(r.name == _MB_HELPERS_LOGGER for r in caplog.records)


def test_create_media_buy_cdn_url_warns_once_dedup(caplog, monkeypatch) -> None: # type: ignore[no-untyped-def]
"""Same CDN URL repeated across requests only logs one warning (dedup)."""
import adcp.compat.legacy.v2_5._media_buy_helpers as _mbh

monkeypatch.setattr(_mbh, "_brand_manifest_path_warned", set())
url = "https://cdn.acmecorp.com/brand-manifest.json"
with caplog.at_level(logging.WARNING, logger=_MB_HELPERS_LOGGER):
v2_5_cmb.adapt_request({"brand_manifest": url})
v2_5_cmb.adapt_request({"brand_manifest": url})
records = [r for r in caplog.records if r.name == _MB_HELPERS_LOGGER]
assert len(records) == 1


# ---------------------------------------------------------------------------
# Package request: creative_ids → creative_assignments (both tools)
# ---------------------------------------------------------------------------
Expand Down
Loading