diff --git a/src/adcp/types/canonical_creative.py b/src/adcp/types/canonical_creative.py index e3d03db60..96014822d 100644 --- a/src/adcp/types/canonical_creative.py +++ b/src/adcp/types/canonical_creative.py @@ -76,16 +76,20 @@ import json import re from collections.abc import Callable, Iterable, Sequence -from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Protocol, TypeVar, cast +from functools import lru_cache +from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Protocol, TypeVar, cast, get_args from pydantic import ( AfterValidator, + AnyUrl, ConfigDict, Field, GetJsonSchemaHandler, + PlainSerializer, PrivateAttr, SerializerFunctionWrapHandler, WithJsonSchema, + WrapSerializer, field_validator, model_serializer, model_validator, @@ -290,13 +294,15 @@ def validate(value: str) -> str: def _walk_for_credential_keys(value: Any, *, path: str = "") -> str | None: """Return the first credential-shaped key path under ``value``.""" + if isinstance(value, AdCPBaseModel): + value = _model_boundary_values(value) if isinstance(value, dict): for key, nested in value.items(): + if isinstance(key, str) and key.lower().endswith(_CREDENTIAL_SHAPED_KEY_SUFFIXES): + return f"{path}.{key}" if path else str(key) + if not isinstance(nested, (dict, list, tuple, AdCPBaseModel)): + continue nested_path = f"{path}.{key}" if path else str(key) - if isinstance(key, str) and any( - key.lower().endswith(suffix) for suffix in _CREDENTIAL_SHAPED_KEY_SUFFIXES - ): - return nested_path found = _walk_for_credential_keys(nested, path=nested_path) if found is not None: return found @@ -305,11 +311,102 @@ def _walk_for_credential_keys(value: Any, *, path: str = "") -> str | None: found = _walk_for_credential_keys(nested, path=f"{path}[{index}]") if found is not None: return found - elif isinstance(value, AdCPBaseModel): - return _walk_for_credential_keys(value.model_dump(mode="python"), path=path) return None +def _has_boundary_serialization_metadata(annotation: Any) -> bool: + """Find functional serializers, including ones inside container annotations.""" + if isinstance(annotation, (PlainSerializer, WrapSerializer)): + return True + if hasattr(annotation, "__get_pydantic_core_schema__"): + is_adcp_model = isinstance(annotation, type) and issubclass(annotation, AdCPBaseModel) + if not is_adcp_model and getattr(annotation, "__module__", "").split(".")[0] != "pydantic": + return True + return any(_has_boundary_serialization_metadata(item) for item in get_args(annotation)) + + +@lru_cache(maxsize=1024) +def _direct_boundary_fields(cls: type[AdCPBaseModel]) -> frozenset[str] | None: + """Cache the fields of models whose stored values match their default dump.""" + if cls.model_dump not in (AdCPBaseModel.model_dump, CanonicalBoundaryModel.model_dump): + return None + if getattr(cls.__get_pydantic_core_schema__, "__func__", None) is not getattr( + AdCPBaseModel.__get_pydantic_core_schema__, "__func__", None + ): + return None + decorators = cls.__pydantic_decorators__ + if decorators.computed_fields or decorators.field_serializers: + return None + stock_serializers = ( + AdCPBaseModel._notification_config_wire_defaults, + _serialize_canonical_model, + ) + if any(item.func not in stock_serializers for item in decorators.model_serializers.values()): + return None + if cls.model_config.get("serialize_by_alias"): + return None + # This stock serializer conditionally drops a field based on sibling values. + if {"subscriber_id", "url", "event_types", "product_payload_view"} <= cls.model_fields.keys(): + return None + fields = set() + for name, field in cls.model_fields.items(): + if field.exclude: + continue + if field.exclude_if is not None or _has_boundary_serialization_metadata(field.annotation): + return None + if any(_has_boundary_serialization_metadata(item) for item in field.metadata): + return None + fields.add(name) + return frozenset(fields) + + +def _is_direct_boundary_value(value: Any, active: set[int] | None = None) -> bool: + """Other models and Python objects need Pydantic's conversion before scanning.""" + if value is None or isinstance(value, (str, int, float, bool, AnyUrl, AdCPBaseModel)): + return True + if not isinstance(value, (dict, list, tuple)): + return False + active = set() if active is None else active + identity = id(value) + if identity in active: + return False + active.add(identity) + try: + items = value.values() if isinstance(value, dict) else value + return all(_is_direct_boundary_value(item, active) for item in items) + finally: + active.remove(identity) + + +def _model_boundary_values(value: AdCPBaseModel) -> Any: + """Read ordinary fields directly and preserve custom serialization behavior. + + Canonical instances can be constructed or mutated without validation, so + they still need scanning. Private compatibility state is deliberately not + part of the public boundary. + """ + fields = _direct_boundary_fields(type(value)) + if fields is None: + return value.model_dump(mode="python") + # Match AdCPBaseModel.model_dump's default omission of null model fields; + # null values inside open dictionaries still retain their keys and checks. + values = { + **{ + name: item + for name, item in value.__dict__.items() + if name in fields and item is not None + }, + **{ + name: item + for name, item in (value.__pydantic_extra__ or {}).items() + if item is not None + }, + } + if not all(_is_direct_boundary_value(item) for item in values.values()): + return value.model_dump(mode="python") + return values + + class _JsonSchemaHandlerWithGenerator(Protocol): """The concrete handler pydantic passes in carries the active generator. @@ -324,14 +421,18 @@ class _JsonSchemaHandlerWithGenerator(Protocol): def is_legacy_creative_identity_key(key: object) -> bool: """Return whether *key* names legacy creative routing identity.""" - return isinstance(key, str) and bool(_LEGACY_IDENTITY_KEY.search(key)) + return isinstance(key, str) and _is_legacy_creative_identity_string(key) + + +@lru_cache(maxsize=1024) +def _is_legacy_creative_identity_string(key: str) -> bool: + return bool(_LEGACY_IDENTITY_KEY.search(key)) def _is_format_scoped_agent_tuple(value: dict[Any, Any], *, path: str) -> bool: """Recognize legacy format owners, including tuples with extension keys.""" - keys = set(value) - if not {"agent_url", "id"} <= keys: + if "agent_url" not in value or "id" not in value: return False # Several protocol surfaces intentionally carry ordinary agent, signal, # and list descriptors with the same structural pair. Match RC3's explicit @@ -371,10 +472,10 @@ def strip_legacy_creative_identity( path=_path, ) return { - key: strip_legacy_creative_identity( - item, - _path=f"{_path}.{key}", - _format_scope=_format_scope or (isinstance(key, str) and "format" in key.lower()), + key: ( + strip_legacy_creative_identity(item, _path=f"{_path}.{key}") + if isinstance(item, (dict, list, tuple)) + else item ) for key, item in value.items() if not is_legacy_creative_identity_key(key) @@ -382,19 +483,19 @@ def strip_legacy_creative_identity( } if isinstance(value, list): return [ - strip_legacy_creative_identity( - item, - _path=f"{_path}[{index}]", - _format_scope=_format_scope, + ( + strip_legacy_creative_identity(item, _path=f"{_path}[{index}]") + if isinstance(item, (dict, list, tuple)) + else item ) for index, item in enumerate(value) ] if isinstance(value, tuple): return tuple( - strip_legacy_creative_identity( - item, - _path=f"{_path}[{index}]", - _format_scope=_format_scope, + ( + strip_legacy_creative_identity(item, _path=f"{_path}[{index}]") + if isinstance(item, (dict, list, tuple)) + else item ) for index, item in enumerate(value) ) @@ -411,7 +512,7 @@ def _legacy_creative_identity_path( """Locate legacy creative identity in model input without mutating it.""" if isinstance(value, AdCPBaseModel): - value = value.model_dump(mode="python") + value = _model_boundary_values(value) if isinstance(value, dict): if _is_format_scoped_agent_tuple(value, path=path): return f"{path}.agent_url" @@ -420,6 +521,8 @@ def _legacy_creative_identity_path( if allow_root_v1_ref and path == "$" and key == "v1_format_ref": continue return f"{path}.{key}" + if not isinstance(nested, (dict, list, tuple, AdCPBaseModel)): + continue found = _legacy_creative_identity_path( nested, path=f"{path}.{key}", @@ -489,13 +592,23 @@ def _serialize_canonical_model( ) -> Any: """Enforce the boundary for nested and TypeAdapter serialization too.""" - # ``self`` is whatever pydantic is serializing at this position, which is not - # always a model: a field annotated with a canonical model can hold a plain - # dict (a mismatched value pydantic serializes with a warning). Read the - # capability off the type defensively, because ``dict`` does not carry it. - return strip_legacy_creative_identity( - handler(self), - _format_scope=getattr(type(self), "__adcp_format_declaration_scope__", False), + # Each serializer remains its own boundary: a custom serializer can start + # an independent TypeAdapter dump and embed that result as a JSON string, + # which an enclosing cleanup pass cannot inspect. + return strip_legacy_creative_identity(handler(self)) + + +@lru_cache(maxsize=128) +def _uses_canonical_serializer(model: type[AdCPBaseModel]) -> bool: + """Whether Pydantic selected our inherited canonical wrap serializer.""" + if getattr(model.__get_pydantic_core_schema__, "__func__", None) is not getattr( + AdCPBaseModel.__get_pydantic_core_schema__, "__func__", None + ): + return False + serializers = model.__pydantic_decorators__.model_serializers + return ( + bool(serializers) + and next(reversed(serializers.values())).func is _serialize_canonical_model ) @@ -581,23 +694,24 @@ def _reject_legacy_creative_identity(cls, value: Any) -> Any: def model_dump(self, **kwargs: Any) -> dict[str, Any]: kwargs.setdefault("serialize_as_any", False) - stripped: dict[str, Any] = strip_legacy_creative_identity( - super().model_dump(**kwargs), - _format_scope=type(self).__adcp_format_declaration_scope__, - ) - return stripped + value = super().model_dump(**kwargs) + if _uses_canonical_serializer(type(self)): + return value + # A subclass can replace the inherited serializer. Keep the public + # method's boundary for its output without rescanning ordinary dumps. + cleaned: dict[str, Any] = strip_legacy_creative_identity(value) + return cleaned def model_dump_json(self, **kwargs: Any) -> str: kwargs.setdefault("serialize_as_any", False) raw = super().model_dump_json(**kwargs) - clean = strip_legacy_creative_identity( - json.loads(raw), - _format_scope=type(self).__adcp_format_declaration_scope__, - ) + if _uses_canonical_serializer(type(self)): + return raw + clean = strip_legacy_creative_identity(json.loads(raw)) indent = kwargs.get("indent") return json.dumps( clean, - ensure_ascii=False, + ensure_ascii=kwargs.get("ensure_ascii", False), indent=indent, separators=None if indent is not None else (",", ":"), ) diff --git a/tests/conformance/reporting/test_reporting_change_feed_consumers_pg.py b/tests/conformance/reporting/test_reporting_change_feed_consumers_pg.py index 166c2587b..edf3a1a17 100644 --- a/tests/conformance/reporting/test_reporting_change_feed_consumers_pg.py +++ b/tests/conformance/reporting/test_reporting_change_feed_consumers_pg.py @@ -7,6 +7,7 @@ from __future__ import annotations import asyncio +from contextlib import asynccontextmanager from typing import Any import pytest @@ -158,46 +159,79 @@ async def test_a_change_of_feed_scope_replaces_the_position() -> None: assert await saved(pool, "sink") == ("acct_a", None, 900) -async def test_concurrent_writers_cannot_interleave_a_stale_position_over_a_newer_one() -> None: - """A newer acknowledgment is queued first and a stale one second, both behind a row lock. +async def test_concurrent_writers_cannot_interleave_a_stale_position_over_a_newer_one( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A stale acknowledgment waits on the newer acknowledgment's uncommitted update. - The unconditional upsert let the stale write land last; the conditional one re-evaluates - against the committed row once the lock is released and skips it. + Merely queuing both writers behind a third transaction does not order their writes. + Hold the real newer write before commit, then prove the stale write is blocked by it. + The conditional upsert must re-evaluate against the committed row and skip the stale write. """ async with isolated_reporting_pool() as pool: store = await feed_store(pool) await store.save_feed_consumer_cursor("sink", account_cursor(1)) - holder = await pool.getconn() - try: - await holder.execute( - "UPDATE adcp_reporting_feed_consumers SET cursor_seq = cursor_seq" - " WHERE consumer_name = 'sink'" - ) - newer = asyncio.create_task(store.save_feed_consumer_cursor("sink", account_cursor(9))) - await blocked_writers(pool, 1) - stale = asyncio.create_task(store.save_feed_consumer_cursor("sink", account_cursor(5))) - await blocked_writers(pool, 2) - await holder.commit() - finally: - await pool.putconn(holder) - - results = (await newer, await stale) + newer: asyncio.Task[bool] | None = None + stale: asyncio.Task[bool] | None = None + newer_written = asyncio.Event() + stale_connected = asyncio.Event() + commit_newer = asyncio.Event() + pids: dict[str, int] = {} + checkout = pool.connection + + @asynccontextmanager + async def connection(*args: Any, **kwargs: Any): + async with checkout(*args, **kwargs) as current: + task = asyncio.current_task() + if task is newer: + pids["newer"] = current.info.backend_pid + elif task is stale: + pids["stale"] = current.info.backend_pid + stale_connected.set() + yield current + if task is newer: + newer_written.set() + await commit_newer.wait() + + with monkeypatch.context() as patch: + patch.setattr(pool, "connection", connection) + try: + newer = asyncio.create_task( + store.save_feed_consumer_cursor("sink", account_cursor(9)) + ) + await asyncio.wait_for(newer_written.wait(), timeout=5) + stale = asyncio.create_task( + store.save_feed_consumer_cursor("sink", account_cursor(5)) + ) + await asyncio.wait_for(stale_connected.wait(), timeout=5) + await blocked_writer(pool, pids["stale"], pids["newer"]) + assert not stale.done(), "the stale writer must overlap the uncommitted newer write" + commit_newer.set() + results = tuple(await asyncio.wait_for(asyncio.gather(newer, stale), timeout=10)) + finally: + commit_newer.set() + tasks = [task for task in (newer, stale) if task is not None] + for task in tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + assert await saved(pool, "sink") == ("acct_a", None, 9), "a stale write landed last" assert results == (True, False) -async def blocked_writers(pool: Any, count: int) -> None: +async def blocked_writer(pool: Any, writer_pid: int, blocker_pid: int) -> None: for _ in range(400): rows = await fetch_all( pool, - "SELECT count(*) FROM pg_stat_activity WHERE wait_event_type = 'Lock'" - " AND query LIKE 'INSERT INTO adcp_reporting_feed_consumers%'", + "SELECT %s = ANY(pg_blocking_pids(%s))", + (blocker_pid, writer_pid), ) - if rows[0][0] >= count: + if rows[0][0]: return await asyncio.sleep(0.025) - pytest.fail(f"{count} writers never queued behind the row lock") + pytest.fail("the stale writer never blocked on the newer writer's transaction") async def test_remove_feed_consumer_releases_the_hold() -> None: diff --git a/tests/interop/reporting_row_storage_js_driver.cjs b/tests/interop/reporting_row_storage_js_driver.cjs index 3d78f7951..1e56f08e9 100644 --- a/tests/interop/reporting_row_storage_js_driver.cjs +++ b/tests/interop/reporting_row_storage_js_driver.cjs @@ -2,7 +2,7 @@ /** * JavaScript half of the cross-SDK row storage interop test. * - * Drives the shipped `@adcp/sdk` row storage (`ReportingRowStorageV1`) against a + * Drives the shipped JavaScript row storage (`ReportingRowStorageV1`) against a * PostgreSQL schema and a filesystem bucket root shared with the Python SDK. * * node reporting_row_storage_js_driver.cjs write --out js-written.json @@ -36,6 +36,11 @@ const sdkEntry = fs.existsSync(path.join(sdkDir, 'dist/lib/reporting/ledger/inde ? path.join(sdkDir, 'dist/lib/reporting/ledger/index.js') : require.resolve('@adcp/sdk/reporting/ledger', { paths: [sdkDir] }); const ledger = require(sdkEntry); +// SDK 16 moved the storage implementations into their own workspace package. +const rowStorage = typeof ledger.ReportingRowStorageV1 === 'function' + && typeof ledger.createFilesystemReportingRowObjectProviderV1 === 'function' + ? ledger + : requireFromSdk('@adcp/reporting-storage'); const canonicalize = fs.existsSync(path.join(sdkDir, 'dist/lib/utils/jcs.js')) ? require(path.join(sdkDir, 'dist/lib/utils/jcs.js')).canonicalize : requireFromSdk('@adcp/sdk').canonicalize; @@ -71,7 +76,7 @@ function rowsOnly(rows) { async function main() { const pool = new Pool({ connectionString: url, options: `-c search_path="${schema}"` }); - const storage = new ledger.ReportingRowStorageV1(pool, { + const storage = new rowStorage.ReportingRowStorageV1(pool, { deploymentNamespace: namespace, bindings: { inline: { kind: 'postgres' }, @@ -83,7 +88,7 @@ async function main() { compression: 'gzip', }, }, - providers: { filesystem: ledger.createFilesystemReportingRowObjectProviderV1({ roots: { reports: root } }) }, + providers: { filesystem: rowStorage.createFilesystemReportingRowObjectProviderV1({ roots: { reports: root } }) }, }); await storage.ready(); try { diff --git a/tests/test_canonical_boundary_serialization.py b/tests/test_canonical_boundary_serialization.py new file mode 100644 index 000000000..0e1d0fa88 --- /dev/null +++ b/tests/test_canonical_boundary_serialization.py @@ -0,0 +1,277 @@ +"""Canonical serializers avoid redundant public-method cleanup (#1519).""" + +from __future__ import annotations + +import json +from concurrent.futures import ThreadPoolExecutor +from threading import Barrier +from typing import Any + +import pytest +from pydantic import ( + BaseModel, + Field, + TypeAdapter, + ValidationError, + field_serializer, + model_serializer, +) +from pydantic_core import PydanticSerializationError, core_schema + +from adcp.types import Format, Product, ProductFormatDeclaration +from adcp.types import canonical_creative as boundary +from adcp.types.base import AdCPBaseModel + + +def _declaration() -> Format: + return Format.model_construct( + format_kind="image", + params={"width": 300, "nested": [{"format_id": "hidden", "keep": "yes"}]}, + vendor={"agent_url": "https://legacy.example", "id": "hidden"}, + ) + + +@pytest.mark.parametrize("method", ["python", "json", "adapter_python", "adapter_json", "parent"]) +def test_nested_canonical_models_are_cleaned_once_per_model( + method: str, monkeypatch: pytest.MonkeyPatch +) -> None: + class Catalog(boundary.CanonicalBoundaryModel): + options: list[Format] + + class Envelope(BaseModel): + catalog: Catalog + + catalog = Catalog.model_construct(options=[_declaration(), _declaration()]) + strip = boundary.strip_legacy_creative_identity + roots = 0 + + def counted(value: Any, **kwargs: Any) -> Any: + nonlocal roots + if kwargs.get("_path", "$") == "$": + roots += 1 + return strip(value, **kwargs) + + monkeypatch.setattr(boundary, "strip_legacy_creative_identity", counted) + if method == "python": + result = catalog.model_dump() + elif method == "json": + result = json.loads(catalog.model_dump_json()) + elif method == "adapter_python": + result = TypeAdapter(Catalog).dump_python(catalog, exclude_none=True) + elif method == "adapter_json": + result = json.loads(TypeAdapter(Catalog).dump_json(catalog, exclude_none=True)) + else: + result = Envelope(catalog=catalog).model_dump(exclude_none=True)["catalog"] + + # Two declarations and their catalog each enforce their own boundary. + # The public dump methods must not add a fourth pass over the full tree. + assert roots == 3 + assert result == { + "options": [ + { + "format_kind": "image", + "params": {"width": 300, "nested": [{"keep": "yes"}]}, + "vendor": {"id": "hidden"}, + } + ] + * 2 + } + + +def test_serialization_flag_is_reset_after_an_error() -> None: + class Failing(boundary.CanonicalBoundaryModel): + value: int + + @field_serializer("value") + def fail(self, value: int) -> int: + raise ValueError("serialization failed") + + with pytest.raises(PydanticSerializationError, match="serialization failed"): + Failing(value=1).model_dump() + assert _declaration().model_dump()["params"]["nested"] == [{"keep": "yes"}] + + +def test_concurrent_dumps_each_enforce_the_boundary() -> None: + barrier = Barrier(2) + + class Concurrent(boundary.CanonicalBoundaryModel): + payload: dict[str, Any] + + @field_serializer("payload") + def synchronize(self, value: dict[str, Any]) -> dict[str, Any]: + barrier.wait(timeout=10) + return value + + obj = Concurrent.model_construct(payload={"format_id": "hidden", "keep": True}) + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(lambda _: obj.model_dump(), range(2))) + assert results == [{"payload": {"keep": True}}] * 2 + + +def test_dump_options_and_caller_context_are_preserved() -> None: + class Labeled(boundary.CanonicalBoundaryModel): + label: str = Field(serialization_alias="display_label") + unused: str | None = None + + @field_serializer("label") + def prefix(self, value: str, info: Any) -> str: + return info.context["prefix"] + value + + context = {"prefix": "é"} + obj = Labeled(label="x") + options = {"by_alias": True, "exclude_unset": True, "context": context} + assert obj.model_dump(**options) == {"display_label": "éx"} + assert obj.model_dump_json(indent=2, ensure_ascii=True, **options) == ( + '{\n "display_label": "\\u00e9x"\n}' + ) + assert context == {"prefix": "é"} + + +@pytest.mark.parametrize( + "method", ["model_dump", "model_dump_json", "adapter_python", "adapter_json"] +) +def test_explicit_dumps_inside_custom_serializers_keep_their_own_boundary(method: str) -> None: + results = [] + + class Embedded(boundary.CanonicalBoundaryModel): + declaration: Format + + @field_serializer("declaration") + def embed(self, value: Format) -> str: + if method == "adapter_python": + dumped = TypeAdapter(Format).dump_python(value, exclude_none=True) + elif method == "adapter_json": + dumped = TypeAdapter(Format).dump_json(value, exclude_none=True).decode() + else: + dumped = getattr(value, method)() + results.append(dumped) + return dumped if isinstance(dumped, str) else json.dumps(dumped) + + obj = Embedded.model_construct(declaration=_declaration()) + embedded = json.loads(obj.model_dump()["declaration"]) + assert embedded["params"]["nested"] == [{"keep": "yes"}] + assert "agent_url" not in embedded["vendor"] + assert len(results) == 1 + + +@pytest.mark.parametrize("method", ["model_dump", "model_dump_json"]) +def test_replacement_model_serializers_keep_the_public_dump_boundary(method: str) -> None: + class Customized(Format): + @model_serializer(mode="plain") + def custom(self) -> dict[str, Any]: + return { + "format_kind": self.format_kind, + "format_id": "hidden", + "params": {"format_ids": ["hidden"], "keep": True}, + } + + result = getattr(Customized(format_kind="image", params={}), method)() + if isinstance(result, str): + result = json.loads(result) + assert result == {"format_kind": "image", "params": {"keep": True}} + + +@pytest.mark.parametrize("method", ["model_dump", "model_dump_json"]) +def test_custom_core_serializers_keep_the_public_dump_boundary(method: str) -> None: + class Customized(Format): + @classmethod + def __get_pydantic_core_schema__(cls, source: Any, handler: Any) -> Any: + schema = handler(source) + schema["serialization"] = core_schema.plain_serializer_function_ser_schema( + lambda value: {"format_kind": value.format_kind, "format_id": "hidden"} + ) + return schema + + result = getattr(Customized(format_kind="image", params={}), method)() + if isinstance(result, str): + result = json.loads(result) + assert result == {"format_kind": "image"} + + +def test_product_validation_does_not_serialize_nested_declarations( + monkeypatch: pytest.MonkeyPatch, +) -> None: + declaration = ProductFormatDeclaration( + format_kind="image", params={"width": 300, "height": 250} + ) + + def unexpected_dump(*args: Any, **kwargs: Any) -> Any: + pytest.fail("Input scanning must not serialize nested models") + + monkeypatch.setattr(BaseModel, "model_dump", unexpected_dump) + product = Product.model_validate( + { + "product_id": "display", + "name": "Display", + "description": "Display inventory", + "publisher_properties": [{"publisher_domain": "example.com", "selection_type": "all"}], + "channels": ["display"], + "delivery_type": "guaranteed", + "reporting_capabilities": { + "available_reporting_frequencies": ["daily"], + "expected_delay_minutes": 60, + "timezone": "UTC", + "supports_webhooks": False, + "available_metrics": ["impressions"], + "date_range_support": "date_range", + }, + "format_options": [declaration], + "pricing_options": [ + { + "pricing_option_id": "cpm", + "pricing_model": "cpm", + "currency": "USD", + "fixed_price": 1, + } + ], + } + ) + assert product.format_options[0] is declaration + + +@pytest.mark.parametrize("canonical", [False, True]) +def test_validation_still_scans_unvalidated_or_mutated_nested_models(canonical: bool) -> None: + class Extension(AdCPBaseModel): + payload: dict[str, Any] + + if canonical: + declaration = Format(format_kind="image", params={}) + declaration.params["format_id"] = "hidden" + nested: AdCPBaseModel = declaration + else: + nested = Extension.model_construct(payload={"format_id": "hidden"}) + with pytest.raises(ValidationError, match="legacy creative identity"): + Format(format_kind="image", params={}, extension=nested) + + +def test_credential_scan_reads_nested_model_fields_and_extras_without_serializing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class Extension(boundary.CanonicalBoundaryModel): + payload: dict[str, Any] + + def unexpected_dump(*args: Any, **kwargs: Any) -> Any: + pytest.fail("Credential scanning must not serialize nested models") + + monkeypatch.setattr(BaseModel, "model_dump", unexpected_dump) + for nested in ( + Extension.model_construct(payload={"api_token": "hidden"}), + Extension.model_construct(payload={}, vendor={"password": "hidden"}), + ): + with pytest.raises(ValidationError, match="credential-shaped key"): + Format(format_kind="image", params={}, extension=nested) + + +def test_nested_optional_model_fields_keep_the_dump_default_null_semantics() -> None: + class Extension(AdCPBaseModel): + format_id: str | None = None + access_token: str | None = None + + declaration = Format(format_kind="image", params={}, extension=Extension()) + assert declaration.model_dump()["extension"] == {} + + # Nulls in a plain extension dictionary still carry the forbidden key. + with pytest.raises(ValidationError, match="legacy creative identity"): + Format(format_kind="image", params={}, extension={"format_id": None}) + with pytest.raises(ValidationError, match="credential-shaped key"): + Format(format_kind="image", params={}, extension={"access_token": None}) diff --git a/tests/test_canonical_boundary_validation.py b/tests/test_canonical_boundary_validation.py new file mode 100644 index 000000000..609b8f001 --- /dev/null +++ b/tests/test_canonical_boundary_validation.py @@ -0,0 +1,186 @@ +"""Direct boundary scans preserve Pydantic's serialization semantics.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Annotated, Any + +import pytest +from pydantic import ( + BaseModel, + ConfigDict, + Field, + PlainSerializer, + ValidationError, + computed_field, + field_serializer, + model_serializer, +) +from pydantic_core import core_schema + +from adcp.types import Format +from adcp.types import canonical_creative as boundary +from adcp.types.base import AdCPBaseModel + + +def _format(extension: AdCPBaseModel) -> Format: + return Format(format_kind="image", params={}, extension=extension) + + +@pytest.mark.parametrize("key", ["access_token", "format_id"]) +def test_computed_fields_are_scanned(key: str) -> None: + class Computed(AdCPBaseModel): + @computed_field + @property + def value(self) -> dict[str, str]: + return {key: "hidden"} + + with pytest.raises(ValidationError): + _format(Computed()) + + +def test_excluded_fields_are_not_scanned() -> None: + class Internal(AdCPBaseModel): + access_token: str = Field(exclude=True) + format_id: str = Field(exclude=True) + keep: str + + result = _format(Internal(access_token="hidden", format_id="hidden", keep="visible")) + assert result.model_dump()["extension"] == {"keep": "visible"} + + +def test_conditional_exclusions_are_respected() -> None: + class Conditional(AdCPBaseModel): + access_token: str = Field(exclude_if=lambda value: value == "internal") + + assert _format(Conditional(access_token="internal")).model_dump()["extension"] == {} + with pytest.raises(ValidationError, match="credential-shaped key"): + _format(Conditional(access_token="visible")) + + +def test_custom_model_serializers_are_scanned() -> None: + class Custom(AdCPBaseModel): + @model_serializer(mode="plain") + def serialize(self) -> dict[str, str]: + return {"access_token": "hidden"} + + with pytest.raises(ValidationError, match="credential-shaped key"): + _format(Custom()) + + +def test_custom_field_serializers_are_scanned() -> None: + class Custom(AdCPBaseModel): + value: str + + @field_serializer("value") + def serialize(self, value: str) -> dict[str, str]: + return {"access_token": value} + + with pytest.raises(ValidationError, match="credential-shaped key"): + _format(Custom(value="hidden")) + + +def test_serializers_inside_container_annotations_are_scanned() -> None: + class Custom(boundary.CanonicalBoundaryModel): + values: list[Annotated[str, PlainSerializer(lambda value: {"access_token": value})]] + + with pytest.raises(ValidationError, match="credential-shaped key"): + _format(Custom(values=["hidden"])) + + +def test_custom_type_schema_serializers_are_scanned() -> None: + class CustomScalar(str): + @classmethod + def __get_pydantic_core_schema__(cls, source: Any, handler: Any) -> Any: + return core_schema.str_schema( + serialization=core_schema.plain_serializer_function_ser_schema( + lambda value: {"access_token": value} + ) + ) + + class Custom(boundary.CanonicalBoundaryModel): + value: CustomScalar + + with pytest.raises(ValidationError, match="credential-shaped key"): + _format(Custom(value="hidden")) + + +def test_custom_model_dump_is_scanned() -> None: + class Custom(AdCPBaseModel): + def model_dump(self, **kwargs: Any) -> dict[str, Any]: + return {"access_token": "hidden"} + + with pytest.raises(ValidationError, match="credential-shaped key"): + _format(Custom()) + + +def test_default_serialization_aliases_are_scanned() -> None: + class Aliased(AdCPBaseModel): + model_config = ConfigDict(serialize_by_alias=True) + value: str = Field(serialization_alias="access_token") + + with pytest.raises(ValidationError, match="credential-shaped key"): + _format(Aliased(value="hidden")) + + +@pytest.mark.parametrize("kind", ["model", "dataclass"]) +def test_nested_python_objects_are_converted_before_scanning(kind: str) -> None: + class Nested(BaseModel): + access_token: str + + @dataclass + class Data: + access_token: str + + class Extension(AdCPBaseModel): + value: Any + + nested = Nested(access_token="hidden") if kind == "model" else Data(access_token="hidden") + with pytest.raises(ValidationError, match="credential-shaped key"): + _format(Extension(value={"nested": [nested]})) + + +def test_plain_models_keep_the_direct_scan() -> None: + class Plain(AdCPBaseModel): + keep: str + internal: str = Field(exclude=True) + + instance = Plain(keep="visible", internal="hidden") + assert boundary._direct_boundary_fields(Plain) == frozenset({"keep"}) + assert boundary._model_boundary_values(instance) == {"keep": "visible"} + assert boundary._direct_boundary_fields(Format) is not None + + +def test_macro_graph_with_urls_keeps_the_direct_scan(monkeypatch: pytest.MonkeyPatch) -> None: + declaration = Format( + format_kind="image", + params={}, + macro_resolution_capabilities=[ + { + "dialect": "adcp", + "dialect_semantic": "CACHEBUSTER", + "mapping_status": "verified_universal", + "universal_semantic": "CACHEBUSTER", + "operation": "translate_to_native", + "performed_by": "seller", + "supported_contexts": ["url_query_value"], + "translation_target": { + "token": "%%CACHEBUSTER%%", + "dialect": "vendor", + "dialect_namespace": "https://example.com/macros", + "dialect_revision": "1", + "dialect_semantic": "CACHEBUSTER", + "mapping_status": "verified_universal", + "universal_semantic": "CACHEBUSTER", + "performed_by": "source_ad_server", + "encoding": {"kind": "none", "depth": 0}, + }, + } + ], + ) + + def unexpected_dump(*args: Any, **kwargs: Any) -> Any: + pytest.fail("Ordinary format and macro graphs should use the direct scan") + + monkeypatch.setattr(BaseModel, "model_dump", unexpected_dump) + assert boundary._legacy_creative_identity_path(declaration) is None