Test Structured Responses

Why Testing Structured Output Is Different

Structured outputs make responses predictable in shape, but the content still varies between runs and between model versions. That changes how tests should work:

The first three kinds are deterministic and free, and should run on every change. The last one calls the API, costs money, and produces results that vary slightly, so it runs on demand or on a schedule, and checks a score, not exact text. This follows the evaluation approach from Lesson 39.

Layer 1: Contract Tests

A contract test checks what your application promises about the schema: the request your code will send. For example, the schema sent to the API must forbid extra fields, and must list every field as required, so the output always has the same keys:

def test_schema_sent_to_api_is_closed_and_complete():
    schema = transform_schema(EventInfo)
    assert schema["additionalProperties"] is False
    assert set(schema["required"]) == set(EventInfo.model_fields)

Contract tests catch accidental changes, such as a field that becomes optional during a refactor, before they change the data your pipeline stores.

Layer 2: Validation Tests

Business rules should be tested with specific inputs and expected messages. Parametrized tests make each rule visible:

@pytest.mark.parametrize("changes, expected", [
    ({"date": "18/11/2026"}, "YYYY-MM-DD"),
    ({"fee_inr": -5}, "negative"),
    ({"registration_url": "https://other.example"}, "does not appear"),
])
def test_bad_events_are_reported(changes, expected):
    ...

Check both that a problem is reported and that the message says what is wrong. A rule that fires with an unhelpful message is hard to fix later.

Layer 3: Fake Clients for Response Handling

Your code must handle every stop reason and error (Lessons 43, 57, and 77). These paths are rare in practice, so they need deliberate tests. A fake client returns scripted responses without any network access (Lesson 43):

class FakeMessages:
    def __init__(self, outcomes):
        self.outcomes = list(outcomes)

    async def parse(self, **request):
        outcome = self.outcomes.pop(0)
        if isinstance(outcome, Exception):
            raise outcome
        return outcome


class FakeClient:
    def __init__(self, *outcomes):
        self.messages = FakeMessages(outcomes)

Each outcome is either a response object or an exception, so a test can simulate a successful answer, a truncated answer, a refusal, or a connection failure. The fake's parse() accepts the same keyword arguments as the real method, so code under test runs unchanged.

Layer 4: Golden-Set Evaluation

A golden set is a small collection of inputs with known correct outputs, written by people, and checked by code. For extraction, scoring is usually field-level accuracy: the share of expected fields that the model extracted correctly.

def normalize(value) -> str | None:
    return None if value is None else str(value).strip().lower()

Comparing normalized values avoids failures from capitalization and spacing. The test then asserts a threshold, not an exact match:

assert accuracy >= 0.9

Thresholds reflect the reality that a model may vary slightly, and that a few borderline examples can be wrong without the whole system being broken. Choose the threshold from measurements, and raise it as the system improves.

Organizing the Test Suite

event-extraction/
├── events_extractor.py
├── pipeline.py
└── tests/
    ├── test_events_extractor.py   # Layers 1–3, offline
    ├── test_pipeline.py           # Storage behavior, offline
    └── test_golden_live.py        # Layer 4, live, on demand

Install pytest with python -m pip install pytest. The offline tests need nothing else, because httpx and pydantic are already installed with the SDK.

Practical Example: Offline and Live Tests

tests/test_events_extractor.py

"""Offline tests for events_extractor.py. No API calls: responses are simulated."""

import asyncio
from types import SimpleNamespace

import anthropic
import httpx
import pytest
from anthropic import transform_schema
from pydantic import ValidationError

from events_extractor import EventInfo, check_event, extract_event

ANNOUNCEMENT = ("Free webinar: Intro to APIs on 2026-11-18 at 7 PM IST. "
                "Register at https://gocourse.example/api-webinar")
GOOD_EVENT = EventInfo(title="Intro to APIs", date="2026-11-18", venue=None, fee_inr=0,
                       registration_url="https://gocourse.example/api-webinar")


class FakeMessages:
    def __init__(self, outcomes):
        self.outcomes = list(outcomes)

    async def parse(self, **request):
        outcome = self.outcomes.pop(0)
        if isinstance(outcome, Exception):
            raise outcome
        return outcome


class FakeClient:
    def __init__(self, *outcomes):
        self.messages = FakeMessages(outcomes)


def fake_response(stop_reason, event=None):
    return SimpleNamespace(stop_reason=stop_reason, parsed_output=event, id="msg_test")


# Layer 1: contract
def test_schema_sent_to_api_is_closed_and_complete():
    schema = transform_schema(EventInfo)
    assert schema["additionalProperties"] is False
    assert set(schema["required"]) == set(EventInfo.model_fields)


# Layer 2: validation rules
def test_consistent_event_has_no_problems():
    assert check_event(GOOD_EVENT, ANNOUNCEMENT) == []


@pytest.mark.parametrize("changes, expected", [
    ({"date": "18/11/2026"}, "YYYY-MM-DD"),
    ({"fee_inr": -5}, "negative"),
    ({"registration_url": "https://other.example"}, "does not appear"),
    ({"title": None, "date": None, "venue": None}, "no event details"),
])
def test_bad_events_are_reported(changes, expected):
    event = GOOD_EVENT.model_copy(update=changes)
    assert any(expected in problem for problem in check_event(event, ANNOUNCEMENT))


# Layer 3: response handling
def test_valid_response_is_ok():
    result = asyncio.run(extract_event(FakeClient(fake_response("end_turn", GOOD_EVENT)), "ann-1", ANNOUNCEMENT))
    assert result.status == "ok" and result.event == GOOD_EVENT


def test_truncated_response_is_not_stored_as_an_event():
    result = asyncio.run(extract_event(FakeClient(fake_response("max_tokens")), "ann-2", ANNOUNCEMENT))
    assert result.status == "max_tokens" and result.event is None


def test_refusal_is_reported():
    result = asyncio.run(extract_event(FakeClient(fake_response("refusal")), "ann-3", ANNOUNCEMENT))
    assert result.status == "refusal"


def test_api_error_is_reported_not_raised():
    request = httpx.Request("POST", "https://api.anthropic.com/v1/messages")
    error = anthropic.APIConnectionError(request=request)
    result = asyncio.run(extract_event(FakeClient(error), "ann-4", ANNOUNCEMENT))
    assert result.status == "error"


def test_validation_failure_is_reported():
    try:
        EventInfo.model_validate({"title": 42})          # Missing fields and a wrong type
    except ValidationError as error:
        failure = error                                   # Keep a reference: Python deletes `error` after this block
    result = asyncio.run(extract_event(FakeClient(failure), "ann-5", ANNOUNCEMENT))
    assert result.status == "invalid"

tests/test_pipeline.py

"""Offline tests for the storage rules in pipeline.py."""

from events_extractor import EventInfo, Extraction
import pipeline


def test_finished_rows_are_skipped(tmp_path):
    db = pipeline.open_db(str(tmp_path / "test.db"))
    event = EventInfo(title="Webinar", date="2026-11-18", venue=None, fee_inr=0, registration_url=None)
    pipeline.save(db, Extraction("ann-1", "ok", event))
    db.commit()
    assert pipeline.already_done(db, "ann-1")
    assert not pipeline.already_done(db, "ann-2")


def test_failed_rows_are_retried(tmp_path):
    db = pipeline.open_db(str(tmp_path / "test.db"))
    pipeline.save(db, Extraction("ann-3", "invalid", problems=["no event details found"]))
    db.commit()
    assert not pipeline.already_done(db, "ann-3")

tests/test_golden_live.py

"""Live evaluation against a golden set. Calls the API, so it runs only when requested."""

import asyncio
import os

import pytest
from anthropic import AsyncAnthropic

from events_extractor import extract_event

GOLDEN = [
    ("Join the Django Weekend Workshop on 2026-11-21 at Anna Hall, Chennai. Fee Rs. 1500. "
     "Sign up at https://gocourse.example/django",
     {"title": "Django Weekend Workshop", "date": "2026-11-21", "venue": "Anna Hall, Chennai",
      "fee_inr": 1500, "registration_url": "https://gocourse.example/django"}),
    ("SQL Bootcamp starts 2026-12-05 at the Online Studio. Registration link not yet available.",
     {"title": "SQL Bootcamp", "date": "2026-12-05", "venue": "Online Studio",
      "fee_inr": None, "registration_url": None}),
]


def normalize(value) -> str | None:
    return None if value is None else str(value).strip().lower()


@pytest.mark.skipif(not os.getenv("RUN_LIVE_TESTS"), reason="calls the live API; set RUN_LIVE_TESTS=1")
def test_golden_set_field_accuracy():
    async def extract_all():
        async with AsyncAnthropic() as client:
            return await asyncio.gather(
                *(extract_event(client, str(i), text) for i, (text, _) in enumerate(GOLDEN)))

    results = asyncio.run(extract_all())
    correct = total = 0
    for result, (_, expected) in zip(results, GOLDEN):
        actual = result.event.model_dump() if result.event else {}
        for name, value in expected.items():
            total += 1
            correct += normalize(actual.get(name)) == normalize(value)

    accuracy = correct / total
    print(f"field accuracy: {accuracy:.0%}")
    assert accuracy >= 0.9

Run the offline tests:

python -m pytest -q tests/test_events_extractor.py tests/test_pipeline.py

13 passed in 0.31s

Run the live evaluation when you want it:

RUN_LIVE_TESTS=1 python -m pytest -q tests/test_golden_live.py -s

field accuracy: 90%
1 passed in 4.12s

The live result varies from run to run. A score such as 90% passes the threshold, and a drop to 80% would fail it, which signals that a prompt or model change needs attention.

How It Works

1. The contract test checks the schema that would be sent to the API, without calling it. If a refactor makes a field optional, this test fails before the pipeline stores data with missing keys.
2. The validation tests use parametrization, so each rule has a named input and an expected message. The assertion checks that the message contains the expected text, so the rule's wording is part of what is tested.
3. FakeClient replays scripted outcomes. Each response-handling path is covered by one test: success, truncation, refusal, connection error, and validation error. These are the paths that are hardest to trigger on demand.
4. The validation-error test creates a real ValidationError by validating bad data, and stores it in a variable inside the except block. The fake then raises it from parse(), exactly where the real SDK would. The variable is assigned inside the block because Python deletes the except ... as name when the block ends.
5. The pipeline tests use tmp_path, a pytest fixture that gives each test its own temporary directory, so tests never touch your real database.
6. The golden test is skipped unless RUN_LIVE_TESTS is set, so routine test runs cost nothing. It scores each expected field after normalization, and asserts a threshold. The expected values are written by a person, not produced by the model.
7. Only the live test calls the network, which keeps the rest of the suite fast, reliable, and free to run on every change.

Note: a golden set is only as good as its labels. Write the expected values from the source text, not from the model's output, and include some hard cases: missing values, unusual dates, and ambiguous wording. Grow the set whenever a real failure appears.

Common Mistakes

Testing Only the Happy Path

What happens: the error handling, which runs in production when things go wrong, is never exercised.

Fix: write a test for every stop reason, error, and validation path.

Exact-Match Assertions on Model Output

What happens: tests fail when wording changes, even though the extracted data is correct.

Fix: score fields after normalization, and assert a threshold.

Calling the API in Routine Tests

What happens: slow, expensive, flaky test runs, and failures unrelated to code changes.

Fix: keep live evaluation separate and opt-in.

Golden Sets Written from Model Output

What happens: the test confirms the model's current behavior, including its mistakes.

Fix: write expected values from the source, by hand.

Testing Schemas Without Testing Rules

What happens: the schema is correct, but a business rule silently stops working.

Fix: test every rule with inputs that should pass and inputs that should fail.

Best Practices

• Separate offline tests from live evaluation.
• Test every stop reason and error path with a fake client.
• Test each validation rule with named inputs and expected messages.
• Guard the schema with a contract test.
• Build a golden set from real examples, with hand-written labels.
• Assert thresholds for model-dependent results.
• Rerun the live evaluation after changes to prompts, schemas, or models (Lesson 39).

Exercise

1. Add a golden example with a registration link that appears in the text, but with a slightly different capitalization. Does the normalized comparison match?
2. Add a fake client outcome that returns stop_reason == "end_turn" with parsed_output set to None. Write a test for it, and fix extract_event() so the result is invalid, not a stop reason.
3. Add a test for pipeline.save() that stores an event and reads back the JSON in data, checking that fee_inr is preserved.
4. Explain why the live test prints the accuracy with print() and runs with -s, and why the threshold is 0.9 rather than 1.0.

Step 2: the test and the fix:

def test_missing_parsed_output_is_invalid():
    result = asyncio.run(extract_event(FakeClient(fake_response("end_turn", None)), "ann-6", ANNOUNCEMENT))
    assert result.status == "invalid"

In extract_event(), handle this case before the general check:

if response.stop_reason == "end_turn" and response.parsed_output is None:
    return Extraction(doc_id, "invalid", problems=["no structured output returned"],
                      message_id=response.id)

Without the fix, the status would be the stop reason itself, end_turn, which is not one of the statuses the pipeline counts. Writing the test first is what reveals the gap.

Step 4: print() with -s shows the score in the terminal, so a person can see how the system is doing even when the test passes. A threshold of 0.9 instead of 1.0 reflects the nature of model output: a small number of borderline or ambiguous examples can be wrong without the system being broken. An exact requirement of 100% would make the test fail on normal variation. The threshold should be set from measured results, and raised as the system improves.