From 184215360b7b1c1321289c79b4da22555cb9866d Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 17:53:10 +0000 Subject: [PATCH 1/3] feat(reporting): add the reporting test kit, reference seller and pytest fixture adcp.testing.reporting.reporting_test_kit runs a seller in process under a virtual clock with real buyer clients bound to it. kit.clock advances and runs the seller worker at every period boundary crossed; kit.grants revokes, regrants, reduces and winds down; kit.accounts sets a timezone; kit.buys adds and cancels buys. All of them go through the controller's reporting_subscription_probe and seed_media_buy, so a host can test its own platform by passing it with its service and controller. assert_agreement compares the seller's owed set with the buyer's reconcile and the served authorized_at with the one coverage used, and reports each disputed period with the first differing trace rule. The default seller is the SDK reference seller on the managed path, with an in-memory world of accounts, grants and buys and a controller serving seed_media_buy and reporting_subscription_probe. The adcp_reporting_kit fixture lives in the adcp.testing.reporting_pytest plugin. The kit found a seller bug, fixed here: served gaps after a timezone change were computed on the new zone's calendar, so they removed periods the owed set still owed. Gaps now use the zone in effect at the loss. Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 2 +- docs/reporting-subscriptions-testing.md | 52 +- src/adcp/reporting/subscriptions/engine.py | 15 +- src/adcp/testing/decisioning.py | 3 +- src/adcp/testing/reporting.py | 988 +++++++++++++++++++++ src/adcp/testing/reporting_pytest.py | 52 ++ tests/test_reporting_test_kit.py | 135 +++ 7 files changed, 1240 insertions(+), 7 deletions(-) create mode 100644 src/adcp/testing/reporting.py create mode 100644 src/adcp/testing/reporting_pytest.py create mode 100644 tests/test_reporting_test_kit.py diff --git a/README.md b/README.md index 742527fe9..c79ec9672 100644 --- a/README.md +++ b/README.md @@ -375,7 +375,7 @@ outcome = await client.accounts.ensure(key, billing="operator") # AccountReady - **Explain.** Seller `reporting.explain_coverage(...)` and buyer `client.reporting.explain(...)` return the same `CoverageTrace`; `python -m adcp reporting diff` compares them. - **Events.** `client.reporting.poll()` / `events()` derive events from saved state with deterministic ids; `handle_webhook()` treats a verified webhook as a doorbell. - **Receipts.** Off by default; `reconcile(receipts=ReceiptPolicy(mode="verified", commit=...))` receipts a billing revision only after the SDK verified it and your commit returned a reference. -- **Coming in later betas:** the test kit. +- **Test kit.** `adcp.testing.reporting.reporting_test_kit(...)` (and the `adcp_reporting_kit` pytest fixture) runs the reference seller under a virtual clock with real buyers and `assert_agreement`. Guides: [seller](docs/reporting-subscriptions-seller.md) · [buyer](docs/reporting-subscriptions-buyer.md) · [coverage](docs/reporting-subscriptions-coverage.md) · [accounts](docs/accounts-ensure-and-change.md) · [migrating from account-level configs](docs/reporting-subscriptions-migration.md) · [testing](docs/reporting-subscriptions-testing.md) · [errors](docs/reporting-subscriptions-errors.md) diff --git a/docs/reporting-subscriptions-testing.md b/docs/reporting-subscriptions-testing.md index 4a3d87a13..767dd1c55 100644 --- a/docs/reporting-subscriptions-testing.md +++ b/docs/reporting-subscriptions-testing.md @@ -2,11 +2,59 @@ ## Testing today -`ReliableReportingService(storage=memory_storage(), subscriptions=...)` runs in process under pytest. Drive it with a `DeterministicReportingClock` and `run_worker()`, and call `await reporting.drain_subscription_triggers()` before asserting on triggered reconciles. `tests/test_reporting_subscription_managed.py` is a worked example. +`reporting_test_kit(...)` runs a seller in process under a virtual clock and gives you real buyer clients bound to it. The SDK's reference seller is the default. Pass your own platform to test your wiring. + +```python +from adcp.testing.reporting import offerings, reporting_test_kit + +async with reporting_test_kit( + start="2026-11-01T00:00:00-04:00", + offerings=[offerings.daily_analytics(alignment="account_timezone")], + accounts=[{"account_id": "acc_a", "timezone": "America/New_York"}], + buys=[{"media_buy_id": "mb_1", "account_id": "acc_a", "confirmed_at": "2026-11-03T15:00:00Z", + "start_time": "2026-11-03T05:00:00Z", "end_time": "2026-11-30T05:00:00Z"}], +) as kit: + buyer = kit.buyer("agency-x") # a real client.reporting + await buyer.reporting.subscribe(analytics_daily) + await kit.clock.advance("P3D") # the worker runs at each boundary crossed + await kit.grants.revoke("agency-x", "acc_a") + await kit.clock.advance("P1D") + await kit.grants.regrant("agency-x", "acc_a") + await kit.grants.wind_down("agency-x", "acc_a") + await kit.accounts.set_timezone("acc_a", "America/Los_Angeles", principal="agency-x") + await kit.clock.advance_to("2027-03-14T12:00:00-04:00") # DST + await kit.buys.cancel("mb_1", principal="agency-x") + await kit.assert_agreement("agency-x") +``` + +- **Clock.** `kit.clock.advance("P3D")` (days to seconds) and `advance_to(instant)` move the one virtual clock forward. At every period boundary crossed, the seller processes the move as production would: the boundary sweep, then planning. +- **Grants, accounts and buys.** `kit.grants` (`revoke`, `regrant`, `reduce`, `wind_down`), `kit.accounts.set_timezone` and `kit.buys` (`add`, `cancel`) go through the controller's `reporting_subscription_probe` and `seed_media_buy`. That is the same scenario a storyboard drives against a remote seller. +- **Agreement.** `await kit.assert_agreement(principal)` checks that the seller's owed set equals what the buyer's `reconcile()` computes over the wire. It also checks that the `authorized_at` served on `list_accounts` is the one coverage used. On a difference, the `AgreementError` names each disputed period, the first rule at which the two `CoverageTrace`s differ, and both sentences. +- **Your platform.** Pass `platform=` (your installed handler), `service=` (your `ReliableReportingService`) and `controller=` (a `TestControllerStore` serving `reporting_subscription_probe` and `seed_media_buy`). The kit then drives your seller through that controller. +- **Offerings.** `offerings.daily_analytics(...)` and `offerings.hourly_pacing(...)` build offering fixtures that publish `standard_dimensions`. ## The test kit -`reporting_test_kit(...)` and the `adcp_reporting_kit` pytest fixture arrive in a later beta. They add a virtual clock, grants, accounts, buys, and `assert_agreement`. Python sellers run the TypeScript storyboard CLI against their URL: +The pytest fixture `adcp_reporting_kit` yields a started kit and closes it after the test: + +```python +# conftest.py +pytest_plugins = ["adcp.testing.reporting_pytest"] + +# test_reporting.py +@pytest.mark.adcp_reporting_kit(accounts=[{"account_id": "acc_a", "timezone": "UTC"}]) +async def test_owed(adcp_reporting_kit): + buyer = adcp_reporting_kit.buyer("agency-x") + await buyer.reporting.subscribe(analytics_daily) + await adcp_reporting_kit.clock.advance("P3D") + await adcp_reporting_kit.assert_agreement("agency-x") +``` + +The marker's keyword arguments are those of `reporting_test_kit`. Run the fixture under pytest-asyncio. + +For a lower-level test, `ReliableReportingService(storage=memory_storage(), subscriptions=...)` runs in process under pytest. Drive it with a `DeterministicReportingClock` and `run_worker()`, and call `await reporting.drain_subscription_triggers()` before asserting on triggered reconciles. `tests/test_reporting_subscription_managed.py` is a worked example. + +Python sellers run the TypeScript storyboard CLI against their URL: ``` npx adcp storyboard run https://seller.example/mcp --storyboards reporting_subscriptions --auth "$TOKEN" diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index ff564a586..6ed8c9755 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -65,6 +65,7 @@ SubscriptionBuy, SubscriptionException, SubscriptionVersionWindow, + _timezone_at, authorization_chain_start, authorization_regranted_at, owed_subscription_periods, @@ -1892,7 +1893,11 @@ def _schedules( destinations: Mapping[str, Any], now: datetime, ) -> list[tuple[str, PeriodCalendar]]: - """``(offering_id, calendar)`` for each offering the account resolves to.""" + """``(offering_id, calendar)`` for each offering the account resolves to at ``now``. + + Each calendar is in the zone in effect at ``now``, so a gap served for + a loss before a timezone change stays on the calendar it removed. + """ if account is None: return [] selector = SubscriptionSelector.from_subscription(state["configuration"]) @@ -1912,7 +1917,13 @@ def _schedules( ).offerings: found[item.offering_id] = item return [ - (item.offering_id, item.calendar(account.timezone, selector.period_timezone)) + ( + item.offering_id, + item.calendar( + _timezone_at(now, account.timezone, account.timezone_changes, item.offering_id), + selector.period_timezone, + ), + ) for item in sorted(found.values(), key=lambda value: value.offering_id.encode()) ] diff --git a/src/adcp/testing/decisioning.py b/src/adcp/testing/decisioning.py index 0d632a46b..7ff22c489 100644 --- a/src/adcp/testing/decisioning.py +++ b/src/adcp/testing/decisioning.py @@ -289,8 +289,7 @@ def build_asgi_app( raise ValueError(f"Unsupported transport {transport!r}; expected 'mcp', 'a2a', or 'both'.") if (on_startup or on_shutdown) and transport != "both": raise ValueError( - "on_startup / on_shutdown hooks require transport='both', " - f"got transport={transport!r}." + f"on_startup / on_shutdown hooks require transport='both', got transport={transport!r}." ) if transport == "a2a": ignored_session_settings = [] diff --git a/src/adcp/testing/reporting.py b/src/adcp/testing/reporting.py new file mode 100644 index 000000000..8e2673074 --- /dev/null +++ b/src/adcp/testing/reporting.py @@ -0,0 +1,988 @@ +"""The reporting-subscription test kit and the reference seller (AdCP 3.3, experimental). + +``reporting_test_kit(...)`` runs a seller in process under a virtual clock and +hands you real buyer clients bound to it:: + + from adcp.testing.reporting import offerings, reporting_test_kit + + async with reporting_test_kit( + start="2026-11-01T00:00:00-04:00", + offerings=[offerings.daily_analytics(alignment="account_timezone")], + accounts=[{"account_id": "acc_a", "timezone": "America/New_York"}], + buys=[{"media_buy_id": "mb_1", "account_id": "acc_a", + "confirmed_at": "2026-11-03T15:00:00Z", + "start_time": "2026-11-03T05:00:00Z", "end_time": "2026-11-30T05:00:00Z"}], + ) as kit: + buyer = kit.buyer("agency-x") + await buyer.reporting.subscribe(analytics_daily) + await kit.clock.advance("P3D") # runs the worker at each boundary crossed + await kit.grants.revoke("agency-x", "acc_a") + await kit.assert_agreement("agency-x") # seller owed set == buyer computed set + +Without ``platform=``, the kit runs :class:`ReferenceReportingSeller`, the +SDK's reference seller on the managed path. Pass your own installed +``platform``, its ``service`` and a ``controller`` (a +:class:`~adcp.server.test_controller.TestControllerStore` serving +``reporting_subscription_probe`` and ``seed_media_buy``) to test your wiring: +the kit drives grants, timezones, buys and the clock only through that +controller, the same scenario a remote storyboard uses. + +:func:`create_reference_reporting_seller` builds the reference seller alone, +for a storyboard or a standalone server (``python -m adcp reference-seller +reporting``). +""" + +from __future__ import annotations + +import asyncio +import contextlib +import copy +from collections.abc import AsyncIterator, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from typing import Any + +from adcp.reporting.subscriptions._clock import iso_z, parse_delivery_sla, utc_instant +from adcp.server import ADCPHandler +from adcp.server.base import AccountAwareToolContext, ToolContext +from adcp.server.test_controller import TestControllerError, TestControllerStore + +#: The principal the reference seller's standalone server assumes without a token. +DEFAULT_PRINCIPAL = "storyboard-buyer" +#: The sandbox account every principal is granted on the reference seller. +REFERENCE_ACCOUNT = "acct-reference" +#: Every task a fresh grant allows on the reference seller. +GRANTED_TASKS = ( + "get_media_buys", + "get_reporting_status", + "sync_reporting_receipts", + "sync_reporting_status", + "list_accounts", + "sync_principal", + "get_principal", +) +#: Boundaries one ``advance`` may process before it refuses. +MAX_STEPS = 20_000 + + +def _instant(value: Any) -> datetime: + if isinstance(value, datetime): + return utc_instant(value) + from adcp.reporting._timestamp import aware_timestamp + + return utc_instant(aware_timestamp(str(value))) + + +# -- offerings ------------------------------------------------------------------------- + + +#: The report contract the reference source declares (``adcp.reporting.fixtures``). +_DEFINITION_ID = "PAID_MEDIA_DAILY_V1" +_DEFINITION_URI = "https://contracts.example.test/reporting/paid-media-daily-v1" + + +def _offering( + offering_id: str, + *, + feed_purpose: str, + period_duration: str, + alignment: str, + delivery_sla: str, + dimensions: Sequence[str], + finality: Sequence[str], +) -> dict[str, Any]: + return { + "offering_id": offering_id, + "feed_purpose": feed_purpose, + "report_definition_id": _DEFINITION_ID, + "report_definition_uri": _DEFINITION_URI, + "report_definition_sha256": "a" * 64, + "reporting_profile": { + "id": "paid_media_delivery", + "version": "1.0.0", + "schema_uri": f"{_DEFINITION_URI}/schema", + "schema_sha256": "b" * 64, + "schema_dialect": "https://json-schema.org/draft/2020-12/schema", + "schema_ref_policy": "local_fragment_only", + "grain": "one row per media buy per period", + "primary_keys": ["media_buy_id"], + }, + "schedule": { + "period_duration": period_duration, + "delivery_sla": delivery_sla, + "alignment": alignment, + }, + "supported_finality": list(finality), + "reconciliation_mode": "delivery_only", + "standard_dimensions": list(dimensions), + } + + +class _Offerings: + """Offering fixtures for the kit and the reference seller.""" + + @staticmethod + def daily_analytics( + *, + offering_id: str = "REFERENCE_DAILY_V1", + alignment: str = "utc", + dimensions: Sequence[str] = ("package",), + delivery_sla: str = "PT6H", + finality: Sequence[str] = ("snapshot",), + ) -> dict[str, Any]: + """A ``P1D`` analytics offering.""" + return _offering( + offering_id, + feed_purpose="analytics", + period_duration="P1D", + alignment=alignment, + delivery_sla=delivery_sla, + dimensions=dimensions, + finality=finality, + ) + + @staticmethod + def hourly_pacing( + *, + offering_id: str = "REFERENCE_HOURLY_V1", + dimensions: Sequence[str] = ("package",), + delivery_sla: str = "PT10M", + finality: Sequence[str] = ("snapshot",), + ) -> dict[str, Any]: + """A ``PT1H`` UTC pacing offering.""" + return _offering( + offering_id, + feed_purpose="pacing", + period_duration="PT1H", + alignment="utc", + delivery_sla=delivery_sla, + dimensions=dimensions, + finality=finality, + ) + + +offerings = _Offerings() + + +# -- the world ------------------------------------------------------------------------- + + +@dataclass +class _Grant: + active: bool = True + allowed_tasks: tuple[str, ...] = GRANTED_TASKS + + +@dataclass +class ReportingWorld: + """The reference seller's state: accounts, grants and buys, all in memory.""" + + accounts: dict[str, dict[str, Any]] = field(default_factory=dict) + buys: dict[str, dict[str, Any]] = field(default_factory=dict) + grants: dict[tuple[str, str], _Grant] = field(default_factory=dict) + #: Accounts no principal may read: the others are granted on first sight. + restricted: set[str] = field(default_factory=set) + + def grant(self, principal: str, account_id: str) -> _Grant: + key = (principal, account_id) + if key not in self.grants: + self.grants[key] = _Grant(active=account_id not in self.restricted) + return self.grants[key] + + def readable(self, principal: str, account_id: str) -> bool: + return account_id in self.accounts and self.grant(principal, account_id).active + + +def _principal(context: Any) -> str: + identity = getattr(context, "caller_identity", None) + return str(identity) if identity else DEFAULT_PRINCIPAL + + +class ReferenceReportingSeller(ADCPHandler[ToolContext]): + """The tasks a reporting-subscriptions seller already serves, over :class:`ReportingWorld`. + + ``list_accounts`` lists the accounts the caller holds a grant on, and + ``get_media_buys`` the buys on an account it can read. The Reliable + Reporting service installed on it serves everything else. + """ + + def __init__(self, world: ReportingWorld) -> None: + self.world = world + + async def get_adcp_capabilities(self, params: Any, context: Any = None) -> Any: + return {"adcp": {"major_versions": [3]}, "supported_protocols": ["media_buy"]} + + async def list_accounts(self, params: Any, context: Any = None) -> Any: + principal = _principal(context) + accounts = [] + for account_id, account in sorted(self.world.accounts.items()): + if not self.world.readable(principal, account_id): + continue + grant = self.world.grant(principal, account_id) + accounts.append( + { + **copy.deepcopy(account), + "authorization": {"allowed_tasks": list(grant.allowed_tasks)}, + } + ) + return {"accounts": accounts} + + async def get_media_buys(self, params: Any, context: Any = None) -> Any: + principal = _principal(context) + payload = ( + params.model_dump(mode="json", exclude_none=True) + if hasattr(params, "model_dump") + else dict(params or {}) + ) + account_id = (payload.get("account") or {}).get("account_id") + wanted = set(payload.get("media_buy_ids") or ()) + buys = [ + copy.deepcopy(buy) + for buy in self.world.buys.values() + if self.world.readable(principal, buy["account_id"]) + and (account_id is None or buy["account_id"] == account_id) + and (not wanted or buy["media_buy_id"] in wanted) + ] + buys.sort(key=lambda buy: buy["media_buy_id"]) + return {"media_buys": buys} + + +class _Source: + """Answers every slice with one row per requested media buy.""" + + def __init__(self, capabilities: Any) -> None: + self._capabilities = capabilities + + @property + def capabilities(self) -> Any: + return self._capabilities + + def fetch_slice(self, request: Any) -> list[dict[str, Any]]: + return [ + { + "media_buy_id": item.media_buy_id, + "campaign_id": "campaign-reference", + "impressions": 100, + "spend": "1.00", + } + for item in request.coverage.constituents + ] + + +@dataclass +class ReferenceSeller: + """The reference seller: its world, service, installed platform, clock and controller.""" + + world: ReportingWorld + service: Any + platform: Any + clock: Any + controller: ReportingTestController + offerings: tuple[dict[str, Any], ...] + + def context(self, principal: str = DEFAULT_PRINCIPAL, account_id: str | None = None) -> Any: + return AccountAwareToolContext(caller_identity=principal, account_id=account_id) + + def mcp_server( + self, *, principal: str = DEFAULT_PRINCIPAL, name: str = "adcp-reference-seller" + ) -> Any: + """An in-process MCP server, with the caller fixed to ``principal``.""" + from adcp.server.serve import create_mcp_server + from adcp.server.test_controller import register_test_controller + + def context_factory(_meta: Any) -> ToolContext: + return ToolContext(caller_identity=principal) + + server = create_mcp_server(self.platform, name=name, context_factory=context_factory) + register_test_controller(server, self.controller, context_factory=context_factory) + return server + + +def create_reference_reporting_seller( + *, + start: Any = "2026-11-01T00:00:00Z", + offerings: Sequence[Mapping[str, Any]] | None = None, + accounts: Sequence[Mapping[str, Any]] | None = None, + buys: Sequence[Mapping[str, Any]] = (), + max_accounts: int = 100, + backfill_on_regrant: bool = False, + storage: Any = None, +) -> ReferenceSeller: + """The SDK's reference seller for reporting subscriptions, on the managed path. + + Every principal is granted on every account (``acct-reference`` by + default) the first time it is seen, unless the account is ``restricted``. + ``storage`` defaults to process-local memory storage. + """ + from adcp.reporting import SubscriptionOptions + from adcp.reporting.fixtures import SNAPSHOT_OFFERING_ID, redacted_capabilities + from adcp.reporting.ledger import ReportingDefinitionBinding + from adcp.reporting.service import ReliableReportingService, ReportingAccountContext + from adcp.reporting.testing import DeterministicReportingClock + + clock = DeterministicReportingClock(_instant(start)) + world = ReportingWorld() + for account in accounts or ( + { + "account_id": REFERENCE_ACCOUNT, + "name": "Reference account", + "status": "active", + "timezone": "UTC", + "operator": "agency.example", + "sandbox": True, + }, + ): + record = {"name": account["account_id"], "status": "active", "timezone": "UTC", **account} + world.accounts[str(record["account_id"])] = record + for buy in buys: + _add_buy(world, buy) + advertised = tuple(dict(item) for item in (offerings or (_Offerings.daily_analytics(),))) + definitions = { + item["offering_id"]: ReportingDefinitionBinding( + report_definition_uri=item["report_definition_uri"], + report_definition_sha256=item["report_definition_sha256"], + schema_version=item["reporting_profile"]["version"], + schema_uri=item["reporting_profile"]["schema_uri"], + schema_sha256=item["reporting_profile"]["schema_sha256"], + ) + for item in advertised + } + capabilities = redacted_capabilities() + by_id = {item["offering_id"]: item for item in advertised} + + def account_context(configuration: Any) -> Any: + offering = by_id.get(getattr(configuration, "offering_id", None) or "") or advertised[0] + account = world.accounts.get(configuration.account_id) or {} + return ReportingAccountContext( + account_id=configuration.account_id, + adapter="reference", + currency="USD", + account_timezone=account.get("timezone", "UTC"), + source_scope=capabilities.source_scope, + snapshot_offering_id=SNAPSHOT_OFFERING_ID, + requested_dimensions=("campaign_id",), + capability_offering=offering, + publication_namespace="reporting-source:reference", + ) + + def caller_resolver(request: Any, context: Any) -> Any: + """The caller is the authenticated principal; the account is the one the request names.""" + from adcp.reporting.ledger import ReportingStatusCaller + + payload = ( + request.model_dump(mode="json", exclude_none=True) + if hasattr(request, "model_dump") + else dict(request or {}) + ) + account_id = (payload.get("account") or {}).get("account_id") or getattr( + context, "account_id", None + ) + if not account_id: + raise TestControllerError("INVALID_PARAMS", "the request names no account") + return ReportingStatusCaller(account_id=str(account_id), consumer_id=_principal(context)) + + options = SubscriptionOptions( + max_accounts=max_accounts, + backfill_on_regrant=backfill_on_regrant, + definitions=definitions, + ) + if storage is None: + service = ReliableReportingService.memory( + account_context=account_context, + caller_resolver=caller_resolver, + clock=clock, + read_jitter_window=timedelta(0), + subscriptions=options, + ) + else: + service = ReliableReportingService( + storage=storage, + account_context=account_context, + caller_resolver=caller_resolver, + clock=clock, + read_jitter_window=timedelta(0), + subscriptions=options, + ) + service.sources.register( + "reference", _Source(capabilities), capability_offerings=list(advertised) + ) + platform = service.install(ReferenceReportingSeller(world)) + seller = ReferenceSeller(world, service, platform, clock, None, advertised) # type: ignore[arg-type] + seller.controller = ReportingTestController(seller) + return seller + + +def _add_buy(world: ReportingWorld, buy: Mapping[str, Any]) -> dict[str, Any]: + record: dict[str, Any] = {"status": "active", "currency": "USD", "total_budget": 1000.0, **buy} + if "account_id" not in record: + raise TestControllerError("INVALID_PARAMS", "a media buy needs account_id") + if record["account_id"] not in world.accounts: + raise TestControllerError("NOT_FOUND", f"unknown account {record['account_id']!r}") + record.setdefault( + "packages", [{"package_id": f"{record['media_buy_id']}-pkg", "product_id": "reference"}] + ) + world.buys[str(record["media_buy_id"])] = record + return record + + +# -- the controller -------------------------------------------------------------------- + + +class ReportingTestController(TestControllerStore): + """``seed_media_buy`` and ``reporting_subscription_probe`` for the reference seller. + + Every probe acts on the authenticated caller's own grant on a sandbox + account, then processes the change as production would (a reconcile and, + for ``advance_time``, a worker turn at every period boundary crossed). + """ + + def __init__(self, seller: ReferenceSeller) -> None: + self._seller = seller + self._lock = asyncio.Lock() + + async def seed_media_buy( + self, + fixture: dict[str, Any] | None = None, + media_buy_id: str | None = None, + *, + context: ToolContext | None = None, + ) -> dict[str, Any]: + payload = dict(fixture or {}) + if media_buy_id is not None: + payload["media_buy_id"] = media_buy_id + if "media_buy_id" not in payload: + raise TestControllerError("INVALID_PARAMS", "seed_media_buy needs media_buy_id") + record = _add_buy(self._seller.world, payload) + await self._touch(record["account_id"], context) + return {"media_buy_id": record["media_buy_id"]} + + async def reporting_subscription_probe( + self, + operation: str, + account_id: str | None = None, + allowed_tasks: list[str] | None = None, + timezone: str | None = None, + target_time: str | None = None, + *, + context: ToolContext | None = None, + ) -> dict[str, Any]: + seller = self._seller + principal = _principal(context) + result: dict[str, Any] = {} + async with self._lock: + if operation == "advance_time": + assert target_time is not None + await self._advance_to(_instant(target_time)) + else: + assert account_id is not None + if account_id not in seller.world.accounts: + raise TestControllerError("NOT_FOUND", f"unknown account {account_id!r}") + grant = seller.world.grant(principal, account_id) + if operation == "revoke_grant": + grant.active = False + await self._touch(account_id, context, principal=principal) + elif operation == "regrant": + grant.active = True + grant.allowed_tasks = GRANTED_TASKS + await self._touch(account_id, context, principal=principal) + result.update(await self._authorization(principal, account_id)) + elif operation == "reduce_grant": + grant.allowed_tasks = tuple(allowed_tasks or ()) + await self._touch(account_id, context, principal=principal) + elif operation == "set_account_timezone": + assert timezone is not None + account = seller.world.accounts[account_id] + result["previous_timezone"] = account.get("timezone") + account["timezone"] = timezone + await self._touch(account_id, context, principal=principal) + result["cutovers"] = await self._cutovers(principal, account_id) + else: # pragma: no cover - the dispatcher validated it + raise TestControllerError("INVALID_PARAMS", f"unknown operation {operation}") + return {"success": True, "simulated_now": iso_z(seller.clock()), **result} + + async def _touch(self, account_id: str, context: Any, *, principal: str | None = None) -> None: + service = self._seller.service + ctx = self._seller.context(principal or _principal(context), account_id) + await service.touch(account_id=account_id, context=ctx) + await service.drain_subscription_triggers() + + async def _authorization(self, principal: str, account_id: str) -> dict[str, Any]: + engine = self._seller.service._subscriptions.engine + from adcp._principal_ref import PrincipalRef + + ref = PrincipalRef(None, principal) + state = await engine._read(ref.subject) + account = state.accounts.get(account_id) + if account is None or not account.grants: + return {} + from adcp.reporting.subscriptions.coverage import ( + authorization_chain_start, + authorization_regranted_at, + ) + + days = engine.status_retention_days + start = authorization_chain_start(account.grants, status_retention_days=days) + regranted = authorization_regranted_at(account.grants, status_retention_days=days) + chain = ( + "continued" if regranted is not None else ("new" if len(account.grants) > 1 else "new") + ) + result: dict[str, Any] = {"chain": chain} + if start is not None: + result["authorized_at"] = iso_z(start) + if regranted is not None: + result["regranted_at"] = iso_z(regranted) + return result + + async def _cutovers(self, principal: str, account_id: str) -> list[dict[str, Any]]: + from adcp._principal_ref import PrincipalRef + + engine = self._seller.service._subscriptions.engine + result = [] + for entry in await engine.subscription_coverage(PrincipalRef(None, principal), account_id): + for item in entry.timezone_cutovers or (): + result.append(item.model_dump(mode="json", exclude_none=True)) + return result + + async def _advance_to(self, target: datetime) -> None: + seller = self._seller + now = seller.clock() + if target < now: + raise TestControllerError( + "INVALID_PARAMS", "the sandbox clock moves forward only; target_time is earlier" + ) + steps = 0 + for boundary in _boundaries(seller, now, target): + steps += 1 + if steps > MAX_STEPS: + raise TestControllerError( + "INVALID_PARAMS", f"advance_time would cross more than {MAX_STEPS} boundaries" + ) + seller.clock.set(boundary) + await seller.service.run_worker() + await seller.service.drain_subscription_triggers() + seller.clock.set(target) + await seller.service.run_worker() + await seller.service.drain_subscription_triggers() + + +def _boundaries(seller: ReferenceSeller, after: datetime, through: datetime) -> list[datetime]: + """Every period boundary of every offering in ``(after, through]``.""" + from adcp.reporting.subscriptions.resolution import SubscriptionOffering + + zones = {str(account.get("timezone") or "UTC") for account in seller.world.accounts.values()} + points: set[datetime] = set() + for raw in seller.offerings: + offering = SubscriptionOffering.from_offering(raw) + for zone in zones: + try: + calendar = offering.calendar(zone) + except Exception: # noqa: BLE001 - a zone the offering cannot use + continue + for start, end in calendar.periods( + starts_from=calendar.floor(after), starts_before=through + ): + for point in (start, end): + if after < point <= through: + points.add(point) + return sorted(points) + + +# -- the kit --------------------------------------------------------------------------- + + +class KitBuyer: + """A buyer bound to the kit's seller: ``.reporting`` plus the raw tasks it reads.""" + + def __init__(self, platform: Any, principal: str, service: Any = None) -> None: + from adcp.reporting.subscriptions.facade import ReportingClient + + self.principal = principal + self._platform = platform + self._service = service + self.reporting = ReportingClient(self) + + def _context(self, account_id: str | None = None) -> Any: + return AccountAwareToolContext(caller_identity=self.principal, account_id=account_id) + + async def _call(self, task: str, request: Any, model: Any) -> Any: + from adcp.types.core import TaskResult, TaskStatus + + params = request.model_dump(mode="json", exclude_none=True) + account_id = (params.get("account") or {}).get("account_id") + value = await getattr(self._platform, task)(params, self._context(account_id)) + if self._service is not None: + # A served write queues a reconcile; let it land before the next read. + await self._service.drain_subscription_triggers() + if hasattr(value, "model_dump"): + value = value.model_dump(mode="json", exclude_none=True, by_alias=True) + try: + data = model.model_validate(value) + except Exception: # noqa: BLE001 - fixture buys are minimal + data = value + return TaskResult.model_construct(status=TaskStatus.COMPLETED, data=data, success=True) + + async def get_principal(self, request: Any) -> Any: + from adcp.types import GetPrincipalResponse + + return await self._call("get_principal", request, GetPrincipalResponse) + + async def sync_principal(self, request: Any) -> Any: + from adcp.types import SyncPrincipalResponse + + return await self._call("sync_principal", request, SyncPrincipalResponse) + + async def list_accounts(self, request: Any) -> Any: + from adcp.types import ListAccountsResponse + + return await self._call("list_accounts", request, ListAccountsResponse) + + async def get_media_buys(self, request: Any) -> Any: + from adcp.types import GetMediaBuysResponse + + return await self._call("get_media_buys", request, GetMediaBuysResponse) + + async def get_reporting_status(self, request: Any) -> Any: + from adcp.types import GetReportingStatusResponse + + return await self._call("get_reporting_status", request, GetReportingStatusResponse) + + async def sync_reporting_receipts(self, request: Any) -> Any: + from adcp.types import SyncReportingReceiptsResponse + + return await self._call("sync_reporting_receipts", request, SyncReportingReceiptsResponse) + + async def fetch_capabilities(self) -> Any: + value = await self._platform.get_adcp_capabilities({}, self._context()) + return ( + value.model_dump(mode="json", exclude_none=True) + if hasattr(value, "model_dump") + else value + ) + + +class KitClock: + """The kit's virtual clock: every move runs the seller's worker at each boundary.""" + + def __init__(self, kit: ReportingTestKit) -> None: + self._kit = kit + + def now(self) -> datetime: + return utc_instant(self._kit.service._clock()) + + def __call__(self) -> datetime: + return self.now() + + async def advance(self, duration: str | timedelta) -> datetime: + """Move forward by an ISO 8601 duration of days to seconds (``"P3D"``) or a timedelta.""" + delta = parse_delivery_sla(duration) if isinstance(duration, str) else duration + return await self.advance_to(self.now() + delta) + + async def advance_to(self, instant: str | datetime) -> datetime: + target = _instant(instant) + await self._kit._probe("advance_time", target_time=iso_z(target)) + return self.now() + + +class KitGrants: + """Grant changes, through the controller's ``reporting_subscription_probe``.""" + + def __init__(self, kit: ReportingTestKit) -> None: + self._kit = kit + + async def revoke(self, principal: str, account_id: str) -> dict[str, Any]: + return await self._kit._probe("revoke_grant", principal=principal, account_id=account_id) + + async def regrant(self, principal: str, account_id: str) -> dict[str, Any]: + return await self._kit._probe("regrant", principal=principal, account_id=account_id) + + async def reduce( + self, principal: str, account_id: str, allowed_tasks: Sequence[str] + ) -> dict[str, Any]: + return await self._kit._probe( + "reduce_grant", + principal=principal, + account_id=account_id, + allowed_tasks=list(allowed_tasks), + ) + + async def wind_down(self, principal: str, account_id: str) -> dict[str, Any]: + """Reduce to the wind-down tasks: coverage stays, everything else is refused.""" + from adcp.server.reporting_grants import REPORTING_WIND_DOWN_TASKS + + return await self.reduce(principal, account_id, sorted(REPORTING_WIND_DOWN_TASKS)) + + +class KitAccounts: + """Account changes, through the controller.""" + + def __init__(self, kit: ReportingTestKit) -> None: + self._kit = kit + + async def set_timezone( + self, account_id: str, timezone_name: str, *, principal: str | None = None + ) -> dict[str, Any]: + return await self._kit._probe( + "set_account_timezone", + principal=principal or self._kit.default_principal, + account_id=account_id, + timezone=timezone_name, + ) + + +class KitBuys: + """Buys, through the controller's ``seed_media_buy``.""" + + def __init__(self, kit: ReportingTestKit) -> None: + self._kit = kit + + async def add(self, buy: Mapping[str, Any], *, principal: str | None = None) -> str: + result = await self._kit._controller_call( + "seed_media_buy", principal or self._kit.default_principal, fixture=dict(buy) + ) + return str(result["media_buy_id"]) + + async def cancel( + self, media_buy_id: str, *, at: Any = None, principal: str | None = None + ) -> None: + """Cancel a buy at ``at`` (default now): its stop time becomes that instant.""" + world = self._kit.world + if world is None: + raise NotImplementedError("cancel needs the reference seller; seed the change yourself") + buy = world.buys[media_buy_id] + when = self._kit.clock.now() if at is None else _instant(at) + buy["status"] = "canceled" + buy["cancellation"] = {"canceled_at": iso_z(when), "canceled_by": "buyer"} + await self._kit._controller_call( + "seed_media_buy", principal or self._kit.default_principal, fixture=buy + ) + + +class AgreementError(AssertionError): + """The seller's owed set and the buyer's computed set differ.""" + + +@dataclass +class ReportingTestKit: + """A seller in process, a virtual clock and buyers bound to it. See the module doc.""" + + service: Any + platform: Any + controller: TestControllerStore + world: ReportingWorld | None = None + default_principal: str = DEFAULT_PRINCIPAL + + def __post_init__(self) -> None: + self.clock = KitClock(self) + self.grants = KitGrants(self) + self.accounts = KitAccounts(self) + self.buys = KitBuys(self) + self._buyers: dict[str, KitBuyer] = {} + + def buyer(self, principal: str) -> KitBuyer: + """A real ``client.reporting`` for ``principal``, bound to the in-process seller.""" + if principal not in self._buyers: + self._buyers[principal] = KitBuyer(self.platform, principal, self.service) + return self._buyers[principal] + + async def _controller_call( + self, scenario: str, principal: str, **params: Any + ) -> dict[str, Any]: + method = getattr(self.controller, scenario) + context = AccountAwareToolContext(caller_identity=principal) + try: + return dict(await method(**params, context=context)) + except NotImplementedError as error: + raise NotImplementedError( + f"the controller does not implement {scenario}; serve it to drive the kit" + ) from error + + async def _probe( + self, operation: str, *, principal: str | None = None, **params: Any + ) -> dict[str, Any]: + return await self._controller_call( + "reporting_subscription_probe", + principal or self.default_principal, + operation=operation, + **params, + ) + + async def assert_agreement(self, principal: str) -> None: + """The seller's owed set equals the buyer's, for every subscription and account. + + The seller side is the engine's owed periods that have ended; the + buyer side is ``reconcile()`` over the wire. Also checks that the + ``authorized_at`` served on ``list_accounts`` is the one coverage + used. On a difference, the error carries both sides' traces and + where they first differ. + """ + from adcp._principal_ref import PrincipalRef + from adcp.reporting.subscriptions import diff_coverage_traces + + now = self.clock.now() + ref = PrincipalRef(None, principal) + engine = self.service._subscriptions.engine + buyer = self.buyer(principal) + report = await buyer.reporting.reconcile(now=now) + buyer_set = { + ( + line.account_id, + line.subscription_id, + line.subscription_version, + line.offering_id, + line.period["start"], + line.period["end"], + tuple(sorted(line.media_buy_ids)), + ) + for line in report.periods + if line.status != "not_owed" + } + seller_set: set[tuple[Any, ...]] = set() + accounts = await engine.account_ids(ref) + unauthorized = set() + readback = await self.service.principals.reporting_subscriptions(ref) + for state in readback: + for entry in state.accounts_unauthorized or (): + unauthorized.add(entry.account_id) + for state in readback: + for account_id in sorted(set(accounts) | unauthorized): + for period in await engine.owed_periods( + ref, state.subscription_id, account_id, now=now + ): + seller_set.add( + ( + account_id, + period.subscription_id, + period.delivery_config_version, + period.offering_id, + period.period_start, + period.period_end, + tuple(sorted(period.media_buy_ids)), + ) + ) + problems: list[str] = [] + listed = await buyer.list_accounts(_list_accounts_request()) + for account in _plain_accounts(listed): + served = (account.get("authorization") or {}).get("authorized_at") + used = await engine.authorized_at(ref, account["account_id"]) + if served is not None and used is not None and _instant(served) != used: + problems.append( + f"{account['account_id']}: list_accounts serves authorized_at {served}, " + f"coverage used {iso_z(used)}" + ) + only_seller = sorted(seller_set - buyer_set) + only_buyer = sorted(buyer_set - seller_set) + for side, rows in (("seller", only_seller), ("buyer", only_buyer)): + for row in rows[:10]: + account_id, subscription_id, _version, offering_id, start, end, buys = row + detail = f"only the {side} owes {row}" + try: + period = {"start": start, "end": end} + seller_traces = await engine.explain_coverage( + ref, + account_id, + period=period, + subscription_id=subscription_id, + media_buy_id=buys[0], + offering_id=offering_id, + now=now, + ) + buyer_traces = await buyer.reporting.explain( + account_id=account_id, + period=period, + subscription_id=subscription_id, + media_buy_id=buys[0], + offering_id=offering_id, + now=now, + ) + if seller_traces and buyer_traces: + difference = diff_coverage_traces(buyer_traces[0], seller_traces[0]) + if difference is not None: + detail += f"; {difference.describe()}" + detail += ( + f"; seller: {seller_traces[0].explanation} " + f"buyer: {buyer_traces[0].explanation}" + ) + except Exception as error: # noqa: BLE001 - the diff is best effort + detail += f"; explain failed: {error}" + problems.append(detail) + if problems: + raise AgreementError( + f"buyer and seller disagree for {principal!r} ({len(only_seller)} only the " + f"seller owes, {len(only_buyer)} only the buyer owes):\n " + "\n ".join(problems) + ) + + +def _list_accounts_request() -> Any: + from adcp.types import ListAccountsRequest + + return ListAccountsRequest.model_validate({}) + + +def _plain_accounts(result: Any) -> list[dict[str, Any]]: + data: Any = getattr(result, "data", None) + if data is not None and hasattr(data, "model_dump"): + data = data.model_dump(mode="json", exclude_none=True) + return list((data or {}).get("accounts") or []) + + +@contextlib.asynccontextmanager +async def reporting_test_kit( + *, + start: Any = "2026-11-01T00:00:00Z", + offerings: Sequence[Mapping[str, Any]] | None = None, + accounts: Sequence[Mapping[str, Any]] | None = None, + buys: Sequence[Mapping[str, Any]] = (), + platform: Any = None, + service: Any = None, + controller: TestControllerStore | None = None, + max_accounts: int = 100, + backfill_on_regrant: bool = False, + default_principal: str = DEFAULT_PRINCIPAL, +) -> AsyncIterator[ReportingTestKit]: + """A reporting test kit (see the module doc). Closes the service on exit.""" + if platform is None: + if service is not None or controller is not None: + raise ValueError("service and controller come with platform=; omit all three") + seller = create_reference_reporting_seller( + start=start, + offerings=offerings, + accounts=accounts, + buys=buys, + max_accounts=max_accounts, + backfill_on_regrant=backfill_on_regrant, + ) + kit = ReportingTestKit( + seller.service, seller.platform, seller.controller, seller.world, default_principal + ) + await seller.service.run_worker() + else: + if service is None or controller is None: + raise ValueError( + "platform= needs service= (your ReliableReportingService) and controller= " + "(a TestControllerStore serving reporting_subscription_probe)" + ) + kit = ReportingTestKit(service, platform, controller, None, default_principal) + try: + yield kit + finally: + await kit.service.drain_subscription_triggers() + close = getattr(kit.service, "aclose", None) + if close is not None: + with contextlib.suppress(Exception): + await close() + + +__all__ = [ + "DEFAULT_PRINCIPAL", + "REFERENCE_ACCOUNT", + "AgreementError", + "KitBuyer", + "ReferenceReportingSeller", + "ReferenceSeller", + "ReportingTestController", + "ReportingTestKit", + "ReportingWorld", + "create_reference_reporting_seller", + "offerings", + "reporting_test_kit", +] diff --git a/src/adcp/testing/reporting_pytest.py b/src/adcp/testing/reporting_pytest.py new file mode 100644 index 000000000..24f8781ca --- /dev/null +++ b/src/adcp/testing/reporting_pytest.py @@ -0,0 +1,52 @@ +"""pytest plugin: the ``adcp_reporting_kit`` fixture (AdCP 3.3, experimental). + +Enable it in a ``conftest.py``:: + + pytest_plugins = ["adcp.testing.reporting_pytest"] + +then take the fixture, configured by an optional marker:: + + @pytest.mark.adcp_reporting_kit(accounts=[{"account_id": "acc_a", "timezone": "UTC"}]) + async def test_owed(adcp_reporting_kit): + buyer = adcp_reporting_kit.buyer("agency-x") + ... + await adcp_reporting_kit.assert_agreement("agency-x") + +The marker's keyword arguments are those of +:func:`adcp.testing.reporting.reporting_test_kit`. The fixture is async; run +it under pytest-asyncio (``asyncio_mode = "auto"``, or the plugin's strict +mode, which this fixture supports). +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from typing import Any + +import pytest + +try: # pragma: no cover - which decorator depends on the installed plugin + import pytest_asyncio + + _fixture: Any = pytest_asyncio.fixture +except ImportError: # pragma: no cover + _fixture = pytest.fixture + + +def pytest_configure(config: Any) -> None: + config.addinivalue_line( + "markers", + "adcp_reporting_kit(**options): options for the adcp_reporting_kit fixture " + "(the keyword arguments of adcp.testing.reporting.reporting_test_kit)", + ) + + +@_fixture # type: ignore[untyped-decorator] +async def adcp_reporting_kit(request: Any) -> AsyncIterator[Any]: + """A started :class:`~adcp.testing.reporting.ReportingTestKit`, closed after the test.""" + from adcp.testing.reporting import reporting_test_kit + + marker = request.node.get_closest_marker("adcp_reporting_kit") + options = dict(marker.kwargs) if marker is not None else {} + async with reporting_test_kit(**options) as kit: + yield kit diff --git a/tests/test_reporting_test_kit.py b/tests/test_reporting_test_kit.py new file mode 100644 index 000000000..994f408ed --- /dev/null +++ b/tests/test_reporting_test_kit.py @@ -0,0 +1,135 @@ +"""The reporting test kit: a seller in process, a virtual clock, buyers, and agreement.""" + +from __future__ import annotations + +import pytest + +from adcp.testing.reporting import AgreementError, offerings, reporting_test_kit + +pytest_plugins = ["adcp.testing.reporting_pytest"] +pytestmark = pytest.mark.asyncio + +SUBSCRIPTION = { + "subscription_id": "analytics-daily", + "scope": {"all_authorized_accounts": True}, + "selector": { + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "snapshot", + }, + "reconciliation_mode": "delivery_only", +} +ACCOUNTS = [{"account_id": "acc_a", "timezone": "America/New_York"}] +BUYS = [ + { + "media_buy_id": "mb_1", + "account_id": "acc_a", + "confirmed_at": "2026-11-03T15:00:00Z", + "start_time": "2026-11-03T05:00:00Z", + "end_time": "2026-11-30T05:00:00Z", + } +] + + +def _kit(**options): + return reporting_test_kit( + accounts=ACCOUNTS, + offerings=[offerings.daily_analytics(alignment="account_timezone")], + buys=BUYS, + **options, + ) + + +async def test_buyer_and_seller_agree_across_grants_cutovers_dst_and_cancellation() -> None: + async with _kit() as kit: + buyer = kit.buyer("agency-x") + assert (await buyer.reporting.subscribe(SUBSCRIPTION)).action == "created" + await kit.clock.advance("P5D") + report = await buyer.reporting.reconcile(now=kit.clock.now()) + assert report.definitive, report.issues + assert [(line.period["start"], line.status) for line in report.periods] == [ + ("2026-11-03T05:00:00Z", "delivered"), + ("2026-11-04T05:00:00Z", "delivered"), + ] + await kit.assert_agreement("agency-x") + await kit.grants.revoke("agency-x", "acc_a") + await kit.clock.advance("P1D") + regrant = await kit.grants.regrant("agency-x", "acc_a") + assert regrant["chain"] == "continued" and "regranted_at" in regrant + await kit.clock.advance("P2D") + await kit.assert_agreement("agency-x") + cutover = await kit.accounts.set_timezone( + "acc_a", "America/Los_Angeles", principal="agency-x" + ) + assert cutover["previous_timezone"] == "America/New_York" and cutover["cutovers"] + await kit.clock.advance("P3D") + await kit.assert_agreement("agency-x") + await kit.buys.cancel("mb_1", principal="agency-x") + await kit.clock.advance("P2D") + await kit.assert_agreement("agency-x") + + +async def test_wind_down_keeps_coverage_and_reductions_refuse_other_tasks() -> None: + async with _kit() as kit: + buyer = kit.buyer("agency-x") + await buyer.reporting.subscribe(SUBSCRIPTION) + await kit.clock.advance("P4D") + await kit.grants.wind_down("agency-x", "acc_a") + await kit.clock.advance("P1D") + report = await buyer.reporting.reconcile(now=kit.clock.now()) + assert {line.status for line in report.periods} <= {"delivered", "pending"} + await kit.assert_agreement("agency-x") + + +async def test_a_dst_day_is_owed_as_a_25_hour_period() -> None: + async with reporting_test_kit( + start="2026-10-30T00:00:00Z", + accounts=ACCOUNTS, + offerings=[offerings.daily_analytics(alignment="account_timezone")], + buys=[ + { + **BUYS[0], + "confirmed_at": "2026-10-30T00:00:00Z", + "start_time": "2026-10-31T04:00:00Z", + } + ], + ) as kit: + buyer = kit.buyer("agency-x") + await buyer.reporting.subscribe(SUBSCRIPTION) + await kit.clock.advance_to("2026-11-03T00:00:00Z") + report = await buyer.reporting.reconcile(now=kit.clock.now()) + periods = {line.period["start"]: line.period["end"] for line in report.periods} + assert periods["2026-11-01T04:00:00Z"] == "2026-11-02T05:00:00Z" + await kit.assert_agreement("agency-x") + + +async def test_a_seller_that_drops_an_owed_period_fails_agreement_with_traces(monkeypatch) -> None: + async with _kit() as kit: + buyer = kit.buyer("agency-x") + await buyer.reporting.subscribe(SUBSCRIPTION) + await kit.clock.advance("P5D") + engine = kit.service._subscriptions.engine + original = engine.owed_periods + + async def dropping(*args, **kwargs): + return (await original(*args, **kwargs))[1:] + + monkeypatch.setattr(engine, "owed_periods", dropping) + with pytest.raises(AgreementError) as error: + await kit.assert_agreement("agency-x") + assert "only the buyer owes" in str(error.value) + assert "2026-11-03T05:00:00Z" in str(error.value) + + +async def test_a_custom_platform_needs_its_service_and_controller() -> None: + with pytest.raises(ValueError, match="controller"): + async with reporting_test_kit(platform=object()): + pass + + +@pytest.mark.adcp_reporting_kit(accounts=ACCOUNTS, buys=BUYS) +async def test_the_pytest_fixture_runs_the_reference_seller(adcp_reporting_kit) -> None: + buyer = adcp_reporting_kit.buyer("agency-x") + await buyer.reporting.subscribe(SUBSCRIPTION) + await adcp_reporting_kit.clock.advance("P3D") + await adcp_reporting_kit.assert_agreement("agency-x") From d69132b186984cdfffb6d34828424b9173dd9d76 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 18:15:56 +0000 Subject: [PATCH 2/3] fix(reporting): validate every kit buyer read except fixture buys The kit buyer accepted any response that failed validation as a raw dict. Only get_media_buys keeps that fallback, for fixture buys carrying just what coverage reads; get_principal, list_accounts, get_reporting_status and the other reads reconcile and assert_agreement rely on now validate, so a malformed seller response fails instead of being compared. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/testing/reporting.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/src/adcp/testing/reporting.py b/src/adcp/testing/reporting.py index 8e2673074..418a91b01 100644 --- a/src/adcp/testing/reporting.py +++ b/src/adcp/testing/reporting.py @@ -615,10 +615,17 @@ async def _call(self, task: str, request: Any, model: Any) -> Any: await self._service.drain_subscription_triggers() if hasattr(value, "model_dump"): value = value.model_dump(mode="json", exclude_none=True, by_alias=True) - try: + if task == "get_media_buys": + # Fixture buys carry only what coverage reads; the buyer reads them + # as plain data, so they need not be complete wire buys. + try: + data = model.model_validate(value) + except ValueError: + data = value + else: + # Every response reconcile() and assert_agreement rely on must be + # valid: a malformed one fails here instead of being compared. data = model.model_validate(value) - except Exception: # noqa: BLE001 - fixture buys are minimal - data = value return TaskResult.model_construct(status=TaskStatus.COMPLETED, data=data, success=True) async def get_principal(self, request: Any) -> Any: From 6a4145829908ad7c235304fb232ad48537889b23 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Sun, 1 Nov 2026 18:56:34 +0000 Subject: [PATCH 3/3] fix(reporting): serve gaps in the boundaries of the schedule they removed After a revoke, a re-grant and a later version cutover, the managed seller restated subscription_coverage gaps[] in the new version's boundaries, so a buyer applying them computed a different owed set from the seller's own engine and ledger. Gaps are now computed for the version in force at the loss, in the zone then in effect. The minimized repro from adcp-client-python#1770 is a test-kit test asserting agreement before and after the cutover. The probe's regrant chain value is also derived from the grant history rather than a dead branch. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/reporting/subscriptions/engine.py | 20 +++++++- src/adcp/testing/reporting.py | 5 +- tests/test_reporting_test_kit.py | 53 ++++++++++++++++++++++ 3 files changed, 74 insertions(+), 4 deletions(-) diff --git a/src/adcp/reporting/subscriptions/engine.py b/src/adcp/reporting/subscriptions/engine.py index 6ed8c9755..25459c9b6 100644 --- a/src/adcp/reporting/subscriptions/engine.py +++ b/src/adcp/reporting/subscriptions/engine.py @@ -2123,7 +2123,10 @@ def _gaps( continue if expired: continue # a fresh coverage start claims no gap - schedules = self._schedules(state, account, buys, destinations, lost) + # The gap removed periods of the schedules in force at the loss: + # those of the version then current, in the zone then in effect, + # never a later version's (adcp-client-python#1770). + schedules = self._schedules(_state_at(state, lost), account, buys, destinations, lost) entries = [ (offering_id, calendar.floor(lost), calendar.ceil(regranted)) for offering_id, calendar in schedules @@ -2494,6 +2497,21 @@ async def receipt_roster(self, identity: PrincipalLike, account_id: str) -> list # -- helpers ------------------------------------------------------------------- +def _state_at(state: Mapping[str, Any], instant: datetime) -> Mapping[str, Any]: + """``state`` with the configuration of the version in force at ``instant``.""" + for prior in state.get("prior_versions") or (): + accepted = _instant(prior.get("accepted_at")) + superseded = _instant(prior.get("superseded_at")) + if ( + prior.get("configuration") is not None + and accepted is not None + and accepted <= instant + and (superseded is None or instant < superseded) + ): + return {**state, "configuration": prior["configuration"]} + return state + + def _buy_offering_ids(buys: Sequence[SubscriptionBuy]) -> list[str]: ids: set[str] = set() for buy in buys: diff --git a/src/adcp/testing/reporting.py b/src/adcp/testing/reporting.py index 418a91b01..d7ee7df7f 100644 --- a/src/adcp/testing/reporting.py +++ b/src/adcp/testing/reporting.py @@ -523,9 +523,8 @@ async def _authorization(self, principal: str, account_id: str) -> dict[str, Any days = engine.status_retention_days start = authorization_chain_start(account.grants, status_retention_days=days) regranted = authorization_regranted_at(account.grants, status_retention_days=days) - chain = ( - "continued" if regranted is not None else ("new" if len(account.grants) > 1 else "new") - ) + latest = account.grants[-1].authorized_at + chain = "continued" if start is not None and start < latest else "new" result: dict[str, Any] = {"chain": chain} if start is not None: result["authorized_at"] = iso_z(start) diff --git a/tests/test_reporting_test_kit.py b/tests/test_reporting_test_kit.py index 994f408ed..8b3771854 100644 --- a/tests/test_reporting_test_kit.py +++ b/tests/test_reporting_test_kit.py @@ -133,3 +133,56 @@ async def test_the_pytest_fixture_runs_the_reference_seller(adcp_reporting_kit) await buyer.reporting.subscribe(SUBSCRIPTION) await adcp_reporting_kit.clock.advance("P3D") await adcp_reporting_kit.assert_agreement("agency-x") + + +async def test_issue_1770_gaps_keep_the_boundaries_of_the_schedule_they_removed() -> None: + """After revoke, re-grant and a version cutover, served gaps[] match the owed set.""" + local = offerings.daily_analytics(offering_id="off_daily_local", alignment="account_timezone") + utc = offerings.daily_analytics(offering_id="off_daily_utc", alignment="utc") + selector = { + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "snapshot", + } + + def version(alignment: str) -> dict: + return { + "subscription_id": "analytics-daily", + "scope": {"all_authorized_accounts": True}, + "selector": {**selector, "alignment": alignment, "dimensions": ["package"]}, + "reconciliation_mode": "delivery_only", + } + + async with reporting_test_kit( + start="2026-09-01T00:00:00Z", + offerings=[local, utc], + accounts=[{"account_id": "acct_1", "timezone": "America/Santiago"}], + buys=[ + { + "media_buy_id": "mb_1", + "account_id": "acct_1", + "confirmed_at": "2026-09-01T00:00:00Z", + "start_time": "2026-09-02T00:00:00Z", + "end_time": "2026-09-20T00:00:00Z", + "packages": [ + { + "package_id": "p1", + "reporting_delivery_offering_ids": ["off_daily_local", "off_daily_utc"], + } + ], + } + ], + ) as kit: + buyer = kit.buyer("buyer-x") + await buyer.reporting.subscribe(version("account_timezone")) + await kit.clock.advance_to("2026-09-04T12:00:00Z") + await buyer.reporting.subscribe(version("utc")) + await kit.clock.advance_to("2026-09-08T06:00:00Z") + await kit.grants.revoke("buyer-x", "acct_1") + await kit.clock.advance_to("2026-09-09T18:00:00Z") + await kit.grants.regrant("buyer-x", "acct_1") + await kit.clock.advance_to("2026-09-12T12:00:00Z") + await kit.assert_agreement("buyer-x") + await buyer.reporting.subscribe(version("account_timezone")) + await kit.clock.advance_to("2026-09-16T00:00:00Z") + await kit.assert_agreement("buyer-x")