diff --git a/docs/reporting-subscriptions-seller.md b/docs/reporting-subscriptions-seller.md index 24f223592..e4a2ea729 100644 --- a/docs/reporting-subscriptions-seller.md +++ b/docs/reporting-subscriptions-seller.md @@ -31,6 +31,20 @@ The SDK reads these surfaces under each principal's identity. A principal sees e A decisioning `Account` carries `timezone=` and `operator=`. An account without a timezone is refused rather than read as UTC. +### Capabilities and serving + +The service fills in what it serves and leaves the rest to you: +- **What the service adds.** + - The `media_buy.reporting_delivery` block. + - The `adcp.principal` block, with `optimistic_concurrency: true` and `reporting_subscriptions` in `supported_sections`. + - `adcp.supported_versions`, when your capabilities do not state it. Without it a buyer negotiates down from `major_versions` and never reaches 3.3. +- **What your capabilities must carry.** + - `adcp.idempotency`, which says whether you honor `idempotency_key` replay. The schema requires it, and only you know it. + - To accept `reporting_destinations`, `adcp.principal.max_reporting_destinations`, `reporting_destination_offerings` and `suspension_interval_seconds`. Without all three, that section is not advertised. +- **Serving over MCP.** Pass `serve(..., context_factory=auth_context_factory)` (or your own factory), so `list_accounts` and `get_media_buys` see the authenticated `caller_identity`. On the managed path, a reporting task with no resolved account reads the account the request names. The subscription grant check then refuses an account the caller holds no grant on, so no `caller_resolver` is needed for `get_reporting_status` over plain MCP. + +The reference seller (`python -m adcp reference-seller reporting`) is a complete example. + ## AdCP 3.3.0-beta.3 rules the engine applies - **Admission under `max_accounts`.** Candidates are ordered by `authorization.authorized_at`, then `account_id` in byte order, and an admitted account keeps its slot. A waiting candidate carries a `capacity` exception. @@ -41,6 +55,8 @@ A decisioning `Account` carries `timezone=` and `operator=`. An account without - **Partial grants.** A grant that keeps the account covered refuses a task it does not list with `SCOPE_INSUFFICIENT`. - **Billing stubs and `regranted_at`.** A billing cutover owes a stub obligation (`stub: true`), and `list_accounts` serves `authorization.regranted_at`. +The reference seller, `adcp.testing.reporting.create_reference_reporting_seller`, is a complete example on this path. A ~100-line handler serves `list_accounts` and `get_media_buys`, and `ReliableReportingService(subscriptions=SubscriptionOptions(...))` adds the rest. Run it with `python -m adcp reference-seller reporting`; see [Testing](reporting-subscriptions-testing.md#the-reference-seller). + ## What the SDK derives - **The principal runtime.** `install()` serves `get_principal` and `sync_principal` and advertises the principal section. The service's `PrincipalService` is `reporting.principals`. diff --git a/docs/reporting-subscriptions-testing.md b/docs/reporting-subscriptions-testing.md index 767dd1c55..057eaa15c 100644 --- a/docs/reporting-subscriptions-testing.md +++ b/docs/reporting-subscriptions-testing.md @@ -59,3 +59,27 @@ Python sellers run the TypeScript storyboard CLI against their URL: ``` npx adcp storyboard run https://seller.example/mcp --storyboards reporting_subscriptions --auth "$TOKEN" ``` + +## The reference seller + +`python -m adcp reference-seller reporting --port 3000` (or `python examples/reporting_subscriptions_seller.py`) serves the SDK's reference seller as a standalone sandbox MCP server: +- MCP is at `/mcp` and A2A at `/`. +- The bearer token names the principal, and `storyboard-buyer` is used without one. +- Storage is process-local. +- The virtual clock starts at `--start` (default: now) and moves only through `comply_test_controller`. + +Its controller serves `seed_media_buy` and the `reporting_subscription_probe` scenario: +- `revoke_grant`; +- `regrant`, which returns `chain`, `authorized_at` and `regranted_at`; +- `reduce_grant`; +- `set_account_timezone`, which returns `previous_timezone` and the cutovers; +- `advance_time`, which processes every period boundary crossed. + +It passes the TypeScript `reporting_subscriptions` storyboard (15 steps, strict schema checks): + +``` +npx adcp storyboard run http://127.0.0.1:3000/mcp --file reporting-subscriptions.yaml \ + --auth storyboard-buyer --allow-http --compliance-version 3.3.0-beta.3 +``` + +The SDK's own suite runs the same storyboard in process on every change (`tests/test_reporting_reference_seller.py`). A negative control there drops `prior_versions` from the readback and fails exactly the steps that read it. diff --git a/examples/reporting_subscriptions_seller.py b/examples/reporting_subscriptions_seller.py new file mode 100644 index 000000000..0095e78fb --- /dev/null +++ b/examples/reporting_subscriptions_seller.py @@ -0,0 +1,25 @@ +"""The reference seller for AdCP 3.3 reporting subscriptions, as a standalone MCP server. + +Run it, then drive it with the storyboard runner:: + + python examples/reporting_subscriptions_seller.py --port 3000 + npx adcp storyboard run http://127.0.0.1:3000/mcp --storyboards reporting_subscriptions \ + --auth storyboard-buyer --allow-http + +It is ``adcp.testing.reporting.create_reference_reporting_seller``: a seller +that already serves ``list_accounts`` and ``get_media_buys`` (about a hundred +lines over an in-memory world), with ``ReliableReportingService(subscriptions= +SubscriptionOptions(...))`` adding everything else. The bearer token names the +principal, storage is process-local, and the virtual clock moves only through +``comply_test_controller`` ``reporting_subscription_probe`` ``advance_time``. +Equivalent: ``python -m adcp reference-seller reporting --port 3000``. +""" + +from __future__ import annotations + +import sys + +from adcp.testing.reporting import main + +if __name__ == "__main__": + sys.exit(main(["reporting", *sys.argv[1:]])) diff --git a/src/adcp/__main__.py b/src/adcp/__main__.py index 89c43203c..a5e055a87 100644 --- a/src/adcp/__main__.py +++ b/src/adcp/__main__.py @@ -694,6 +694,11 @@ def main() -> None: print(error, file=sys.stderr) sys.exit(2) sys.exit(migrate_main(sys.argv[2:])) + if len(sys.argv) > 1 and sys.argv[1] == "reference-seller": + # A sandbox reference seller, for storyboards and cross-SDK interop. + from adcp.testing.reporting import main as reference_main + + sys.exit(reference_main(sys.argv[2:])) if len(sys.argv) > 1 and sys.argv[1] == "reporting": # Reporting-subscription tooling: explain coverage, compare traces. from adcp.reporting.cli import main as reporting_main diff --git a/src/adcp/reporting/service.py b/src/adcp/reporting/service.py index b8b97f695..0389e6e7b 100644 --- a/src/adcp/reporting/service.py +++ b/src/adcp/reporting/service.py @@ -600,7 +600,9 @@ def __init__( self._initialize_source_storage = initialize_source_storage self.sources = ReportingAdapterRegistry(clock=effective_clock, staging=staging, seals=seals) self._context_resolver = account_context - self._caller_resolver = caller_resolver or self._default_caller + self._caller_resolver = caller_resolver or ( + self._managed_caller if subscriptions is not None else self._default_caller + ) self._consumer_status_enabled = consumer_status_enabled self._escalation = escalation or ReportingDeliveryEscalation() self._clock = effective_clock @@ -1611,6 +1613,29 @@ async def caller_for(self, request: Any, context: Any | None) -> ReportingStatus raise TypeError("caller_resolver must resolve ReportingStatusCaller") return caller + @classmethod + def _managed_caller(cls, request: Any, context: Any | None) -> ReportingStatusCaller: + """The managed subscription path's default caller, which also works over plain MCP. + + As :meth:`_default_caller`; when the context carries an authenticated + caller but no resolved account (a plain ``ToolContext`` from + ``serve()``), the account is the one the request names. The + subscription grant check then refuses an account the caller holds no + grant on, as for any other read (adcp-client-python#1771). + """ + try: + return cls._default_caller(request, context) + except ReliableReportingConfigurationError: + from adcp.reporting.subscriptions.managed import request_principal_id + + consumer_id = request_principal_id(context) + payload = _wire(request) if request is not None else {} + account = payload.get("account") if isinstance(payload, Mapping) else None + account_id = account.get("account_id") if isinstance(account, Mapping) else None + if not consumer_id or not account_id: + raise + return ReportingStatusCaller(account_id=str(account_id), consumer_id=consumer_id) + @staticmethod def _default_caller(request: Any, context: Any | None) -> ReportingStatusCaller: del request @@ -1886,6 +1911,19 @@ def inject_capabilities( serves_subscriptions, ) + if self._managed is not None: + # Subscriptions are served from 3.3; without supported_versions + # a buyer negotiates down from major_versions and never reaches + # them (adcp-client-python#1771). Advertise the releases this + # SDK serves unless the seller states its own. + adcp = dict(payload.get("adcp") or {}) + if not adcp.get("supported_versions"): + from adcp._version import get_supported_adcp_versions + + adcp["supported_versions"] = [ + item for item in get_supported_adcp_versions() if "-rc" not in item + ] + payload["adcp"] = adcp if self._managed is not None and serves_subscriptions(adcp_version): payload = self._managed.advertise_principal(payload) try: diff --git a/src/adcp/reporting/subscriptions/managed.py b/src/adcp/reporting/subscriptions/managed.py index ef88eb143..ee444afa0 100644 --- a/src/adcp/reporting/subscriptions/managed.py +++ b/src/adcp/reporting/subscriptions/managed.py @@ -1096,16 +1096,35 @@ async def _project_page(self, principal: PrincipalRef, page: list[Any]) -> list[ @staticmethod def advertise_principal(payload: dict[str, Any]) -> dict[str, Any]: - """Advertise the principal layer the service serves, merged with the seller's.""" + """Advertise the principal layer the service serves, merged with the seller's. + + ``reporting_subscriptions`` is always added, and ``optimistic_concurrency`` + is ``true``: the principal runtime enforces + ``expected_configuration_version``. ``reporting_destinations`` is added + only when the seller's own block carries the facts the schema requires + with it (``max_reporting_destinations``, + ``reporting_destination_offerings`` and ``suspension_interval_seconds``), + so the advertisement always validates. + """ adcp = dict(payload.get("adcp") or {}) principal = dict(adcp.get("principal") or {}) principal.setdefault("supported", True) principal.setdefault("sync_task", "sync_principal") principal.setdefault("read_task", "get_principal") + principal.setdefault("optimistic_concurrency", True) sections = [str(item) for item in principal.get("supported_sections") or ()] - for section in ("reporting_destinations", "reporting_subscriptions"): - if section not in sections: - sections.append(section) + destination_facts = ( + "max_reporting_destinations", + "reporting_destination_offerings", + "suspension_interval_seconds", + ) + if all(principal.get(name) for name in destination_facts): + if "reporting_destinations" not in sections: + sections.append("reporting_destinations") + elif "reporting_destinations" in sections: + sections.remove("reporting_destinations") + if "reporting_subscriptions" not in sections: + sections.append("reporting_subscriptions") principal["supported_sections"] = sections adcp["principal"] = principal payload = {**payload, "adcp": adcp} diff --git a/src/adcp/testing/reporting.py b/src/adcp/testing/reporting.py index d7ee7df7f..0a68a04a2 100644 --- a/src/adcp/testing/reporting.py +++ b/src/adcp/testing/reporting.py @@ -39,7 +39,7 @@ import copy from collections.abc import AsyncIterator, Mapping, Sequence from dataclasses import dataclass, field -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from typing import Any from adcp.reporting.subscriptions._clock import iso_z, parse_delivery_sla, utc_instant @@ -210,7 +210,16 @@ 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"]} + return { + "adcp": { + "major_versions": [3], + # Reporting subscriptions need the 3.3 line; advertise it so a + # runner negotiates it rather than falling back to 3.0. + "supported_versions": ["3.3-beta.3"], + "idempotency": {"supported": True, "replay_ttl_seconds": 86_400}, + }, + "supported_protocols": ["media_buy"], + } async def list_accounts(self, params: Any, context: Any = None) -> Any: principal = _principal(context) @@ -280,6 +289,17 @@ class ReferenceSeller: controller: ReportingTestController offerings: tuple[dict[str, Any], ...] + def account_resolver(self, ref: Mapping[str, Any] | None, *, auth_info: Any = None) -> Any: + """The controller's sandbox-authority gate: only the world's sandbox accounts.""" + if not ref: + return None + account = self.world.accounts.get(str(ref.get("account_id") or "")) + if account is None or not account.get("sandbox", True): + raise TestControllerError("NOT_FOUND", "not a sandbox account of this seller") + from types import SimpleNamespace + + return SimpleNamespace(id=account["account_id"], mode="sandbox", sandbox=True) + def context(self, principal: str = DEFAULT_PRINCIPAL, account_id: str | None = None) -> Any: return AccountAwareToolContext(caller_identity=principal, account_id=account_id) @@ -294,7 +314,12 @@ 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) + register_test_controller( + server, + self.controller, + context_factory=context_factory, + account_resolver=self.account_resolver, + ) return server @@ -332,7 +357,13 @@ def create_reference_reporting_seller( "sandbox": True, }, ): - record = {"name": account["account_id"], "status": "active", "timezone": "UTC", **account} + record = { + "name": account["account_id"], + "status": "active", + "timezone": "UTC", + "sandbox": True, + **account, + } world.accounts[str(record["account_id"])] = record for buy in buys: _add_buy(world, buy) @@ -586,6 +617,78 @@ def _boundaries(seller: ReferenceSeller, after: datetime, through: datetime) -> return sorted(points) +def _bearer_principal(meta: Any) -> ToolContext: + """The sandbox trust model: the bearer token names the principal.""" + request = getattr(meta, "request_context", None) + header = "" + if request is not None: + header = str(getattr(request, "headers", {}).get("authorization", "")) + token = header[7:].strip() if header.lower().startswith("bearer ") else "" + return ToolContext(caller_identity=token or DEFAULT_PRINCIPAL) + + +def serve_reference_reporting_seller( + *, + port: int | None = None, + host: str | None = None, + start: Any = None, + **options: Any, +) -> None: + """Serve the reference seller over MCP until interrupted (sandbox only). + + The bearer token names the principal (``storyboard-buyer`` without one), + storage is process-local, and the virtual clock starts at ``start`` + (default: now) and moves only through ``comply_test_controller`` + ``reporting_subscription_probe`` ``advance_time``. ``comply_test_controller`` + also serves ``seed_media_buy`` and the grant and timezone probes. + """ + import os + + from adcp.server.serve import serve + + os.environ.setdefault("ADCP_ENV", "development") + seller = create_reference_reporting_seller( + start=start or datetime.now(timezone.utc).replace(microsecond=0), **options + ) + + async def initialize() -> None: + await seller.service.run_worker() + + serve( + seller.platform, + name="adcp-reference-reporting-seller", + port=port, + host=host, + # MCP at /mcp and A2A at /; this transport runs the startup hook that + # opens the reporting service. + transport="both", + test_controller=seller.controller, + test_controller_account_resolver=seller.account_resolver, + context_factory=_bearer_principal, + on_startup=[initialize], + ) + + +def main(argv: Sequence[str]) -> int: + """``python -m adcp reference-seller reporting [--port N] [--host H] [--start ISO]``.""" + import argparse + + parser = argparse.ArgumentParser(prog="python -m adcp reference-seller reporting") + parser.add_argument("kind", choices=["reporting"]) + parser.add_argument("--port", type=int, default=None) + parser.add_argument("--host", default=None) + parser.add_argument("--start", default=None, help="virtual clock start (RFC 3339)") + parser.add_argument("--backfill-on-regrant", action="store_true") + args = parser.parse_args(list(argv)) + serve_reference_reporting_seller( + port=args.port, + host=args.host, + start=args.start, + backfill_on_regrant=args.backfill_on_regrant, + ) + return 0 + + # -- the kit --------------------------------------------------------------------------- @@ -979,6 +1082,7 @@ async def reporting_test_kit( __all__ = [ + "serve_reference_reporting_seller", "DEFAULT_PRINCIPAL", "REFERENCE_ACCOUNT", "AgreementError", diff --git a/tests/fixtures/storyboards/reporting-subscriptions.yaml b/tests/fixtures/storyboards/reporting-subscriptions.yaml new file mode 100644 index 000000000..cd4137e74 --- /dev/null +++ b/tests/fixtures/storyboards/reporting-subscriptions.yaml @@ -0,0 +1,608 @@ +id: reporting_subscriptions +version: '1.0.0' +title: 'Principal-level reporting subscriptions' +category: schema_validation +summary: 'Validates the experimental sync_principal reporting_subscriptions section: capability discovery, dry run, applied lifecycle with readback and idempotent replay, rejections, versioning with prior_versions, get_reporting_status subscription_coverage, and deactivation by omission.' +track: core +required_tools: + - get_adcp_capabilities + - sync_principal + - get_principal + - get_reporting_status + +narrative: | + A buyer registers reporting once for its whole relationship with a seller: + one `reporting_subscriptions` entry in `sync_principal` covers every account + the principal is authorized on, now and later. The seller versions each + subscription, keeps superseded versions readable through `prior_versions`, + deactivates a subscription the buyer omits instead of deleting it, and + reports per-account coverage exceptions in `get_reporting_status` + `subscription_coverage`. + + The subscription here names no destination (`reconciliation_mode: + delivery_only`), so no destination proof or outbound call is needed. The + seller under test authorizes the authenticated principal on the account + `acct-reference` (the SDK reference seller in + `examples/reporting-subscriptions-seller/seller.mjs` does), and the principal + starts with no reporting subscriptions. + + Not covered, because the runner cannot express it: + - Owed periods. Which periods a subscription owes on each account depends on + elapsed time and on media buys the runner cannot create, and the runner + cannot move the seller's clock or compare an owed set to a ledger. + - Grant changes. Revoking or reducing a principal's grant on an account, and + the `accounts_unauthorized` and `gaps` that follow, need a seller-side + change the runner cannot make. `comply_test_controller` scenarios are a + spec-defined enum, so an SDK-local storyboard cannot add one. + - The `delivery_config_id` namespace clash. It needs an account-level + `reporting_delivery_configs` entry made through `sync_accounts`, which a + seller that lists only `sync_principal` in `configuration_tasks` does not + serve, and which the runner cannot seed. + - Destination delivery, receipts and `replace_account_configs`. + +agent: + interaction_model: any + capabilities: + - 'protocol.principal' + - 'media_buy.reporting_subscriptions' + examples: + - 'The SDK reference seller in examples/reporting-subscriptions-seller' + +caller: + role: buyer_agent + example: 'Compliance test harness acting as a stable authenticated principal' + +prerequisites: + description: | + The agent under test advertises `sync_principal` with the + `reporting_subscriptions` section and the experimental + `media_buy.reporting_subscriptions` feature. The authenticated caller is + authorized on account `acct-reference` and has no reporting subscriptions. + controller_seeding: false + +phases: + - id: capability_declaration + title: 'Advertise principal-level reporting subscriptions' + narrative: | + The principal section, the reporting registration tasks, the + subscriptions block and the experimental feature id agree. + + steps: + - id: get_capabilities_declares_subscriptions + title: 'Read the subscription capability' + task: get_adcp_capabilities + schema_ref: 'protocol/get-adcp-capabilities-request.json' + response_schema_ref: 'protocol/get-adcp-capabilities-response.json' + stateful: false + sample_request: + context: + correlation_id: 'reporting_subscriptions--capabilities' + validations: + - check: response_schema + description: 'Response matches get-adcp-capabilities-response.json' + - check: field_contains + path: 'adcp.principal.supported_sections[*]' + value: 'reporting_subscriptions' + description: 'sync_principal accepts the reporting_subscriptions section' + - check: field_contains + path: 'media_buy.reporting_delivery.configuration_tasks[*]' + value: 'sync_principal' + description: 'Reporting registrations are accepted through sync_principal' + - check: field_value + path: 'media_buy.reporting_delivery.subscriptions.supported' + value: true + description: 'The subscriptions block is advertised' + - check: field_contains + path: 'experimental_features[*]' + value: 'media_buy.reporting_subscriptions' + description: 'The experimental subscriptions feature is declared' + - check: field_contains + path: 'experimental_features[*]' + value: 'protocol.principal' + description: 'The experimental principal feature is declared' + + - id: dry_run + title: 'Validate a subscription without persisting it' + depends_on: ['capability_declaration'] + narrative: | + dry_run validates the section and persists nothing. + + steps: + - id: dry_run_subscription + title: 'Dry-run one subscription' + task: sync_principal + schema_ref: 'protocol/sync-principal-request.json' + response_schema_ref: 'protocol/sync-principal-response.json' + stateful: false + sample_request: + idempotency_key: '$generate:uuid_v4#subscriptions_dry_run' + dry_run: true + configuration: + reporting_subscriptions: + - &subscription_v1 + subscription_id: 'analytics-daily' + subscription_version: 1 + active: true + scope: + all_authorized_accounts: true + selector: + feed_purpose: analytics + period_duration: P1D + required_finality: snapshot + reconciliation_mode: delivery_only + context: + correlation_id: 'reporting_subscriptions--dry_run' + validations: + - check: response_schema + description: 'Response matches sync-principal-response.json' + - check: field_value + path: 'result.kind' + value: 'validated' + description: 'The subscription validates' + - check: field_value + path: 'result.dry_run' + value: true + description: 'Nothing was applied' + + - id: dry_run_persisted_nothing + title: 'Read back: the dry run left no subscription' + task: get_principal + schema_ref: 'protocol/get-principal-request.json' + response_schema_ref: 'protocol/get-principal-response.json' + stateful: true + sample_request: + context: + correlation_id: 'reporting_subscriptions--after_dry_run' + validations: + - check: response_schema + description: 'Response matches get-principal-response.json' + - check: field_absent + path: 'result.configuration.reporting_subscriptions[0]' + description: 'The dry run persisted no subscription' + + - id: applied_lifecycle + title: 'Apply, read back and replay a subscription' + depends_on: ['dry_run'] + narrative: | + The applied response, a get_principal read and an exact replay describe + the same subscription version, acceptance time and configuration + version. + + steps: + - id: apply_subscription + title: 'Apply one subscription' + task: sync_principal + schema_ref: 'protocol/sync-principal-request.json' + response_schema_ref: 'protocol/sync-principal-response.json' + stateful: true + context_outputs: + - name: subscription_configuration_version + path: 'result.configuration_version' + - name: subscription_v1_accepted_at + path: 'result.configuration.reporting_subscriptions[0].accepted_at' + sample_request: + idempotency_key: '$generate:uuid_v4#subscriptions_apply' + configuration: + reporting_subscriptions: + - *subscription_v1 + context: + correlation_id: 'reporting_subscriptions--apply' + validations: + - check: response_schema + description: 'Response matches sync-principal-response.json' + - check: field_value + path: 'result.kind' + value: 'applied' + description: 'The subscription is persisted' + - check: field_value + path: 'result.dry_run' + value: false + description: 'Applied responses assert dry_run false' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].subscription_id' + value: 'analytics-daily' + description: 'Readback echoes the subscription_id' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].subscription_version' + value: 1 + description: 'Readback carries the accepted version' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].state' + value: 'active' + description: 'A subscription without a destination is active at once' + - check: field_present + path: 'result.configuration.reporting_subscriptions[0].accepted_at' + description: 'The seller records when it accepted this version' + - check: field_absent + path: 'result.configuration.reporting_subscriptions[0].prior_versions' + description: 'A first version has no prior versions' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].configuration.selector.feed_purpose' + value: 'analytics' + description: 'Readback carries the accepted configuration' + + - id: read_back_subscription + title: 'Read the subscription back' + task: get_principal + schema_ref: 'protocol/get-principal-request.json' + response_schema_ref: 'protocol/get-principal-response.json' + stateful: true + sample_request: + context: + correlation_id: 'reporting_subscriptions--read_back' + validations: + - check: response_schema + description: 'Response matches get-principal-response.json' + - check: field_value + path: 'result.kind' + value: 'current' + description: 'The principal record is readable' + - check: field_equals_context + path: 'result.configuration_version' + context_key: 'subscription_configuration_version' + description: 'Reading does not advance the configuration version' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].state' + value: 'active' + description: 'The subscription reads back active' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].subscription_version' + value: 1 + description: 'The subscription reads back at version 1' + - check: field_equals_context + path: 'result.configuration.reporting_subscriptions[0].accepted_at' + context_key: 'subscription_v1_accepted_at' + description: 'accepted_at is stable' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].configuration.scope.all_authorized_accounts' + value: true + description: 'Readback carries the accepted scope' + + - id: replay_subscription + title: 'Replay the applied sync with its original key and body' + task: sync_principal + schema_ref: 'protocol/sync-principal-request.json' + response_schema_ref: 'protocol/sync-principal-response.json' + stateful: true + sample_request: + idempotency_key: '$generate:uuid_v4#subscriptions_apply' + configuration: + reporting_subscriptions: + - *subscription_v1 + context: + correlation_id: 'reporting_subscriptions--replay' + validations: + - check: response_schema + description: 'Response matches sync-principal-response.json' + - check: field_value + path: 'result.kind' + value: 'applied' + description: 'The replay resolves to the applied outcome' + - check: field_equals_context + path: 'result.configuration_version' + context_key: 'subscription_configuration_version' + description: 'The replay does not advance the configuration version' + - check: field_equals_context + path: 'result.configuration.reporting_subscriptions[0].accepted_at' + context_key: 'subscription_v1_accepted_at' + description: 'The replay does not re-accept the subscription' + + - id: rejections + title: 'Reject invalid subscription sections' + depends_on: ['applied_lifecycle'] + narrative: | + An invalid section is refused whole and changes nothing. + + steps: + - id: reject_duplicate_subscription_id + title: 'Reject two entries with one subscription_id' + task: sync_principal + schema_ref: 'protocol/sync-principal-request.json' + response_schema_ref: 'protocol/sync-principal-response.json' + stateful: true + expect_error: true + negative_path: payload_well_formed + sample_request: + idempotency_key: '$generate:uuid_v4#subscriptions_duplicate' + configuration: + reporting_subscriptions: + - *subscription_v1 + - *subscription_v1 + context: + correlation_id: 'reporting_subscriptions--duplicate' + validations: + - check: field_value + path: 'result.kind' + value: 'failed' + description: 'The section is refused' + - check: field_value + path: 'result.errors[0].code' + value: 'INVALID_REQUEST' + description: 'A duplicate subscription_id is an invalid request' + + - id: reject_invalid_scope + title: 'Reject a scope that names both scope forms' + task: sync_principal + schema_ref: 'protocol/sync-principal-request.json' + stateful: true + expect_error: true + negative_path: schema_invalid + sample_request: + idempotency_key: '$generate:uuid_v4#subscriptions_invalid_scope' + configuration: + reporting_subscriptions: + - subscription_id: 'analytics-daily' + subscription_version: 1 + active: true + scope: + all_authorized_accounts: true + account_ids: ['acct-reference'] + selector: + feed_purpose: analytics + period_duration: P1D + required_finality: snapshot + reconciliation_mode: delivery_only + context: + correlation_id: 'reporting_subscriptions--invalid_scope' + validations: + - check: error_code + allowed_values: ['VALIDATION_ERROR', 'INVALID_REQUEST'] + description: 'A scope must name exactly one of all_authorized_accounts or account_ids' + + - id: reject_reused_version + title: 'Reject changed content under the same subscription_version' + task: sync_principal + schema_ref: 'protocol/sync-principal-request.json' + response_schema_ref: 'protocol/sync-principal-response.json' + stateful: true + expect_error: true + negative_path: payload_well_formed + sample_request: + idempotency_key: '$generate:uuid_v4#subscriptions_reused_version' + configuration: + # A selector change is semantic. Membership edits (account_ids, exclude_account_ids) apply + # in place under the same version (adcp#8619). + reporting_subscriptions: + - subscription_id: 'analytics-daily' + subscription_version: 1 + active: true + scope: + all_authorized_accounts: true + selector: + feed_purpose: analytics + period_duration: P1D + required_finality: official + reconciliation_mode: delivery_only + context: + correlation_id: 'reporting_subscriptions--reused_version' + validations: + - check: field_value + path: 'result.kind' + value: 'failed' + description: 'Changed content needs a new subscription_version' + - check: field_value + path: 'result.errors[0].code' + value: 'INVALID_REQUEST' + description: 'Reusing a version for different content is an invalid request' + + - id: rejections_changed_nothing + title: 'Read back: the rejections changed nothing' + task: get_principal + schema_ref: 'protocol/get-principal-request.json' + response_schema_ref: 'protocol/get-principal-response.json' + stateful: true + sample_request: + context: + correlation_id: 'reporting_subscriptions--after_rejections' + validations: + - check: field_equals_context + path: 'result.configuration_version' + context_key: 'subscription_configuration_version' + description: 'No rejected request advanced the configuration version' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].subscription_version' + value: 1 + description: 'The subscription is still at version 1' + + - id: versioning + title: 'Supersede a version and keep the prior one readable' + depends_on: ['rejections'] + narrative: | + A changed subscription carries a new subscription_version. The readback + lists the superseded version in prior_versions with when it was + accepted and when it was superseded. + + steps: + - id: apply_version_2 + title: 'Apply version 2 with an account exclusion' + task: sync_principal + schema_ref: 'protocol/sync-principal-request.json' + response_schema_ref: 'protocol/sync-principal-response.json' + stateful: true + context_outputs: + - name: subscription_v2_accepted_at + path: 'result.configuration.reporting_subscriptions[0].accepted_at' + sample_request: + idempotency_key: '$generate:uuid_v4#subscriptions_version_2' + configuration: + reporting_subscriptions: + - subscription_id: 'analytics-daily' + subscription_version: 2 + active: true + scope: + all_authorized_accounts: true + exclude_account_ids: ['acct-excluded'] + selector: + feed_purpose: analytics + period_duration: P1D + required_finality: snapshot + reconciliation_mode: delivery_only + context: + correlation_id: 'reporting_subscriptions--version_2' + validations: + - check: response_schema + description: 'Response matches sync-principal-response.json' + - check: field_value + path: 'result.kind' + value: 'applied' + description: 'The new version is persisted' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].subscription_version' + value: 2 + description: 'Readback carries the new version' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].prior_versions[0].subscription_version' + value: 1 + description: 'The superseded version is listed' + - check: field_equals_context + path: 'result.configuration.reporting_subscriptions[0].prior_versions[0].accepted_at' + context_key: 'subscription_v1_accepted_at' + description: 'The prior version keeps its acceptance time' + - check: field_present + path: 'result.configuration.reporting_subscriptions[0].prior_versions[0].superseded_at' + description: 'The prior version records when it was superseded' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].configuration.exclude_account_ids[0]' + value: 'acct-excluded' + description: 'Readback carries the new configuration' + + - id: read_back_version_2 + title: 'Read version 2 back' + task: get_principal + schema_ref: 'protocol/get-principal-request.json' + response_schema_ref: 'protocol/get-principal-response.json' + stateful: true + context_outputs: + - name: subscription_v2_configuration_version + path: 'result.configuration_version' + sample_request: + context: + correlation_id: 'reporting_subscriptions--read_version_2' + validations: + - check: response_schema + description: 'Response matches get-principal-response.json' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].subscription_version' + value: 2 + description: 'The subscription reads back at version 2' + - check: field_equals_context + path: 'result.configuration.reporting_subscriptions[0].accepted_at' + context_key: 'subscription_v2_accepted_at' + description: 'Version 2 keeps its acceptance time' + - check: field_equals_context + path: 'result.configuration.reporting_subscriptions[0].prior_versions[0].superseded_at' + context_key: 'subscription_v2_accepted_at' + description: 'Version 1 was superseded when version 2 was accepted' + + - id: coverage + title: 'Report subscription coverage per account' + depends_on: ['versioning'] + narrative: | + get_reporting_status lists one subscription_coverage entry for each of + the caller's subscriptions whose scope includes the account. The account + is in scope and has no exceptions. + + steps: + - id: get_reporting_status_coverage + title: 'Read the account summary' + task: get_reporting_status + schema_ref: 'media-buy/get-reporting-status-request.json' + response_schema_ref: 'media-buy/get-reporting-status-response.json' + stateful: true + sample_request: + account: + account_id: 'acct-reference' + view: summary + context: + correlation_id: 'reporting_subscriptions--coverage' + validations: + - check: response_schema + description: 'Response matches get-reporting-status-response.json' + - check: field_value + path: 'subscription_coverage[0].subscription_id' + value: 'analytics-daily' + description: 'The subscription covers the account' + - check: field_value + path: 'subscription_coverage[0].subscription_version' + value: 2 + description: 'Coverage follows the current version' + - check: field_value + path: 'subscription_coverage[0].status' + value: 'active' + description: 'The account has no coverage exception' + + - id: deactivation + title: 'Deactivate a subscription by omission' + depends_on: ['coverage'] + narrative: | + sync_principal replaces the whole section. A subscription the buyer + omits is deactivated, not deleted: it stays readable as inactive with + deactivated_at and its version history. + + steps: + - id: omit_subscription + title: 'Sync an empty reporting_subscriptions section' + task: sync_principal + schema_ref: 'protocol/sync-principal-request.json' + response_schema_ref: 'protocol/sync-principal-response.json' + stateful: true + context_outputs: + - name: subscription_deactivated_at + path: 'result.configuration.reporting_subscriptions[0].deactivated_at' + sample_request: + idempotency_key: '$generate:uuid_v4#subscriptions_omit' + configuration: + reporting_subscriptions: [] + context: + correlation_id: 'reporting_subscriptions--omit' + validations: + - check: response_schema + description: 'Response matches sync-principal-response.json' + - check: field_value + path: 'result.kind' + value: 'applied' + description: 'The omission is persisted' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].subscription_id' + value: 'analytics-daily' + description: 'The omitted subscription stays readable' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].state' + value: 'inactive' + description: 'Omission deactivates' + - check: field_present + path: 'result.configuration.reporting_subscriptions[0].deactivated_at' + description: 'The seller records when it deactivated' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].configuration.active' + value: false + description: 'The configuration reads back inactive' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].prior_versions[0].subscription_version' + value: 1 + description: 'Version history survives deactivation' + + - id: read_back_inactive + title: 'Read the inactive subscription back' + task: get_principal + schema_ref: 'protocol/get-principal-request.json' + response_schema_ref: 'protocol/get-principal-response.json' + stateful: true + sample_request: + context: + correlation_id: 'reporting_subscriptions--read_inactive' + validations: + - check: response_schema + description: 'Response matches get-principal-response.json' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].state' + value: 'inactive' + description: 'The subscription reads back inactive' + - check: field_equals_context + path: 'result.configuration.reporting_subscriptions[0].deactivated_at' + context_key: 'subscription_deactivated_at' + description: 'deactivated_at is stable' + - check: field_value + path: 'result.configuration.reporting_subscriptions[0].subscription_version' + value: 2 + description: 'The current version is unchanged by deactivation' diff --git a/tests/storyboard_runner.py b/tests/storyboard_runner.py new file mode 100644 index 000000000..818b556ea --- /dev/null +++ b/tests/storyboard_runner.py @@ -0,0 +1,211 @@ +"""A minimal in-process runner for AdCP storyboard YAML (test helper). + +It executes the checks the reporting-subscriptions storyboard uses +(``response_schema``, ``field_value``, ``field_present``, ``field_absent``, +``field_contains``, ``field_equals_context``, ``error_code``), with +``context_outputs`` and ``$generate:uuid_v4#name``. The authoritative runner +is the TypeScript ``adcp storyboard run``; this one lets Python run the same +file in process, so the reference seller is checked on every test run. +""" + +from __future__ import annotations + +import copy +import re +import uuid +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + +_MISSING = object() +_TOKEN = re.compile(r"([^.\[\]]+)|\[(\d+|\*)\]") + +#: ``(task, arguments) -> (structured response, adcp_error or None)`` +Call = Callable[[str, dict[str, Any]], Awaitable[tuple[dict[str, Any], dict[str, Any] | None]]] + + +@dataclass +class ValidationResult: + check: str + passed: bool + path: str | None + message: str + + +@dataclass +class StepResult: + step_id: str + passed: bool + validations: list[ValidationResult] = field(default_factory=list) + + +@dataclass +class StoryboardResult: + steps: list[StepResult] + + @property + def passed(self) -> bool: + return all(step.passed for step in self.steps) + + @property + def failed_steps(self) -> list[str]: + return [step.step_id for step in self.steps if not step.passed] + + +def load_storyboard(path: Path) -> dict[str, Any]: + document = yaml.safe_load(path.read_text()) + assert isinstance(document, dict) + return document + + +def resolve_path(document: Any, path: str) -> Any: + """``a.b[0].c`` or ``a[*].b``; a ``[*]`` returns the list of matches.""" + values = [document] + fanned = False + for name, index in _TOKEN.findall(path): + next_values = [] + for value in values: + if name: + if isinstance(value, Mapping) and name in value: + next_values.append(value[name]) + elif index == "*": + fanned = True + if isinstance(value, list): + next_values.extend(value) + elif isinstance(value, list) and int(index) < len(value): + next_values.append(value[int(index)]) + values = next_values + if fanned: + return values + return values[0] if values else _MISSING + + +def _generate(value: Any, generated: dict[str, str]) -> Any: + if isinstance(value, str) and value.startswith("$generate:uuid_v4"): + name = value.partition("#")[2] or str(uuid.uuid4()) + return generated.setdefault(name, str(uuid.uuid4())) + if isinstance(value, Mapping): + return {key: _generate(item, generated) for key, item in value.items()} + if isinstance(value, list): + return [_generate(item, generated) for item in value] + return value + + +def _error_code(structured: Mapping[str, Any], error: Mapping[str, Any] | None) -> str | None: + if error is not None: + return str(error.get("code")) + errors = resolve_path(structured, "result.errors") + if isinstance(errors, list) and errors: + return str(errors[0].get("code")) + errors = structured.get("errors") + if isinstance(errors, list) and errors: + return str(errors[0].get("code")) + return None + + +def _validate( + check: Mapping[str, Any], + step: Mapping[str, Any], + structured: dict[str, Any], + error: dict[str, Any] | None, + context: dict[str, Any], + schema: Callable[[str], Any] | None, +) -> ValidationResult: + kind = str(check["check"]) + path = check.get("path") + actual = resolve_path(structured, path) if path else _MISSING + if kind == "response_schema": + ref = step.get("response_schema_ref") + if not ref or schema is None: + return ValidationResult(kind, True, None, "no response schema") + validator = schema(ref) + if validator is None: + return ValidationResult(kind, False, None, f"no validator for {ref}") + problems = [item.message for item in validator.iter_errors(structured)] + return ValidationResult(kind, not problems, None, "; ".join(problems[:3])) + if kind == "field_value": + ok = actual is not _MISSING and actual == check["value"] + return ValidationResult(kind, ok, path, f"{actual!r} != {check['value']!r}") + if kind == "field_present": + return ValidationResult(kind, actual is not _MISSING, path, "absent") + if kind == "field_absent": + return ValidationResult(kind, actual is _MISSING, path, f"present: {actual!r}") + if kind == "field_contains": + ok = isinstance(actual, list) and check["value"] in actual + return ValidationResult(kind, ok, path, f"{check['value']!r} not in {actual!r}") + if kind == "field_equals_context": + expected = context.get(check["context_key"], _MISSING) + ok = actual is not _MISSING and actual == expected + return ValidationResult(kind, ok, path, f"{actual!r} != context {expected!r}") + if kind == "error_code": + code = _error_code(structured, error) + ok = code in set(check.get("allowed_values") or [check.get("value")]) + return ValidationResult(kind, ok, None, f"error code {code!r}") + return ValidationResult(kind, False, path, f"unsupported check {kind!r}") + + +async def run_storyboard( + call: Call, + storyboard: Mapping[str, Any], + *, + schema: Callable[[str], Any] | None = None, + adcp_version: str | None = None, +) -> StoryboardResult: + """Run every phase and step in order against ``call``. + + ``adcp_version`` is sent on every request, as the TypeScript runner sends + the negotiated release. + """ + generated: dict[str, str] = {} + context: dict[str, Any] = {} + results: list[StepResult] = [] + for phase in storyboard["phases"]: + for step in phase["steps"]: + payload = _generate(copy.deepcopy(step.get("sample_request") or {}), generated) + if adcp_version is not None: + payload.setdefault("adcp_version", adcp_version) + structured, error = await call(step["task"], payload) + validations = [ + _validate(check, step, structured, error, context, schema) + for check in step.get("validations") or () + ] + if error is not None and not step.get("expect_error"): + validations.append( + ValidationResult("transport", False, None, f"unexpected error {error}") + ) + for output in step.get("context_outputs") or (): + value = resolve_path(structured, output["path"]) + if value is not _MISSING: + context[output["name"]] = value + results.append( + StepResult(step["id"], all(item.passed for item in validations), validations) + ) + return StoryboardResult(results) + + +def mcp_call(server: Any) -> Call: + """A :data:`Call` over an in-process FastMCP server (``server.call_tool``).""" + + async def call( + task: str, payload: dict[str, Any] + ) -> tuple[dict[str, Any], dict[str, Any] | None]: + from mcp.types import CallToolResult + + try: + raw = await server.call_tool(task, payload) + except Exception as exc: # noqa: BLE001 - a tool error is a step outcome + data = getattr(exc, "args", [None])[0] + return {}, {"code": "TOOL_ERROR", "message": str(data)} + if isinstance(raw, CallToolResult): + structured: dict[str, Any] = dict(raw.structured_content or {}) + elif isinstance(raw, (tuple, list)) and len(raw) == 2: + structured = dict(raw[1] or {}) + else: + structured = dict(raw) + error = structured.get("adcp_error") + return structured, error + + return call diff --git a/tests/test_reporting_reference_seller.py b/tests/test_reporting_reference_seller.py new file mode 100644 index 000000000..5c506d421 --- /dev/null +++ b/tests/test_reporting_reference_seller.py @@ -0,0 +1,187 @@ +"""The Python reference seller for reporting subscriptions. + +It passes the same storyboard the TypeScript reference seller passes +(adcp-client ``examples/storyboards/reporting-subscriptions.yaml``, vendored +below), run in process over MCP; a seller whose readback drops +``prior_versions`` fails it. Its ``comply_test_controller`` serves +``reporting_subscription_probe``, so a remote runner can drive grants, +timezones and the clock, and owed periods match what a buyer computes. +""" + +from __future__ import annotations + +import hashlib +from pathlib import Path + +import pytest + +from adcp.testing.reporting import create_reference_reporting_seller +from adcp.validation.schema_loader import get_named_validator +from tests.storyboard_runner import load_storyboard, mcp_call, run_storyboard + +pytestmark = pytest.mark.asyncio + +STORYBOARD_PATH = ( + Path(__file__).parent / "fixtures" / "storyboards" / "reporting-subscriptions.yaml" +) +#: adcontextprotocol/adcp-client commit the vendored storyboard comes from (PR #3534 and #3560). +STORYBOARD_COMMIT = "eeeb370dec63002ddf599823f63b9c1115df3436" +STORYBOARD_SHA256 = "5eac5cb2468d408d5fa08a4fb7735eab3b508b68e970f6c985aad9fdf191db7a" +VERSION = "3.3-beta.3" + + +def test_the_vendored_storyboard_is_the_pinned_typescript_file() -> None: + assert hashlib.sha256(STORYBOARD_PATH.read_bytes()).hexdigest() == STORYBOARD_SHA256 + + +async def _run(seller): + await seller.service.run_worker() + return await run_storyboard( + mcp_call(seller.mcp_server()), + load_storyboard(STORYBOARD_PATH), + schema=get_named_validator, + adcp_version=VERSION, + ) + + +async def test_the_reference_seller_passes_every_storyboard_step() -> None: + result = await _run(create_reference_reporting_seller()) + failures = { + step.step_id: [(v.check, v.path, v.message) for v in step.validations if not v.passed] + for step in result.steps + if not step.passed + } + assert result.passed, failures + assert len(result.steps) >= 15 + + +async def test_a_seller_whose_readback_drops_prior_versions_fails_it() -> None: + seller = create_reference_reporting_seller() + platform = seller.platform + + def dropping(original): + async def call(params, context=None): + response = await original(params, context) + configuration = getattr(getattr(response, "result", None), "configuration", None) + for state in getattr(configuration, "reporting_subscriptions", None) or (): + state.prior_versions = None + return response + + return call + + platform.sync_principal = dropping(platform.sync_principal) + platform.get_principal = dropping(platform.get_principal) + result = await _run(seller) + assert not result.passed + # Every step that reads prior_versions fails, and nothing else does. + assert result.failed_steps == ["apply_version_2", "read_back_version_2", "omit_subscription"] + failures = [v for step in result.steps for v in step.validations if not v.passed] + assert failures + assert all(".prior_versions[0]." in (v.path or "") for v in failures), failures + + +async def test_the_controller_drives_grants_and_the_clock_over_mcp() -> None: + from adcp.testing.reporting import KitBuyer + + seller = create_reference_reporting_seller(start="2026-11-01T00:00:00Z") + await seller.service.run_worker() + call = mcp_call(seller.mcp_server(principal="agency-x")) + + account = {"account_id": "acct-reference", "sandbox": True} + + async def control(scenario, **params): + structured, error = await call( + "comply_test_controller", + {"scenario": scenario, "params": params, "account": account, "adcp_version": VERSION}, + ) + assert error is None and structured.get("success"), structured + return structured + + listed, _ = await call( + "comply_test_controller", {"scenario": "list_scenarios", "account": account} + ) + assert {"reporting_subscription_probe", "seed_media_buy"} <= set(listed["scenarios"]) + await control( + "seed_media_buy", + media_buy_id="mb_1", + fixture={ + "account_id": "acct-reference", + "confirmed_at": "2026-11-01T00:00:00Z", + "start_time": "2026-11-01T00:00:00Z", + "end_time": "2026-11-20T00:00:00Z", + }, + ) + buyer = KitBuyer(seller.platform, "agency-x", seller.service) + await buyer.reporting.subscribe( + { + "subscription_id": "analytics-daily", + "scope": {"all_authorized_accounts": True}, + "selector": { + "feed_purpose": "analytics", + "period_duration": "P1D", + "required_finality": "snapshot", + }, + "reconciliation_mode": "delivery_only", + } + ) + moved = await control( + "reporting_subscription_probe", operation="advance_time", target_time="2026-11-04T07:00:00Z" + ) + assert moved["simulated_now"] == "2026-11-04T07:00:00Z" + status, _ = await call( + "get_reporting_status", + {"account": {"account_id": "acct-reference"}, "view": "periods", "adcp_version": VERSION}, + ) + served = sorted(period["period"]["start"] for period in status["periods"]) + report = await buyer.reporting.reconcile(now=seller.clock()) + # owed_periods_match: what the seller planned is what the buyer computes. + assert served == sorted(line.period["start"] for line in report.periods) + assert served == ["2026-11-01T00:00:00Z", "2026-11-02T00:00:00Z", "2026-11-03T00:00:00Z"] + await control( + "reporting_subscription_probe", operation="revoke_grant", account_id="acct-reference" + ) + principal, _ = await call("get_principal", {"adcp_version": VERSION}) + state = principal["result"]["configuration"]["reporting_subscriptions"][0] + assert [entry["account_id"] for entry in state["accounts_unauthorized"]] == ["acct-reference"] + await control( + "reporting_subscription_probe", operation="advance_time", target_time="2026-11-04T09:00:00Z" + ) + regrant = await control( + "reporting_subscription_probe", operation="regrant", account_id="acct-reference" + ) + assert regrant["chain"] == "continued" and regrant["regranted_at"] == "2026-11-04T09:00:00Z" + zone = await control( + "reporting_subscription_probe", + operation="set_account_timezone", + account_id="acct-reference", + timezone="Europe/Paris", + ) + assert zone["previous_timezone"] == "UTC" + reduced = await control( + "reporting_subscription_probe", + operation="reduce_grant", + account_id="acct-reference", + allowed_tasks=["get_reporting_status"], + ) + assert reduced["success"] + + +def test_the_standalone_server_maps_bearer_tokens_to_principals(monkeypatch) -> None: + from types import SimpleNamespace + + from adcp.testing import reporting + + meta = SimpleNamespace( + request_context=SimpleNamespace(headers={"authorization": "Bearer agency-x"}) + ) + assert reporting._bearer_principal(meta).caller_identity == "agency-x" + assert reporting._bearer_principal(SimpleNamespace()).caller_identity == "storyboard-buyer" + served: dict = {} + import sys + + monkeypatch.setattr( + sys.modules["adcp.server.serve"], "serve", lambda platform, **kwargs: served.update(kwargs) + ) + assert reporting.main(["reporting", "--port", "3999", "--start", "2026-11-01T00:00:00Z"]) == 0 + assert served["port"] == 3999 and served["test_controller"] is not None + assert served["context_factory"] is reporting._bearer_principal diff --git a/tests/test_reporting_subscription_managed.py b/tests/test_reporting_subscription_managed.py index 6a71e2e69..249d33fad 100644 --- a/tests/test_reporting_subscription_managed.py +++ b/tests/test_reporting_subscription_managed.py @@ -183,9 +183,9 @@ async def test_capabilities_advertise_the_principal_layer_the_service_serves() - capabilities = await world.platform.get_adcp_capabilities({}, _ctx()) principal = capabilities["adcp"]["principal"] assert principal["sync_task"] == "sync_principal" and principal["read_task"] == "get_principal" - assert {"reporting_destinations", "reporting_subscriptions"} <= set( - principal["supported_sections"] - ) + # reporting_destinations needs seller facts this fixture does not advertise. + assert principal["supported_sections"] == ["reporting_subscriptions"] + assert principal["optimistic_concurrency"] is True assert {"protocol.principal", "media_buy.reporting_subscriptions"} <= set( capabilities["experimental_features"] ) @@ -572,3 +572,43 @@ async def hung(params: Any, context: Any = None) -> Any: [failure] = world.failures assert failure.error.code == "SUBSCRIPTION_ROSTER_UNAVAILABLE" assert "RuntimeError" in str(failure.error) + + +@pytest.mark.asyncio +async def test_issue_1771_managed_capabilities_pass_the_sdk_response_validation( + monkeypatch, +) -> None: + from adcp.validation.schema_loader import get_named_validator + + base = Seller.get_adcp_capabilities + + async def capabilities(self: Any, params: Any, context: Any = None) -> Any: + payload = await base(self, params, context) + # The seller's own fact: whether it honors idempotency_key replay. + payload["adcp"]["idempotency"] = {"supported": True, "replay_ttl_seconds": 86400} + return payload + + monkeypatch.setattr(Seller, "get_adcp_capabilities", capabilities) + world = _World() + served = await world.platform.get_adcp_capabilities({"adcp_version": "3.3-beta.3"}, _ctx()) + validator = get_named_validator("protocol/get-adcp-capabilities-response.json") + payload = {"status": "completed", **served} + errors = [f"{list(e.absolute_path)} {e.message}" for e in validator.iter_errors(payload)] + assert errors == [] + # A buyer negotiates the 3.3 line instead of falling back to 3.0. + assert "3.3-beta.3" in served["adcp"]["supported_versions"] + + +@pytest.mark.asyncio +async def test_issue_1771_status_is_served_to_a_plain_mcp_caller() -> None: + from adcp.server.base import ToolContext + + world = _World() + await world.subscribe() + world.clock.set(START + timedelta(hours=3, minutes=30)) + await world.service.run_worker() + status = await world.platform.get_reporting_status( + {"account": {"account_id": ACCOUNT}, "view": "summary"}, + ToolContext(caller_identity=AGENCY), + ) + assert status["subscription_coverage"][0]["subscription_id"] == "analytics-hourly"