Skip to content

testing-guide.md

Latest commit

 

History

History
622 lines (487 loc) · 18.3 KB

testing-guide.md

File metadata and controls

622 lines (487 loc) · 18.3 KB
Raw

AdCP SDK Testing Guide

Best practices for writing tests that use the AdCP Python SDK.

Testing Philosophy

When testing code that uses the AdCP SDK, follow these principles:

  1. Test public API, not internal types - Import from adcp, not adcp.types._generated
  2. Test wire format compatibility - Use JSON fixtures to validate protocol compliance
  3. Test user workflows - Write tests that tell stories about how users accomplish goals
  4. Test behavior, not implementation - Focus on what users can do, not how it works internally

Quick Reference

✅ DO

# Import from public API
from adcp import Product, CreateMediaBuyRequest, CreateMediaBuySuccessResponse

# Test wire format with JSON
def test_product_deserializes():
    json_data = '{"product_id": "test", ...}'
    product = Product.model_validate_json(json_data)
    assert product.product_id == "test"

# Test user workflows
async def test_buyer_discovers_products():
    client = ADCPClient(config)
    result = await client.get_products(request)
    assert result.success

❌ DON'T

# Don't import from the internal consolidation, and never a numbered class
from adcp.types._generated import Product1  # ❌ WRONG
from adcp.types.domains.core.product import PublisherProperties4  # ❌ WRONG (numbered)

# Don't test Pydantic mechanics
def test_discriminator_field_enforced():  # ❌ WRONG - testing Pydantic, not our code
    ...

# Don't test type identity
def test_alias_points_to_generated_type():  # ❌ WRONG - internal detail
    assert CreateMediaBuySuccessResponse is CreateMediaBuyResponse1

Pattern 1: Test Wire Format Compatibility

These tests validate that the SDK correctly handles protocol JSON. They catch serialization bugs, field name mismatches, and type coercion issues.

Example: Deserialize Protocol JSON

def test_get_products_response_deserializes_from_protocol_json():
    """GetProductsResponse deserializes from actual protocol JSON."""
    # This JSON comes from a real agent response
    protocol_json = """
    {
      "products": [
        {
          "product_id": "premium_display",
          "name": "Premium Display Placements",
          "description": "High-visibility ad slots on homepage",
          "publisher_properties": [
            {
              "publisher_domain": "example.com",
              "selection_type": "by_id",
              "property_ids": ["homepage", "mobile_app"]
            }
          ],
          "pricing_options": [
            {
              "pricing_model": "cpm",
              "pricing_option_id": "po-premium-1",
              "currency": "USD",
              "fixed_price": 5.50
            }
          ]
        }
      ]
    }
    """

    from adcp import GetProductsResponse

    # ✅ TEST: Can we parse actual protocol JSON?
    response = GetProductsResponse.model_validate_json(protocol_json)

    # Verify structure
    assert len(response.products) == 1
    product = response.products[0]
    assert product.product_id == "premium_display"

    # ✅ TEST: Round-trip preserves data?
    roundtrip_json = response.model_dump_json()
    roundtrip = GetProductsResponse.model_validate_json(roundtrip_json)
    assert roundtrip.products[0].product_id == product.product_id

Example: Test Discriminated Union Variants

def test_create_media_buy_success_response_wire_format():
    """CreateMediaBuySuccessResponse deserializes success variant."""
    protocol_json = """
    {
      "media_buy_id": "mb_123456",
      "buyer_ref": "campaign_abc",
      "packages": [
        {
          "package_id": "pkg_001",
          "product_id": "premium_display",
          "status": "pending"
        }
      ]
    }
    """

    # ✅ CORRECT: Use semantic alias from public API
    from adcp import CreateMediaBuySuccessResponse

    response = CreateMediaBuySuccessResponse.model_validate_json(protocol_json)

    assert response.media_buy_id == "mb_123456"
    assert not hasattr(response, "errors")  # Success variant doesn't have errors
    assert len(response.packages) == 1

def test_create_media_buy_error_response_wire_format():
    """CreateMediaBuyErrorResponse deserializes error variant."""
    protocol_json = """
    {
      "errors": [
        {
          "code": "budget_exceeded",
          "message": "Requested budget exceeds account limit"
        }
      ]
    }
    """

    from adcp import CreateMediaBuyErrorResponse

    response = CreateMediaBuyErrorResponse.model_validate_json(protocol_json)

    assert len(response.errors) == 1
    assert response.errors[0].code == "budget_exceeded"
    assert not hasattr(response, "media_buy_id")  # Error variant doesn't have media_buy_id

Pattern 2: Test User Workflows

These tests tell stories about how users accomplish goals. They focus on external behavior users care about.

Example: Product Discovery Workflow

@pytest.mark.asyncio
async def test_buyer_discovers_products_for_coffee_campaign(mocker):
    """Buyer gets products suitable for coffee brand campaign."""
    # Setup: Create client
    from adcp import ADCPClient, AgentConfig, Protocol, GetProductsRequest

    config = AgentConfig(
        id="publisher_agent",
        agent_uri="https://publisher.example.com",
        protocol=Protocol.A2A,
    )
    client = ADCPClient(config)

    # Setup: Mock agent response with realistic data
    from adcp.types.core import TaskResult, TaskStatus

    mock_response_data = {
        "products": [
            {
                "product_id": "breakfast_readers",
                "name": "Morning News Readers",
                "description": "Reach coffee drinkers during morning news",
                "publisher_properties": [
                    {
                        "publisher_domain": "news.example.com",
                        "selection_type": "by_tag",
                        "property_tags": ["morning", "lifestyle"],
                    }
                ],
                "pricing_options": [
                    {
                        "pricing_model": "cpm",
                        "pricing_option_id": "po-breakfast-1",
                        "currency": "USD",
                        "fixed_price": 4.50,
                    }
                ],
            }
        ]
    }

    mock_result = TaskResult(
        status=TaskStatus.COMPLETED,
        data=mock_response_data,
        success=True
    )

    mocker.patch.object(client.adapter, "get_products", return_value=mock_result)

    # Action: User discovers products
    request = GetProductsRequest(brief="Coffee brand campaign for morning audience")
    result = await client.get_products(request)

    # Assert: User gets successful result
    assert result.success, f"Discovery failed: {result.error}"
    assert len(result.data.products) > 0, "No products found"

    # Assert: Product has campaign-relevant data
    product = result.data.products[0]
    assert product.product_id
    assert product.name
    assert len(product.pricing_options) > 0

    # Assert: Can plan budget from pricing
    pricing = product.pricing_options[0]
    assert pricing.pricing_model == "cpm"
    # Fixed pricing has fixed_price, auction pricing has floor_price
    assert pricing.fixed_price is not None or pricing.floor_price is not None

Example: Handle Empty Results

@pytest.mark.asyncio
async def test_buyer_handles_no_products_available(mocker):
    """Buyer gracefully handles when no products match criteria."""
    from adcp import ADCPClient, AgentConfig, Protocol, GetProductsRequest
    from adcp.types.core import TaskResult, TaskStatus

    config = AgentConfig(
        id="publisher_agent",
        agent_uri="https://publisher.example.com",
        protocol=Protocol.A2A,
    )
    client = ADCPClient(config)

    # Mock empty response
    mock_result = TaskResult(
        status=TaskStatus.COMPLETED,
        data={"products": []},
        success=True
    )

    mocker.patch.object(client.adapter, "get_products", return_value=mock_result)

    # User makes request
    request = GetProductsRequest(brief="Extremely niche requirement")
    result = await client.get_products(request)

    # Should succeed with empty list (not error)
    assert result.success
    assert result.data.products == []

Pattern 3: Test Public API Behavior

These tests verify that API methods work as documented. They focus on method signatures, return types, and error handling.

Example: Test Method Accepts Correct Types

@pytest.mark.asyncio
async def test_create_media_buy_accepts_request_object(mocker):
    """create_media_buy accepts CreateMediaBuyRequest and returns response."""
    from adcp import (
        ADCPClient,
        AgentConfig,
        Protocol,
        CreateMediaBuyRequest,
        CreateMediaBuySuccessResponse,
    )
    from adcp.types.core import TaskResult, TaskStatus

    config = AgentConfig(
        id="agent",
        agent_uri="https://agent.example.com",
        protocol=Protocol.A2A
    )
    client = ADCPClient(config)

    # Mock successful response
    mock_result = TaskResult(
        status=TaskStatus.COMPLETED,
        data={
            "media_buy_id": "mb_123",
            "buyer_ref": "campaign_456",
            "packages": [],
        },
        success=True,
    )

    mocker.patch.object(client.adapter, "create_media_buy", return_value=mock_result)

    # ✅ TEST: Can user create request and call method?
    request = CreateMediaBuyRequest(
        buyer_ref="campaign_456",
        packages=[
            {
                "product_id": "premium_display",
                "budget": {"amount": 10000.0, "currency": "USD"},
            }
        ],
    )

    result = await client.create_media_buy(request)

    # ✅ TEST: Does result have expected structure?
    assert result.success
    assert isinstance(result.data, CreateMediaBuySuccessResponse)
    assert result.data.media_buy_id == "mb_123"

Pattern 4: Test Error Handling

These tests verify that users get helpful error messages and can diagnose problems.

Example: Test Validation Errors

def test_invalid_json_gives_helpful_error():
    """Invalid JSON produces actionable error message."""
    from pydantic import ValidationError
    from adcp import GetProductsResponse

    invalid_json = '{"products": "not an array"}'

    with pytest.raises(ValidationError) as exc_info:
        GetProductsResponse.model_validate_json(invalid_json)

    # Error should mention the field and expected type
    error_msg = str(exc_info.value)
    assert "products" in error_msg.lower()

def test_missing_required_field_gives_helpful_error():
    """Missing required fields produce clear error messages."""
    from pydantic import ValidationError
    from adcp import Product

    incomplete_data = {
        "product_id": "test",
        "name": "Test Product",
        # Missing: description, publisher_properties, pricing_options
    }

    with pytest.raises(ValidationError) as exc_info:
        Product.model_validate(incomplete_data)

    error_msg = str(exc_info.value)
    # Should tell user what's missing
    assert "field required" in error_msg.lower() or "missing" in error_msg.lower()

Recommended Fixtures

Mock Client Fixture

@pytest.fixture
def mock_adcp_client(mocker):
    """Create a mock ADCPClient for testing.

    Returns a client with mocked adapter so tests can control responses.
    """
    from adcp import ADCPClient, AgentConfig, Protocol

    config = AgentConfig(
        id="test_agent",
        agent_uri="https://test.example.com",
        protocol=Protocol.A2A
    )

    client = ADCPClient(config)

    # Mock the adapter to avoid real network calls
    client.adapter = mocker.MagicMock()

    return client

Sample Data Fixture

@pytest.fixture
def sample_product_json():
    """Realistic product JSON from protocol.

    Use this in tests that need valid product data.
    """
    return {
        "product_id": "premium_display",
        "name": "Premium Display Ad",
        "description": "High-visibility homepage placement",
        "publisher_properties": [
            {
                "publisher_domain": "example.com",
                "selection_type": "by_id",
                "property_ids": ["homepage", "mobile_app"],
            }
        ],
        "pricing_options": [
            {
                "pricing_model": "cpm",
                "pricing_option_id": "po-premium-1",
                "currency": "USD",
                "fixed_price": 5.50,
            }
        ],
    }

In-Process MCP Fixtures

For compliance fleets and integration tests that need a full ADCPClient exercising the real protocol path against an in-process server (no loopback HTTP), wire an InMemoryTransport pair and pass the connected session to ADCPClient.from_mcp_client():

import contextlib

import pytest
from mcp import ClientSession
from mcp.shared.memory import create_client_server_memory_streams

from adcp import ADCPClient


@pytest.fixture
async def in_process_client(my_mcp_server):
    """ADCPClient backed by an in-process MCP transport.

    Caller owns the session lifecycle — `close()` and `async with` exit
    on the returned client are no-ops.
    """
    async with contextlib.AsyncExitStack() as stack:
        (c_read, c_write), (s_read, s_write) = await stack.enter_async_context(
            create_client_server_memory_streams()
        )
        # wire your in-process server to (s_read, s_write) here
        my_mcp_server.connect(s_read, s_write)

        session = await stack.enter_async_context(ClientSession(c_read, c_write))
        await session.initialize()

        yield ADCPClient.from_mcp_client(session, agent_id="fixture")

Why this matters: a loopback HTTP server adds a port-allocation race per test, dies under high parallelism, and obscures bugs that only surface when a real protocol path is exercised. The factory bypasses that without giving up any of the client surface (validation hooks, idempotency, the capability cache).

For parity, see JS AgentClient.fromMCPClient() (v5.19.0).

Anti-Patterns to Avoid

❌ Don't Import from Internal Modules

# ❌ WRONG: Couples tests to a sort-order winner and a renumbered name
from adcp.types._generated import Product1
from adcp.types.domains.core.product import PublisherProperties4

# ✅ CORRECT: Use public API
from adcp import Product

# Test using JSON (wire format)
product_json = {
    "product_id": "test",
    "name": "Test",
    "description": "Test product",
    "publisher_properties": [{
        "publisher_domain": "example.com",
        "selection_type": "by_id",
        "property_ids": ["site1"],
    }],
    "pricing_options": [
        {
            "pricing_model": "cpm",
            "pricing_option_id": "po-test-1",
            "currency": "USD",
            "fixed_price": 5.0,
        }
    ],
}

product = Product.model_validate(product_json)
assert product.publisher_properties[0].selection_type == "by_id"

❌ Don't Test Pydantic Mechanics

# ❌ WRONG: Testing Pydantic's discriminator implementation
def test_discriminator_field_is_enforced():
    # Pydantic already tests this extensively
    ...

# ✅ CORRECT: Test user-facing behavior
def test_user_can_deserialize_success_and_error_responses():
    from adcp import CreateMediaBuySuccessResponse, CreateMediaBuyErrorResponse

    success_json = '{"media_buy_id": "mb_123", "buyer_ref": "ref", "packages": []}'
    error_json = '{"errors": [{"code": "err", "message": "msg"}]}'

    success = CreateMediaBuySuccessResponse.model_validate_json(success_json)
    error = CreateMediaBuyErrorResponse.model_validate_json(error_json)

    # User can distinguish by type
    assert isinstance(success, CreateMediaBuySuccessResponse)
    assert isinstance(error, CreateMediaBuyErrorResponse)

❌ Don't Test Type Identity

# ❌ WRONG: Testing internal implementation detail
def test_alias_points_to_generated_type():
    assert CreateMediaBuySuccessResponse is CreateMediaBuyResponse1

# ✅ CORRECT: Test that alias works in actual usage
def test_semantic_alias_works_for_users():
    from adcp import CreateMediaBuySuccessResponse

    response = CreateMediaBuySuccessResponse(
        media_buy_id="mb_123",
        buyer_ref="ref",
        packages=[]
    )

    # Can serialize to JSON
    json_str = response.model_dump_json()
    assert "media_buy_id" in json_str

    # Can deserialize from JSON
    roundtrip = CreateMediaBuySuccessResponse.model_validate_json(json_str)
    assert roundtrip.media_buy_id == response.media_buy_id

Summary: Key Principles

✅ DO

  1. Test public API (from adcp import X)
  2. Test wire format with JSON (model_validate_json)
  3. Test user workflows (can buyer discover products?)
  4. Test behavior (does API work as documented?)
  5. Use semantic aliases (CreateMediaBuySuccessResponse)
  6. Write tests users can learn from

❌ DON'T

  1. Import from _generated or an internal numbered class
  2. Test Pydantic internals
  3. Test type identity (assert X is Y)
  4. Test implementation details
  5. Use numbered types (CreateMediaBuyResponse1)
  6. Test mechanics instead of behavior

Remember

  • Tests should demonstrate correct SDK usage
  • Tests should catch protocol compatibility bugs
  • Tests should tell user stories
  • Tests should respect public API boundaries

See Also

Freezing time with deferred models

Pydantic schemas retain real datetime types. Build them before freezegun replaces those types, and keep SDK modules on its ignore list. In tests/conftest.py, import your application models first, then configure:

import freezegun
from adcp.testing import build_all_models

freezegun.configure(extend_ignore_list=["adcp"])
build_all_models()

The helper resolves advertised lazy exports on adcp, adcp.types, and other imported SDK facades, recursively builds their imported model descendants (including adopter subclasses), and warms registered SDK adapters. Repeated calls reuse completed schemas. It imports no arbitrary application modules; if you introduce new models or SDK submodules later, import them and call the helper again before freezing. Call it while real datetime classes are active. Normal SDK imports and strict model construction keep their existing behavior.