diff --git a/src/adcp/testing/__init__.py b/src/adcp/testing/__init__.py index 9c05f39b7..719d3f88f 100644 --- a/src/adcp/testing/__init__.py +++ b/src/adcp/testing/__init__.py @@ -20,6 +20,10 @@ from __future__ import annotations +from adcp.testing.platform_helpers import ( + build_asgi_app, + make_request_context, +) from adcp.testing.test_helpers import ( CREATIVE_AGENT_CONFIG, TEST_AGENT_A2A_CONFIG, @@ -37,6 +41,7 @@ ) __all__ = [ + # Remote client helpers (adcp.testing.test_helpers) "test_agent", "test_agent_a2a", "test_agent_no_auth", @@ -50,4 +55,7 @@ "TEST_AGENT_MCP_NO_AUTH_CONFIG", "TEST_AGENT_A2A_NO_AUTH_CONFIG", "CREATIVE_AGENT_CONFIG", + # Server-side platform test seams (adcp.testing.platform_helpers) + "make_request_context", + "build_asgi_app", ] diff --git a/src/adcp/testing/platform_helpers.py b/src/adcp/testing/platform_helpers.py new file mode 100644 index 000000000..668d2f0a2 --- /dev/null +++ b/src/adcp/testing/platform_helpers.py @@ -0,0 +1,178 @@ +"""Server-side test helpers for DecisioningPlatform unit tests. + +These helpers exist so adopters writing platform-handler tests can: + +1. Construct a :class:`~adcp.decisioning.context.RequestContext` in one + line without guessing which factory defaults are safe — use + :func:`make_request_context`. + +2. Get a runnable ASGI app from a platform in one call — use + :func:`build_asgi_app` to wire ``httpx.AsyncClient`` or + ``starlette.testclient.TestClient`` against the server in-process. + +These are the official test seams for platform-handler unit tests. +Do not call ``create_adcp_server_from_platform`` directly in tests — +:func:`build_asgi_app` sets test-appropriate defaults (no auto-emit +webhooks, single-threaded executor) that production ``serve()`` does not. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from adcp.decisioning.context import RequestContext + from adcp.decisioning.platform import DecisioningPlatform + from adcp.decisioning.types import Account + +__all__ = ["make_request_context", "build_asgi_app"] + + +def make_request_context( + account: Account[Any] | None = None, + account_id: str = "test-account", + **overrides: Any, +) -> RequestContext[Any]: + """Build a :class:`~adcp.decisioning.context.RequestContext` for unit tests. + + Adopters testing through ``PlatformHandler`` typically only read + ``ctx.account`` and sometimes ``ctx.request_id``. This helper fills + in all factory defaults so callers don't have to know which ones are + safe — the framework-owned stubs for ``state`` and ``resolve`` are + wired in automatically. + + ``caller_identity`` is defaulted to the resolved account's ``id`` for + test simplicity. The production dispatch path sets a composite opaque + key (``:``); do not assert on the exact value in + adopter tests. Override via ``**overrides`` when testing per-principal + idempotency scoping. + + :param account: Pre-built :class:`~adcp.decisioning.types.Account`. + When provided, ``account_id`` is ignored. + :param account_id: Used to construct ``Account(id=account_id)`` when + ``account`` is ``None``. Defaults to ``"test-account"``. + :param overrides: Forwarded verbatim to + :class:`~adcp.decisioning.context.RequestContext`. Any field on + ``RequestContext`` or its parent :class:`~adcp.server.base.ToolContext` + can be overridden here (e.g. ``request_id="req-123"``, + ``auth_info=...``, ``caller_identity="custom-id"``). + + Example:: + + from adcp.testing import make_request_context + + ctx = make_request_context(account_id="acme-corp") + result = await platform.get_products(req, ctx) + + Example with a pre-built Account:: + + from adcp.decisioning import Account + from adcp.testing import make_request_context + + account = Account(id="acme", metadata={"adapter": my_adapter}) + ctx = make_request_context(account=account) + """ + from adcp.decisioning.context import RequestContext + from adcp.decisioning.types import Account as _Account + + resolved: Account[Any] = account if account is not None else _Account(id=account_id) + if "caller_identity" not in overrides: + overrides["caller_identity"] = resolved.id + return RequestContext(account=resolved, **overrides) + + +def build_asgi_app( + platform: DecisioningPlatform, + *, + name: str | None = None, +) -> Any: + """Build an ASGI app from a :class:`~adcp.decisioning.platform.DecisioningPlatform`. + + Equivalent to calling ``serve(platform, name=name)`` but returns the + ASGI callable instead of blocking. Wire it to ``httpx.AsyncClient`` + (via ``httpx.ASGITransport``) or ``starlette.testclient.TestClient`` + for in-process integration tests. + + Test-appropriate defaults applied automatically (both differ from + production ``serve()``): + + - ``auto_emit_completion_webhooks=False`` — skips the F12 boot-time + webhook-sender gate that fires for webhook-eligible specialisms + (``create_media_buy``, ``activate_signal``, etc.). Tests that + exercise webhook emission should wire a sender explicitly via the + platform's own setup, not through this helper. + - ``thread_pool_size=1`` — allocates a one-thread executor instead + of the production ``min(32, cpu+4)`` default. Reduces OS-thread + churn in test suites that call this helper per test case. + + :param platform: The :class:`~adcp.decisioning.platform.DecisioningPlatform` + instance. + :param name: Server name advertised on AdCP capabilities. Defaults to + ``type(platform).__name__``, matching :func:`~adcp.decisioning.serve.serve`. + + :returns: ASGI callable (Starlette application). Compatible with + ``httpx.AsyncClient(transport=httpx.ASGITransport(app=app), ...)`` + and ``starlette.testclient.TestClient(app)``. + + .. warning:: + ``build_asgi_app`` is **synchronous** and must be called from outside + a running asyncio event loop (i.e., in a sync fixture or at module + scope). Calling it from inside an ``async def`` test will raise + ``RuntimeError: asyncio.run() cannot be called from a running event + loop`` because the boot-time capabilities validator uses + ``asyncio.run()`` internally. + + Pattern for async test suites:: + + def test_my_platform(): # sync — build app here + app = build_asgi_app(MyPlatform()) + + async def _run(): + async with LifespanManager(app): + async with httpx.AsyncClient(...) as client: + ... + + asyncio.run(_run()) + + Or use a sync ``pytest.fixture`` to build the app once per test + module and yield it into async tests. + + Example:: + + import asyncio + import httpx + from asgi_lifespan import LifespanManager + from adcp.testing import build_asgi_app + + def test_get_products(): + app = build_asgi_app(MyPlatform(), name="test-seller") + + async def _run(): + async with LifespanManager(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://localhost", + follow_redirects=True, # /mcp/ → /mcp + ) as client: + resp = await client.post("/mcp/", json={...}) + assert resp.status_code == 200 + + asyncio.run(_run()) + """ + from adcp.decisioning.serve import create_adcp_server_from_platform + from adcp.server.serve import create_mcp_server + + handler, _, _ = create_adcp_server_from_platform( + platform, + thread_pool_size=1, + auto_emit_completion_webhooks=False, + ) + server_name = name or type(platform).__name__ + mcp = create_mcp_server( + handler, + name=server_name, + # DNS rebinding protection rejects arbitrary Host headers from + # httpx.ASGITransport test clients; disable it in test contexts. + enable_dns_rebinding_protection=False, + ) + return mcp.streamable_http_app() diff --git a/tests/test_testing_platform_helpers.py b/tests/test_testing_platform_helpers.py new file mode 100644 index 000000000..fe038ce0a --- /dev/null +++ b/tests/test_testing_platform_helpers.py @@ -0,0 +1,148 @@ +"""Tests for adcp.testing.make_request_context and build_asgi_app.""" + +from __future__ import annotations + +from adcp.decisioning import ( + Account, + DecisioningCapabilities, + DecisioningPlatform, + SingletonAccounts, +) +from adcp.decisioning.capabilities import SupportedProtocol +from adcp.decisioning.context import RequestContext +from adcp.testing import build_asgi_app, make_request_context + + +class _MinimalPlatform(DecisioningPlatform): + """Smallest valid platform: media_buy protocol, agent billing, singleton account.""" + + capabilities = DecisioningCapabilities( + supported_protocols=[SupportedProtocol.media_buy], + supported_billing=("agent",), + ) + accounts = SingletonAccounts(account_id="test-singleton") + + +# --------------------------------------------------------------------------- +# make_request_context +# --------------------------------------------------------------------------- + + +def test_make_request_context_default_account_id(): + ctx = make_request_context() + assert ctx.account.id == "test-account" + + +def test_make_request_context_caller_identity_is_populated(): + # caller_identity is set for test convenience; its exact format is + # opaque (production uses a composite key). Assert non-None, not the + # string value, to avoid teaching adopters to parse the field. + ctx = make_request_context() + assert ctx.caller_identity is not None + + +def test_make_request_context_custom_account_id(): + ctx = make_request_context(account_id="acme-corp") + assert ctx.account.id == "acme-corp" + assert ctx.caller_identity is not None + + +def test_make_request_context_pre_built_account(): + acct = Account(id="pre-built", name="Acme Corp") + ctx = make_request_context(account=acct) + assert ctx.account.id == "pre-built" + assert ctx.account.name == "Acme Corp" + assert ctx.caller_identity is not None + + +def test_make_request_context_pre_built_account_ignores_account_id(): + acct = Account(id="pre-built") + ctx = make_request_context(account=acct, account_id="ignored") + assert ctx.account.id == "pre-built" + + +def test_make_request_context_override_request_id(): + ctx = make_request_context(account_id="acme", request_id="req-123") + assert ctx.request_id == "req-123" + assert ctx.account.id == "acme" + + +def test_make_request_context_override_caller_identity(): + ctx = make_request_context(account_id="acme", caller_identity="custom-principal") + assert ctx.caller_identity == "custom-principal" + assert ctx.account.id == "acme" + + +def test_make_request_context_returns_request_context_instance(): + ctx = make_request_context() + assert isinstance(ctx, RequestContext) + + +def test_make_request_context_state_and_resolve_stubs_present(): + ctx = make_request_context() + assert ctx.state is not None + assert ctx.resolve is not None + + +# --------------------------------------------------------------------------- +# build_asgi_app +# --------------------------------------------------------------------------- + + +def test_build_asgi_app_returns_callable(): + platform = _MinimalPlatform() + app = build_asgi_app(platform) + assert callable(app) + + +def test_build_asgi_app_with_explicit_name(): + platform = _MinimalPlatform() + app = build_asgi_app(platform, name="my-test-server") + assert callable(app) + + +def test_build_asgi_app_responds_to_mcp_initialize(): + """Smoke-test: the returned ASGI app handles an MCP initialize request. + + build_asgi_app calls asyncio.run() internally (validate_capabilities_response_shape); + it must be called from a sync context. The async HTTP exercise runs via + a nested asyncio.run() which is safe here because no event loop is + running in a sync pytest function. + """ + import asyncio + + import httpx + from asgi_lifespan import LifespanManager + + platform = _MinimalPlatform() + app = build_asgi_app(platform, name="smoke-test") + + async def _run() -> None: + async with LifespanManager(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://localhost", + follow_redirects=True, + ) as client: + resp = await client.post( + "/mcp/", + json={ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "test-client", "version": "1.0"}, + }, + }, + headers={ + "content-type": "application/json", + "accept": "application/json, text/event-stream", + }, + ) + assert resp.status_code == 200 + body = resp.json() + assert body.get("result") is not None + + asyncio.run(_run())