diff --git a/docs/reporting-release-notes.md b/docs/reporting-release-notes.md index 575ad019e3..9cd6bcdf80 100644 --- a/docs/reporting-release-notes.md +++ b/docs/reporting-release-notes.md @@ -1,5 +1,63 @@ # Reliable Reporting upgrade and release notes +## AdCP 3.3 reporting rules + +AdCP 3.3 changes several reporting rules +([#1669](https://github.com/adcontextprotocol/adcp-client-python/issues/1669)). +Here is what the SDK does for each, and what stays with the code that produces +delivery. + +- **Delivery rows may omit `impressions`.** The generated delivery row types make + the field optional. Absence means "not measured", never zero. Only + `aggregated_totals` still requires it. `InlineReportingSource` stages such a + row without an `impressions` key. It publishes no `impressions` control total + unless every staged row carries the metric, so an unmeasured row is never + summed as zero. No SDK reader derives `cpm`, `ctr`, `completion_rate` or + `viewable_rate` from delivery rows. A seller that has no impression count for + a row omits the field, along with that row's impression-denominated rates. In a + response served below 3.3, it omits the row instead of sending + `impressions: 0`. +- **Multi-qualifier metrics omit flat totals.** The SDK builds no buy or window + `totals`. A seller whose packages report one `metric_id` under more than one + qualifier set omits that metric's flat carrier, such as `viewability`, from + buy totals, window totals and per-buy webhook totals. Buyers read the package + rows instead. +- **Receivers ignore `aggregated_totals` in delivery webhooks.** `WebhookReceiver` + and `ADCPClient.handle_webhook` accept a `media_buy_delivery` result that + carries it. Nothing in the SDK reads it. On `WebhookReceiver`'s typed result + it is an unmodelled extra; `handle_webhook` returns the result mapping as + received. Do not treat it as authoritative. Reconcile from + `get_media_buy_delivery` or the per-buy rows. Senders still must not include it. +- **Property-grain rollups.** Under the `property_delivery_reporting` and + `installment_property_delivery_reporting` storyboards, a seller builds + `by_property`, `by_installment` and `by_installment_property` only from the + rows that `simulate_delivery` injects. The test-controller dispatcher rejects + these arrays with `INVALID_PARAMS` when they come with `delivery_date`, + because the arrays are undated. It also rejects them when the store's + `simulate_delivery` override does not declare the parameter: dropping the rows + would report success for rows the seller never received. The store rejects + multi-package buys itself. +- **`source.custody` on report definitions.** A definition's `source` may claim + `system_of_record` or `upstream_derived`. The SDK hashes installed definition + bytes exactly as given, so `report_definition_sha256` covers the claim. + Adding the claim to a published definition therefore needs a new digest. + `ReportingRevisionVerifier.custody` returns the claim, or `None` when the + definition makes none. A value outside the enum fails construction with + `UNSUPPORTED_VERIFICATION`. The buyer-side `ManifestReportingInspector` + validates definitions with the generated `ReportingReportDefinition`, which + models the field. +- **Superseded snapshot compaction is not permitted under 3.3.** The notice + takes effect no earlier than the first minor after November 20, 2026. The + ledger keeps every revision's content for its full retention window, which + stays conformant. When compaction becomes permitted, an exact read of a + compacted revision must return the same nondisclosing `REFERENCE_NOT_FOUND` + as an unknown ID, with `field: "reporting_revision_id"`. The SDK's not-found + answers do not set `field` yet. A revision still owed whose content cannot be + served stays a `REVISION_UNREADABLE` issue, served as `SERVICE_UNAVAILABLE`, + never a not-found. Buyers that need a superseded snapshot's rows should + persist them when read, or record a receipt or status statement naming the + revision. + ## Protocol-shape control totals New Core revisions bind `{name, value, value_type[, unit]}` control totals in diff --git a/src/adcp/reporting/materializer/verification.py b/src/adcp/reporting/materializer/verification.py index 3fd36dcc7d..a63f6a842f 100644 --- a/src/adcp/reporting/materializer/verification.py +++ b/src/adcp/reporting/materializer/verification.py @@ -65,6 +65,7 @@ object_path, ) from adcp.reporting.revision_selection import select_reporting_revision +from adcp.types.domains.enums.reporting_custody import ReportingCustody _VERIFICATION_SEAL_KEY = secrets.token_bytes(32) @@ -167,6 +168,10 @@ class ReportingRevisionVerifier(_ClosedValue): (value_type, optional unit, decimal scale). This explicit pinned subset has no expression evaluator, custom callbacks, mutable registration or network. Unsupported formats and definition semantics fail during construction. + + The definition bytes are hashed exactly as installed, so every field they + carry, including AdCP 3.3's optional ``source.custody``, is covered by + ``report_definition_sha256``. :attr:`custody` exposes that claim. """ key: ReportingVerificationKey @@ -207,6 +212,7 @@ def _runtime(self) -> _Canonicalizer: if hashlib.sha256(raw).hexdigest() != expected.lower(): raise failure("UNSUPPORTED_VERIFICATION") report = _mapping(parse_reporting_json(self.definition_bytes, self.limits)) + _definition_custody(report) schema = _mapping(parse_reporting_json(self.schema_bytes, self.limits)) contract = _mapping(parse_reporting_json(self.canonicalization_bytes, self.limits)) if ( @@ -337,6 +343,22 @@ def canonicalize( """Useful to trusted publishers preparing expected evidence before storage.""" return self._runtime().canonicalize(rows) + @property + def custody(self) -> ReportingCustody | None: + """The installed definition's ``source.custody`` claim, or ``None`` if it makes none. + + AdCP 3.3 adds the optional claim: ``system_of_record`` when the producer + controls the numbers, ``upstream_derived`` when it obtained them from the + provider and system its ``source`` names. The definition digest fixes it + for every revision, so no revision carries its own copy. Read it with a + revision's ``finality_basis``: an ``upstream_derived`` definition that + finalizes by ``contractual_cutoff`` closes figures the producer does not + control. Construction already rejected a value outside the enum. + """ + return _definition_custody( + _mapping(parse_reporting_json(self.definition_bytes, self.limits)) + ) + @dataclass(frozen=True, slots=True) class ReportingRevisionVerifierRegistry(_ClosedValue): @@ -722,6 +744,17 @@ def _mapping(value: object) -> dict[str, Any]: return cast(dict[str, Any], value) +def _definition_custody(report: dict[str, Any]) -> ReportingCustody | None: + """``source.custody`` from a parsed report definition; fail closed on a malformed claim.""" + source = report.get("source") + if type(source) is not dict or "custody" not in source: + return None + custody = source["custody"] + if type(custody) is not str or custody not in {c.value for c in ReportingCustody}: + raise failure("UNSUPPORTED_VERIFICATION") + return ReportingCustody(custody) + + def _local_schema(schema: dict[str, Any]) -> None: stack: list[object] = [schema] while stack: diff --git a/src/adcp/server/test_controller.py b/src/adcp/server/test_controller.py index 3b7b6430b1..b4a0a9cda3 100644 --- a/src/adcp/server/test_controller.py +++ b/src/adcp/server/test_controller.py @@ -169,6 +169,16 @@ def __call__(self, ref: dict[str, Any] | None, *, auth_info: AuthInfo | None = N } +#: ``simulate_delivery`` params whose rows the seller must report verbatim +#: (AdCP 3.3): ``by_property``, ``by_installment`` and ``by_installment_property`` +#: are built only from them. +_INJECTED_DELIVERY_ROWS = ( + "property_delivery", + "installment_delivery", + "installment_property_delivery", +) + + class TestControllerError(Exception): """Typed error for test controller store methods. @@ -491,7 +501,20 @@ async def simulate_delivery( ``installment_property_delivery`` (AdCP 3.3) carry the property-grain, installment and installment-by-property rows the seller reports under ``supports_property_breakdown`` / - ``supports_installment_property_breakdown``. + ``supports_installment_property_breakdown``. Build ``by_property``, + ``by_installment`` and ``by_installment_property`` from exactly these + rows (applying ``limit``, ``sort_by`` and ``sort_direction``), never + from catalog eligibility. Each array replaces the previous one, is not + dated and is not added to totals. Reject a multi-package buy with + ``TestControllerError("INVALID_PARAMS", ...)``. The dispatcher already + rejects these arrays alongside ``delivery_date``, and rejects them when + an override does not declare the parameter, rather than dropping rows + the seller would then fail to report. + + ``impressions`` is optional. When a batch carries none, record no + impression count rather than zero: AdCP 3.3 delivery rows omit + ``impressions`` when it was not measured, and a response served below + 3.3 omits such a row instead of reporting ``impressions: 0``. Returns: {"simulated": {...}, "cumulative": {...} | None} @@ -1382,6 +1405,22 @@ async def _handle_test_controller( method_kwargs["result"] = result_value elif scenario == "simulate_delivery": _require_scenario_params(scenario_params, "media_buy_id") + injected = [name for name in _INJECTED_DELIVERY_ROWS if name in scenario_params] + if injected and "delivery_date" in scenario_params: + # The injected arrays replace the previous rows and are never + # dated (AdCP 3.3), so a dated batch cannot carry them. + return _controller_error( + "INVALID_PARAMS", + f"{injected[0]} rows are undated; delivery_date is not allowed with them", + ) + unsupported = [name for name in injected if name not in method_kwargs] + if unsupported: + # The seller must build the breakdown from exactly these rows. + # Dropping them would report success for rows it never received. + return _controller_error( + "INVALID_PARAMS", + f"{unsupported[0]} is not supported by this agent's simulate_delivery", + ) elif scenario == "force_media_buy_confirmation": _require_scenario_params(scenario_params, "action") action = scenario_params["action"] diff --git a/tests/test_reporting_33_rules.py b/tests/test_reporting_33_rules.py new file mode 100644 index 0000000000..e246bd87fc --- /dev/null +++ b/tests/test_reporting_33_rules.py @@ -0,0 +1,488 @@ +"""AdCP 3.3 reporting rules the SDK owns (#1669). + +* Delivery rows may omit ``impressions``: an absent count is "not measured", + never zero, and no impression total is derived from rows that lack it. +* Receivers ignore ``aggregated_totals`` in delivery webhooks and never reject + a payload for carrying it. +* Reliable Reporting report definitions may carry ``source.custody``; the + definition digest covers it and the installed verifier exposes it. +* Property-grain rollups are built only from rows injected through + ``simulate_delivery``. +""" + +from __future__ import annotations + +import asyncio +import copy +import hashlib +import json +from datetime import datetime, timezone +from importlib.resources import files +from pathlib import Path +from typing import Any + +import pytest + +from adcp.client import ADCPClient +from adcp.reporting.conformance import validate_reporting_source_execution +from adcp.reporting.fixtures import redacted_capabilities, redacted_snapshot_request +from adcp.reporting.inline_source import InlineReportingSource +from adcp.reporting.ledger.models import ReportingDefinitionBinding +from adcp.reporting.materializer.contracts import ( + ReportingCanonicalization, + ReportingVerificationKey, + ReportingWriterCapability, + ReportingWriterError, +) +from adcp.reporting.materializer.reference import reference_verifier +from adcp.reporting.materializer.verification import ReportingRevisionVerifier +from adcp.server.idempotency import MemoryBackend, WebhookDedupStore +from adcp.server.test_controller import TestControllerStore, _handle_test_controller +from adcp.signing import StaticJwksResolver, private_key_from_jwk +from adcp.types import ( + AgentConfig, + GetMediaBuyDeliveryResponse, + MediaBuyDeliveryWebhookResult, + Protocol, + ReportingReportDefinition, +) +from adcp.types.core import TaskStatus +from adcp.types.domains.enums.reporting_custody import ReportingCustody +from adcp.webhooks import ( + WebhookReceiver, + WebhookReceiverConfig, + WebhookVerifyOptions, + create_mcp_webhook_payload, + sign_webhook, + to_wire_dict, +) + +# --------------------------------------------------------------------------- +# Delivery rows without impressions +# --------------------------------------------------------------------------- + +OBSERVED_AT = datetime(2026, 11, 6, 12, 0, tzinfo=timezone.utc) + + +def _delivery_response(request: Any, packages: list[dict[str, Any]]) -> Any: + return GetMediaBuyDeliveryResponse.model_validate( + { + "reporting_period": { + "start": request.period.start.isoformat(), + "end": request.period.source_read_cutoff_at.isoformat(), + }, + "media_buy_deliveries": [ + { + "media_buy_id": "media-buy-redacted", + "status": "active", + "currency": "USD", + "totals": {"spend": 1.25}, + "by_package": packages, + } + ], + } + ) + + +def _package(package_id: str, spend: float, **metrics: Any) -> dict[str, Any]: + return { + "package_id": package_id, + "pricing_model": "cpm", + "rate": 1.0, + "currency": "USD", + "spend": spend, + **metrics, + } + + +async def _staged_rows(packages: list[dict[str, Any]]) -> tuple[Any, list[dict[str, Any]]]: + request = redacted_snapshot_request() + source = InlineReportingSource( + fetch=lambda _request: _delivery_response(request, packages), + capabilities=redacted_capabilities(), + clock=lambda: OBSERVED_AT, + ) + manifest = await validate_reporting_source_execution( + capabilities=source.capabilities, + request=request, + result=await source.execute(request, cancel=asyncio.Event()), + object_reader=source.staging, + ) + staged = manifest.objects[0] + payload = await source.staging.read( + object_ref=staged.object_ref, + object_generation=staged.object_generation, + account_id=request.identity.account_id, + source_scope=request.identity.source_scope, + cancel=asyncio.Event(), + ) + return manifest, [json.loads(line) for line in payload.splitlines()] + + +def test_generated_delivery_rows_make_impressions_optional() -> None: + response = _delivery_response(redacted_snapshot_request(), [_package("pkg-1", 0.75)]) + row = response.media_buy_deliveries[0].by_package[0] + assert row.impressions is None + assert "impressions" not in row.model_dump(mode="json", exclude_none=True) + + +async def test_a_row_without_impressions_is_staged_without_a_zero() -> None: + manifest, rows = await _staged_rows( + [_package("pkg-1", 0.75, impressions=6), _package("pkg-2", 0.5)] + ) + by_package = {row["package_id"]: row for row in rows} + assert str(by_package["pkg-1"]["impressions"]) == "6" + assert "impressions" not in by_package["pkg-2"] + totals = {total.name: total.value for total in manifest.control_totals} + # A total over rows that do not all carry the metric would count the + # unmeasured row as zero, so none is published for it. + assert "impressions" not in totals + assert totals["spend"] == "1.25" + + +async def test_rows_that_all_carry_impressions_still_total_them() -> None: + manifest, _ = await _staged_rows( + [_package("pkg-1", 0.75, impressions=6), _package("pkg-2", 0.5, impressions=0)] + ) + totals = {total.name: total.value for total in manifest.control_totals} + assert totals["impressions"] == "6" + + +# --------------------------------------------------------------------------- +# Delivery webhooks: aggregated_totals is ignored, never a rejection +# --------------------------------------------------------------------------- + +VECTORS_DIR = Path(__file__).parent / "conformance" / "vectors" / "request-signing" +KEYS = json.loads((VECTORS_DIR / "keys.json").read_text())["keys"] +WEBHOOK_KEY = { + **copy.deepcopy(next(k for k in KEYS if k["kid"] == "test-ed25519-2026")), + "kid": "test-webhook-ed25519-2026", + "adcp_use": "webhook-signing", +} +URL = "https://buyer.example.com/webhooks/adcp" +JSON_HEADERS = {"Content-Type": "application/json"} + +DELIVERY_RESULT: dict[str, Any] = { + "notification_type": "scheduled", + "sequence_number": 3, + "reporting_period": {"start": "2026-09-01T00:00:00Z", "end": "2026-09-02T00:00:00Z"}, + # Senders must not include it; a 3.3 receiver ignores it if one does. + "aggregated_totals": {"impressions": 999, "spend": 999.0, "media_buy_count": 1}, + "media_buy_deliveries": [ + { + "media_buy_id": "mb-1", + "status": "active", + "currency": "USD", + "totals": {"spend": 12.5}, + "by_package": [{"package_id": "pkg-1", "currency": "USD", "spend": 12.5}], + } + ], +} + + +def _delivery_webhook(key: str) -> dict[str, Any]: + payload = create_mcp_webhook_payload( + task_id="delivery_report_mb-1_000003", + task_type="media_buy_delivery", + operation_id="op-reporting", + status="completed", # type: ignore[arg-type] + result=copy.deepcopy(DELIVERY_RESULT), + idempotency_key=key, + ) + return to_wire_dict(payload) + + +async def test_webhook_receiver_accepts_aggregated_totals_and_rows_without_impressions() -> None: + receiver = WebhookReceiver( + config=WebhookReceiverConfig( + verify_options=WebhookVerifyOptions( + jwks_resolver=StaticJwksResolver({"keys": [WEBHOOK_KEY]}), + ), + dedup=WebhookDedupStore(MemoryBackend(), ttl_seconds=86400), + receiver_scope="test-receiver", + publisher_scope_for=lambda _signer: "test-publisher", + ), + ) + body = json.dumps( + _delivery_webhook("whk_reporting_33_rules_0001"), separators=(",", ":") + ).encode() + signed = sign_webhook( + method="POST", + url=URL, + headers=JSON_HEADERS, + body=body, + private_key=private_key_from_jwk(WEBHOOK_KEY, d_field="_private_d_for_test_only"), + key_id=WEBHOOK_KEY["kid"], + alg="ed25519", + ) + outcome = await receiver.receive( + method="POST", url=URL, headers={**JSON_HEADERS, **signed.as_dict()}, body=body + ) + assert outcome.rejected is False, outcome.rejection_reason + result = outcome.payload.result + result = getattr(result, "root", result) + assert isinstance(result, MediaBuyDeliveryWebhookResult) + assert result.media_buy_deliveries[0].by_package[0].impressions is None + # Nothing in the SDK reads it; it stays an unmodelled extra. + assert "aggregated_totals" not in type(result).model_fields + + +async def test_client_handle_webhook_accepts_a_delivery_report_with_aggregated_totals() -> None: + client = ADCPClient( + AgentConfig(id="seller", agent_uri="https://seller.example.com", protocol=Protocol.MCP), + allow_unauthenticated_webhooks=True, + ) + result = await client.handle_webhook( + _delivery_webhook("whk_reporting_33_rules_0002"), + task_type="media_buy_delivery", + operation_id="op-reporting", + ) + assert result.success is True + assert result.status == TaskStatus.COMPLETED + + +# --------------------------------------------------------------------------- +# Reliable Reporting source.custody +# --------------------------------------------------------------------------- + +_ASSETS = files("adcp.reporting.materializer").joinpath("assets") + + +def _verifier(definition_bytes: bytes) -> ReportingRevisionVerifier: + schema = _ASSETS.joinpath("reference-row-schema.json").read_bytes() + contract = _ASSETS.joinpath("reference-canonicalization.json").read_bytes() + return ReportingRevisionVerifier( + ReportingVerificationKey( + "reference-report-v1", + "paid_media_delivery", + ReportingDefinitionBinding( + "https://contracts.example.test/reference-definition-custody.json", + hashlib.sha256(definition_bytes).hexdigest(), + "1.0.0", + "https://contracts.example.test/reference-row-schema.json", + hashlib.sha256(schema).hexdigest(), + monetary_metric_units=(("spend", "USD"),), + monetary_control_total_units=(("spend", "USD"),), + ), + ReportingCanonicalization( + "reference-jcs-rows-v1", + "https://contracts.example.test/reference-canonicalization.json", + hashlib.sha256(contract).hexdigest(), + ), + ReportingWriterCapability( + "file_transfer", + "reference-memory", + "jsonl", + "canonical_digest", + "producer", + "immutable_location", + "sha256", + "conditional_create", + ), + ), + definition_bytes, + schema, + contract, + ) + + +def _definition(**source: Any) -> dict[str, Any]: + definition = json.loads(_ASSETS.joinpath("reference-definition.json").read_bytes()) + definition["source"].update(source) + return definition + + +def _bytes(definition: dict[str, Any]) -> bytes: + return json.dumps(definition, separators=(",", ":")).encode() + + +def test_a_definition_without_custody_makes_no_claim() -> None: + assert reference_verifier().custody is None + + +@pytest.mark.parametrize("custody", list(ReportingCustody)) +def test_the_verifier_accepts_and_surfaces_custody(custody: ReportingCustody) -> None: + verifier = _verifier(_bytes(_definition(custody=custody.value))) + assert verifier.custody is custody + assert json.loads(verifier.definition_bytes)["source"]["custody"] == custody.value + + +def test_custody_is_covered_by_the_definition_digest() -> None: + claimed = _bytes(_definition(custody="system_of_record")) + unclaimed = _bytes(_definition()) + assert hashlib.sha256(claimed).hexdigest() != hashlib.sha256(unclaimed).hexdigest() + pinned = _verifier(unclaimed).key + # Adding the claim to an existing definition needs a new digest: the old + # pin does not verify the edited bytes. + with pytest.raises(ReportingWriterError) as raised: + ReportingRevisionVerifier( + pinned, + claimed, + _ASSETS.joinpath("reference-row-schema.json").read_bytes(), + _ASSETS.joinpath("reference-canonicalization.json").read_bytes(), + ) + assert raised.value.failure.code == "UNSUPPORTED_VERIFICATION" + + +@pytest.mark.parametrize("custody", ["mixed", "", 1, None, ["system_of_record"]]) +def test_a_malformed_custody_claim_fails_closed(custody: Any) -> None: + with pytest.raises(ReportingWriterError) as raised: + _verifier(_bytes(_definition(custody=custody))) + assert raised.value.failure.code == "UNSUPPORTED_VERIFICATION" + + +def test_the_generated_definition_model_round_trips_custody() -> None: + definition = ReportingReportDefinition.model_validate(_definition(custody="upstream_derived")) + assert definition.source.custody == ReportingCustody.upstream_derived + assert definition.model_dump(mode="json", exclude_none=True)["source"]["custody"] == ( + "upstream_derived" + ) + assert ReportingReportDefinition.model_validate(_definition()).source.custody is None + + +# --------------------------------------------------------------------------- +# simulate_delivery property-grain rows +# --------------------------------------------------------------------------- + +PROPERTY_ROWS = [ + { + "publisher_domain": "nova-sports.example", + "identifier": {"type": "domain", "value": "nova-sports.example"}, + "impressions": 2000, + "spend": 10, + } +] +INSTALLMENT_PROPERTY_ROWS = [ + { + "installment_ref": { + "collection_ref": { + "publisher_domain": "studio-network.example", + "collection_id": "nightly_news", + }, + "installment_id": "ep_102", + }, + "publisher_domain": "streamhaus.example", + "identifier": {"type": "domain", "value": "streamhaus.example"}, + "impressions": 1000, + "spend": 5, + } +] + + +@pytest.fixture +def _sandbox(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ADCP_SANDBOX", "1") + + +class _PropertyStore(TestControllerStore): + def __init__(self) -> None: + self.received: dict[str, Any] = {} + + async def simulate_delivery( + self, + media_buy_id: str, + impressions: int | None = None, + property_delivery: list[dict[str, Any]] | None = None, + installment_property_delivery: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + self.received = { + "media_buy_id": media_buy_id, + "impressions": impressions, + "property_delivery": property_delivery, + "installment_property_delivery": installment_property_delivery, + } + return {"simulated": {"media_buy_id": media_buy_id}} + + +class _LegacyStore(TestControllerStore): + async def simulate_delivery( + self, media_buy_id: str, impressions: int | None = None + ) -> dict[str, Any]: + raise AssertionError("rows the store cannot accept must not be dropped silently") + + +def _simulate(**params: Any) -> dict[str, Any]: + return { + "scenario": "simulate_delivery", + "account": {"sandbox": True}, + "params": {"media_buy_id": "mb-1", **params}, + } + + +@pytest.mark.usefixtures("_sandbox") +async def test_injected_property_rows_reach_the_store_verbatim() -> None: + store = _PropertyStore() + result = await _handle_test_controller( + store, + _simulate( + impressions=10000, + property_delivery=PROPERTY_ROWS, + installment_property_delivery=INSTALLMENT_PROPERTY_ROWS, + ), + validate_schema=True, + ) + assert result["success"] is True, result + assert store.received["property_delivery"] == PROPERTY_ROWS + assert store.received["installment_property_delivery"] == INSTALLMENT_PROPERTY_ROWS + + +@pytest.mark.usefixtures("_sandbox") +async def test_impressions_are_optional_and_never_defaulted_to_zero() -> None: + store = _PropertyStore() + result = await _handle_test_controller(store, _simulate(), validate_schema=True) + assert result["success"] is True, result + assert store.received["impressions"] is None + + +@pytest.mark.usefixtures("_sandbox") +@pytest.mark.parametrize( + ("name", "rows"), + [ + ("property_delivery", PROPERTY_ROWS), + ("installment_property_delivery", INSTALLMENT_PROPERTY_ROWS), + ], +) +async def test_injected_rows_are_undated(name: str, rows: list[dict[str, Any]]) -> None: + store = _PropertyStore() + result = await _handle_test_controller( + store, _simulate(delivery_date="2026-09-01", **{name: rows}), validate_schema=True + ) + assert result["success"] is False + assert result["error"] == "INVALID_PARAMS" + assert name in result["error_detail"] + assert store.received == {} + + +@pytest.mark.usefixtures("_sandbox") +@pytest.mark.parametrize( + ("name", "rows"), + [ + ("property_delivery", PROPERTY_ROWS), + ("installment_property_delivery", INSTALLMENT_PROPERTY_ROWS), + ], +) +async def test_a_store_that_cannot_take_injected_rows_refuses_them( + name: str, rows: list[dict[str, Any]] +) -> None: + result = await _handle_test_controller( + _LegacyStore(), _simulate(**{name: rows}), validate_schema=True + ) + assert result["success"] is False + assert result["error"] == "INVALID_PARAMS" + assert name in result["error_detail"] + + +@pytest.mark.usefixtures("_sandbox") +async def test_a_kwargs_store_receives_every_injected_array() -> None: + received: dict[str, Any] = {} + + class _KwargsStore(TestControllerStore): + async def simulate_delivery(self, media_buy_id: str, **params: Any) -> dict[str, Any]: + received.update(params) + return {"simulated": {"media_buy_id": media_buy_id}} + + result = await _handle_test_controller( + _KwargsStore(), _simulate(property_delivery=PROPERTY_ROWS), validate_schema=True + ) + assert result["success"] is True, result + assert received["property_delivery"] == PROPERTY_ROWS