feat(testing): add make_request_context + build_asgi_app platform test helpers

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",
 ]

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 (``<store>:<account_id>``); 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()

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())