Python LLM application testing with strict contracts, recording fakes, controlled HTTP responses, and blocked external actions

Technical Skills

Testing LLM Applications in Python with pytest: Contracts, Fakes, and Failure Injection

Test Python LLM applications with pytest, provider fakes, adapter contracts, and failure injection for invalid outputs, timeouts, retries, and blocked actions.

Sep 14, 202625 min readMuhammad FarooqLast reviewed: Sep 14, 2026

An LLM can return an uncertain answer while the application around it still has exact obligations: reject invalid requests, validate proposals, enforce permissions, bound retries, and avoid repeating an uncertain external write. This tutorial builds one small Python ticket-action application and tests those obligations with pytest. Recording fakes expose application calls; an HTTPX transport supplies controlled HTTP responses. Each injected failure has an observable assertion, including the absence of an action that must never run.

Define what each test layer can establish

Start with an assertion whose truth the application controls. Given an invalid ticket identifier, the provider receives zero calls. Given a forbidden action, the executor receives zero calls. These are stronger and more precise local assertions than asking whether a real model always returns a helpful answer.

The broader Python skills guide introduces typed boundaries and basic extraction tests. Here we extend that testing seam into controlled protocol failures, exact retry budgets, authorization assertions, and cancellation states.

Four layers answer different questions; the runnable suite covers the first two.
Layer	Evidence it can provide	Evidence it cannot provide
Unit / application	With a scripted provider and executor, validation, routing, permission checks, call arguments, and terminal states match the assertions.	A real provider accepts our requests or a real ticket system behaves like the fake.
Adapter / local contract	Our HTTP adapter sends the intended request and translates controlled responses and transport errors.	The external service implements this contract; no consumer-provider verification exchange occurs here.
Integration	A configured adapter interoperates with a selected real service or sandbox under the exercised conditions.	Every outage, deployment configuration, race, or possible model answer behaves correctly.
Model / evaluation	A defined dataset and grading method provide evidence about semantic task quality and failure slices.	A denied executor call is absent unless the evaluation also observes and checks that boundary.

Turn the ticket workflow into a contract matrix

The application asks a provider to propose closing or escalating one ticket. Trusted caller context determines which ticket and action may be used. The proposal is data, never authority. Only a validated, authorized proposal can reach the executor; only an acknowledged execution can produce success.

Each invocation owns a fresh in-memory trace. It is an observation surface for these tests, not a database, a durable job record, or a lock. The executor is an interface with a recording fake; the HTTP provider adapter is concrete and runs through HTTPX even when the transport is controlled.

Ticket request flows through strict validation, a provider adapter, structured validation, trusted policy, and an external executor. Pre-dispatch failures block execution; post-dispatch timeout or cancellation leaves an ambiguous outcome. Generation retry timing and semantic evaluation are separate boundaries. — Tests observe both the returned state and calls across boundaries. Crossing the executor boundary changes what an error can tell us about a remote write.

Injected failures and the evidence to inspect.
Boundary / injection	Required observation
Invalid request	Controlled invalid-request error; zero provider calls and zero executor calls.
Truncated or malformed JSON	Output rejected; no success receipt and no executor call.
Parseable but invalid proposal	Wrong types, unexpected fields, or invalid domain values rejected before execution.
Provider refusal or error envelope	Explicit provider failure; no accidental success.
Forbidden resource or action	Executor recording remains empty.
HTTP 401 / 403	One attempt and no requested retry delay.
HTTP 429 / 503 and selected timeouts	Only the generation policy may retry; assert attempts and requested delays.
Retry exhaustion	Exact configured attempt count, deterministic terminal error, and no executor call.
Write applied before lost acknowledgement	One executor call, ambiguous state, no success receipt, and no automatic repeat.
Cancellation before dispatch	Cancellation propagates, executor remains untouched, owned resources close.
Cancellation after dispatch	Cancellation propagates with ambiguous state because remote completion is unknown.
Sensitive synthetic fixture	Sentinel absent from the inspected application logs and public error text.

Build the strict application and its effect boundary

Save all five Python files and requirements.txt below in one empty directory. The modules form one runnable example: application.py owns domain behavior; http_adapter.py implements the provider protocol; fakes.py supplies test doubles; the two test files exercise those same modules. No API credentials or web framework are required.

The models use Pydantic strict validation. A string ticket ID is not silently converted to an integer; booleans are rejected as IDs, unknown fields are forbidden, and blank instructions fail. JSON is parsed first and then validated as Python data. Strictness is type-dependent: Pydantic documents different allowances for some types when validating directly from JSON.

There are three distinct checks on a proposal. json.loads establishes syntax. ProposedAction checks structure, types, allowed action values, positive identifiers, and a nonblank bounded reason. The application then requires the proposed ticket to match the requested ticket. None of these checks establishes that the issue is actually resolved; that is a semantic question.

Access must come from authenticated, trusted application code. This tutorial does not implement authentication or permission lookup. It authorizes the proposed external action before dispatch; a real system should also authorize access to ticket data before loading or sending that data to a provider. The private_note fixture is deliberately excluded from the concrete adapter request.

Provider implementations return a JSON string or a controlled AppError. Executor implementations return an acknowledgement string or raise. An exception after dispatch is conservatively ambiguous. This compact example does not classify every possible remote error or implement the real ticket-service adapter. A fresh Trace is required for every invocation; do not share it between concurrent runs.

The application owns strict validation, trusted policy checks, state, and error translation. It never retries an executor call.application.py

import asyncio
import json
import logging
from dataclasses import dataclass
from typing import Literal, Protocol

from pydantic import BaseModel, ConfigDict, Field, ValidationError

logger = logging.getLogger("ticket_app")
Action = Literal["close", "escalate"]
Code = Literal[
    "invalid_request", "invalid_output", "forbidden", "provider_auth",
    "provider_refused", "provider_error", "provider_protocol",
    "provider_exhausted", "provider_deferred", "effect_unknown",
]


class AppError(Exception):
    def __init__(self, code: Code):
        self.code = code
        super().__init__(code)


class StrictModel(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid", frozen=True)


class TicketRequest(StrictModel):
    ticket_id: int = Field(gt=0)
    instruction: str = Field(min_length=1, max_length=500, pattern=r"\S")
    private_note: str = Field(default="", max_length=500, repr=False)


class ProposedAction(StrictModel):
    ticket_id: int = Field(gt=0)
    action: Action
    reason: str = Field(min_length=1, max_length=120, pattern=r"\S")


@dataclass(frozen=True)
class Access:
    # Supplied by trusted application code, never by the model response.
    ticket_ids: frozenset[int]
    actions: frozenset[Action]


@dataclass
class Trace:
    # One fresh instance per invocation; this is not a durable job record.
    phase: Literal[
        "validating", "generating", "authorizing", "dispatching",
        "succeeded", "failed", "cancelled", "ambiguous",
    ] = "validating"
    receipt: str | None = None


class Provider(Protocol):
    async def propose(self, request: TicketRequest) -> str: ...


class Executor(Protocol):
    async def execute(self, proposal: ProposedAction) -> str: ...


async def run(
    raw_request: dict[str, object], access: Access,
    provider: Provider, executor: Executor, trace: Trace,
) -> str:
    try:
        try:
            request = TicketRequest.model_validate(raw_request)
        except ValidationError:
            raise AppError("invalid_request") from None

        trace.phase = "generating"
        raw_output = await provider.propose(request)
        try:
            proposal = ProposedAction.model_validate(json.loads(raw_output))
        except (ValueError, ValidationError):
            raise AppError("invalid_output") from None
        if proposal.ticket_id != request.ticket_id:
            raise AppError("invalid_output")

        trace.phase = "authorizing"
        if (proposal.ticket_id not in access.ticket_ids
                or proposal.action not in access.actions):
            raise AppError("forbidden")

        trace.phase = "dispatching"
        try:
            receipt = await executor.execute(proposal)
        except Exception:
            # Dispatch may have reached the remote service. Never retry here.
            trace.phase = "ambiguous"
            raise AppError("effect_unknown") from None
        trace.receipt = receipt
        trace.phase = "succeeded"
        return receipt
    except asyncio.CancelledError:
        trace.phase = "ambiguous" if trace.phase == "dispatching" else "cancelled"
        raise
    except AppError as error:
        if trace.phase != "ambiguous":
            trace.phase = "failed"
        logger.warning("application_failed code=%s", error.code)
        raise

Implement an HTTP adapter with a documented tutorial contract

HttpProvider owns an async HTTPX client and must be used as an async context manager. Async I/O gives cancellation a real boundary to interrupt; the domain workflow itself stays sequential. The /propose endpoint and its payload are invented for this tutorial. They are not an OpenAI, Anthropic, or other vendor API contract.

POST /propose sends only ticket_id and instruction. A 200 response must contain exactly status and content: status is ok, refused, or error; content is a string. For ok, content contains the proposed action as JSON text. A refusal is provider_refused, an error envelope is provider_error, and malformed envelopes are provider_protocol. A valid envelope containing broken proposal JSON reaches the application and becomes invalid_output.

The fictional endpoint only generates a proposal: it does not mutate tickets, invoke tools, or bill for requests. Its contract permits repeated generation. That explicit assumption justifies the limited retry policy below. A real paid, tool-enabled, or stateful generation endpoint needs its own documented retry and cost policy.

The local contract tests use HTTPX mock and custom transports to run real request construction and response translation with controlled I/O. This tests our adapter against our fixtures. Independent provider conformance remains an integration task.

An owned HTTPX client, strict tutorial envelope, controlled error codes, and a generation-only retry budget.http_adapter.py

import asyncio
from collections.abc import Awaitable, Callable
from types import TracebackType
from typing import Literal, Self

import httpx
from pydantic import ValidationError

from application import AppError, StrictModel, TicketRequest


class Envelope(StrictModel):
    # Tutorial-owned wire format; this is not a vendor API specification.
    status: Literal["ok", "refused", "error"]
    content: str


class HttpProvider:
    def __init__(
        self, *, transport: httpx.AsyncBaseTransport | None = None,
        sleeper: Callable[[float], Awaitable[None]] = asyncio.sleep,
    ):
        self.client = httpx.AsyncClient(
            base_url="https://provider.example", transport=transport,
            timeout=2.0, follow_redirects=False, trust_env=False,
        )
        self.sleep = sleeper
        self.delays = (0.25, 0.5)  # Two retries, three total attempts.

    async def __aenter__(self) -> Self:
        await self.client.__aenter__()
        return self

    async def __aexit__(
        self, exc_type: type[BaseException] | None,
        exc: BaseException | None, tb: TracebackType | None,
    ) -> None:
        await self.client.__aexit__(exc_type, exc, tb)

    def retry_delay(self, response: httpx.Response, fallback: float) -> float:
        value = response.headers.get("Retry-After")
        if value is None:
            return fallback
        # Support bounded delta-seconds. Defer dates/large/invalid values;
        # never shorten a server's requested wait to fit our local budget.
        if not value.isascii() or not value.isdecimal() or len(value) > 2:
            raise AppError("provider_deferred")
        delay = float(value)
        if delay > 2.0:
            raise AppError("provider_deferred")
        return max(fallback, delay)

    async def propose(self, request: TicketRequest) -> str:
        # /propose generates text only: no ticket mutation, tool execution,
        # or billable operation in THIS fictional contract. Repetition is
        # permitted by that contract, not merely because a status is 429/503.
        for attempt in range(len(self.delays) + 1):
            response = None
            try:
                response = await self.client.post("/propose", json={
                    "ticket_id": request.ticket_id,
                    "instruction": request.instruction,
                })
            except (httpx.ConnectTimeout, httpx.ReadTimeout):
                pass
            except httpx.RequestError:
                raise AppError("provider_error") from None
            if response is not None and response.status_code not in (429, 503):
                if response.status_code in (401, 403):
                    raise AppError("provider_auth")
                if response.status_code != 200:
                    raise AppError("provider_error")
                try:
                    envelope = Envelope.model_validate(response.json())
                except (ValueError, ValidationError):
                    raise AppError("provider_protocol") from None
                if envelope.status == "refused":
                    raise AppError("provider_refused")
                if envelope.status == "error":
                    raise AppError("provider_error")
                return envelope.content
            if attempt == len(self.delays):
                raise AppError("provider_exhausted")
            delay = self.delays[attempt]
            if response is not None:
                delay = self.retry_delay(response, delay)
            await self.sleep(delay)
        raise AssertionError("unreachable")

Tutorial-owned HTTP contract; no live endpoint implements it as part of this example.
Response / exception	Adapter policy
200 with ok envelope	Return content for strict proposal validation.
200 refused / error / malformed envelope	Terminal provider_refused / provider_error / provider_protocol.
401 or 403	Terminal provider_auth; no credential refresh or retry.
429 or 503	Up to three total attempts, subject to Retry-After handling.
ConnectTimeout or ReadTimeout	Same bounded generation retry policy.
Other HTTP status or HTTPX RequestError	Terminal provider_error; the suite includes 500, WriteTimeout, and ConnectError.

Use recording fakes that expose calls and scripted failures

ProviderFake consumes a sequence of supplied strings or exceptions and records each typed request. An unexpected extra call exhausts the script loudly. ExecutorFake records both entry and a simulated committed write; it can then lose the acknowledgement or wait for cancellation. Keeping calls and writes separate makes the lost-receipt test observable.

SleeperFake records requested delays and returns immediately. It does not advance a simulated clock or model concurrent scheduling. That is enough to inspect this retry policy because the adapter uses an attempt budget and delay values, not elapsed-time calculations.

These fakes verify what the application does when its dependencies behave as scripted. They do not independently verify a real provider, a transaction boundary, or a remote write. The fake write list is a test observation, not a deduplication store.

For a legacy dependency, unittest.mock autospeccing can catch incorrect callable signatures. Here the small explicit fakes make the contract easier to inspect. Autospeccing would still not prove remote behavior.

Recording doubles show received arguments, scripted provider failures, simulated writes, cleanup, and requested sleep durations.fakes.py

import asyncio
from collections import deque

from application import ProposedAction, TicketRequest


class ProviderFake:
    def __init__(self, *outcomes: str | Exception):
        self.outcomes = deque(outcomes)
        self.calls: list[TicketRequest] = []

    async def propose(self, request: TicketRequest) -> str:
        self.calls.append(request)
        if not self.outcomes:
            raise AssertionError("provider script exhausted")
        outcome = self.outcomes.popleft()
        if isinstance(outcome, Exception):
            raise outcome
        return outcome


class ExecutorFake:
    def __init__(self, *, fail_after_write: bool = False, block: bool = False):
        self.calls: list[ProposedAction] = []
        self.writes: list[ProposedAction] = []
        self.fail_after_write = fail_after_write
        self.block = block
        self.entered = asyncio.Event()
        self.cleaned = False

    async def execute(self, proposal: ProposedAction) -> str:
        self.calls.append(proposal)
        self.writes.append(proposal)  # Simulated remote commit.
        self.entered.set()
        try:
            if self.block:
                await asyncio.Event().wait()
            if self.fail_after_write:
                raise TimeoutError("remote receipt was lost")
            return "receipt-001"
        finally:
            self.cleaned = True


class SleeperFake:
    def __init__(self):
        self.calls: list[float] = []

    async def __call__(self, seconds: float) -> None:
        self.calls.append(seconds)

Assert strict contracts and the absence of prohibited effects

Start with the successful path: inspect the provider request, the executor proposal, and the acknowledged receipt. Then vary one boundary at a time. Invalid requests use an empty provider script, so an accidental call cannot quietly succeed. Output cases include truncated JSON, an array, a coerced-looking identifier, a mismatched resource, an unsupported action, an empty reason, and an extra authorized field.

The pytest parametrization decorator runs each supplied case as a separate test instance. Each row shares an invariant: failure must leave the executor recording empty and the trace without a success receipt. This makes adding a new failure case inexpensive without reducing the assertion to a generic exception check.

The authorization test uses a structurally valid close proposal. One trusted context lacks ticket 7; the other permits only escalation. Both must raise forbidden with executor.calls == executor.writes == []. Checking only the error text could miss an implementation that wrote first and rejected afterward.

The final tests in this file inspect two harder contracts. A simulated remote commit followed by a timeout leaves one write and no receipt. Cancellation after that boundary also leaves ambiguity. The sensitive-data test puts a synthetic private sentinel into invalid output and inspects both application logging and the formatted public exception.

Application tests inspect arguments, strict rejection, policy denials, ambiguous writes, cancellation, and bounded diagnostic redaction.test_application.py

import asyncio
import json
import logging
import traceback

import pytest

from application import Access, AppError, Trace, run
from fakes import ExecutorFake, ProviderFake

REQUEST = {"ticket_id": 7, "instruction": "Close the resolved ticket."}
PROPOSAL = {"ticket_id": 7, "action": "close", "reason": "Issue resolved"}
ALLOWED = Access(frozenset({7}), frozenset({"close", "escalate"}))


def test_success_records_arguments_and_receipt():
    provider = ProviderFake(json.dumps(PROPOSAL))
    executor, trace = ExecutorFake(), Trace()
    assert asyncio.run(run(REQUEST, ALLOWED, provider, executor, trace)) == "receipt-001"
    assert len(provider.calls) == len(executor.calls) == 1
    assert provider.calls[0].model_dump(exclude={"private_note"}) == REQUEST
    assert executor.calls[0].model_dump() == PROPOSAL
    assert trace.phase == "succeeded" and trace.receipt == "receipt-001"


@pytest.mark.parametrize("change", [
    {"ticket_id": "7"}, {"ticket_id": True}, {"ticket_id": 0},
    {"instruction": "   "}, {"unexpected": "field"},
])
def test_invalid_request_never_calls_provider(change):
    provider, executor, trace = ProviderFake(), ExecutorFake(), Trace()
    with pytest.raises(AppError, match="^invalid_request$"):
        asyncio.run(run(REQUEST | change, ALLOWED, provider, executor, trace))
    assert provider.calls == executor.calls == []
    assert trace.phase == "failed" and trace.receipt is None


@pytest.mark.parametrize("raw_output", [
    '{"ticket_id":', "not JSON", "[]",
    json.dumps(PROPOSAL | {"ticket_id": "7"}),
    json.dumps(PROPOSAL | {"ticket_id": 0}),
    json.dumps(PROPOSAL | {"ticket_id": 8}),
    json.dumps(PROPOSAL | {"action": "delete"}),
    json.dumps(PROPOSAL | {"reason": ""}),
    json.dumps(PROPOSAL | {"authorized": True}),
])
def test_invalid_output_never_executes(raw_output):
    provider, executor, trace = ProviderFake(raw_output), ExecutorFake(), Trace()
    with pytest.raises(AppError, match="^invalid_output$"):
        asyncio.run(run(REQUEST, ALLOWED, provider, executor, trace))
    assert len(provider.calls) == 1 and executor.calls == []
    assert trace.phase == "failed" and trace.receipt is None


@pytest.mark.parametrize("code", ["provider_refused", "provider_error"])
def test_provider_failure_is_not_success(code):
    provider, executor, trace = ProviderFake(AppError(code)), ExecutorFake(), Trace()
    with pytest.raises(AppError) as error:
        asyncio.run(run(REQUEST, ALLOWED, provider, executor, trace))
    assert error.value.code == code
    assert len(provider.calls) == 1 and executor.calls == []
    assert trace.phase == "failed" and trace.receipt is None


@pytest.mark.parametrize("access", [
    Access(frozenset({8}), frozenset({"close"})),
    Access(frozenset({7}), frozenset({"escalate"})),
])
def test_forbidden_resource_or_action_has_zero_effects(access):
    provider = ProviderFake(json.dumps(PROPOSAL))
    executor, trace = ExecutorFake(), Trace()
    with pytest.raises(AppError, match="^forbidden$"):
        asyncio.run(run(REQUEST, access, provider, executor, trace))
    assert executor.calls == executor.writes == []
    assert trace.phase == "failed" and trace.receipt is None


def test_lost_write_receipt_is_ambiguous_and_is_never_retried():
    provider = ProviderFake(json.dumps(PROPOSAL))
    executor, trace = ExecutorFake(fail_after_write=True), Trace()
    with pytest.raises(AppError, match="^effect_unknown$"):
        asyncio.run(run(REQUEST, ALLOWED, provider, executor, trace))
    assert len(executor.calls) == len(executor.writes) == 1
    assert trace.phase == "ambiguous" and trace.receipt is None
    assert executor.cleaned


def test_cancellation_after_dispatch_does_not_claim_rollback():
    async def scenario():
        executor, trace = ExecutorFake(block=True), Trace()
        task = asyncio.create_task(run(
            REQUEST, ALLOWED, ProviderFake(json.dumps(PROPOSAL)), executor, trace,
        ))
        await asyncio.wait_for(executor.entered.wait(), timeout=2)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        assert task.cancelled() and executor.cleaned
        assert len(executor.calls) == len(executor.writes) == 1
        assert trace.phase == "ambiguous" and trace.receipt is None
    asyncio.run(scenario())


def test_sensitive_invalid_output_is_not_in_application_diagnostics(caplog):
    secret = "fixture-private-customer-note-91"
    request = REQUEST | {"private_note": secret}
    provider = ProviderFake(json.dumps(PROPOSAL | {"ticket_id": secret}))
    executor, trace = ExecutorFake(), Trace()
    with caplog.at_level(logging.WARNING, logger="ticket_app"):
        with pytest.raises(AppError) as error:
            asyncio.run(run(request, ALLOWED, provider, executor, trace))
    assert error.value.code == "invalid_output" and executor.calls == []
    assert "application_failed code=invalid_output" in caplog.text
    rendered = "".join(traceback.format_exception(error.value))
    assert secret not in caplog.text + str(error.value) + rendered
    # This does not inspect debugger locals, third-party telemetry, or dumps.

Inject protocol failures through the concrete HTTP adapter

The transport handler receives a real HTTPX Request. The serialization test checks POST, the exact tutorial URL, the JSON content type, the selected payload fields, and omission of private_note. Controlled responses then exercise the adapter and application together through run; the executor remains a recording fake.

Malformed envelope JSON and malformed generated JSON belong to different boundaries and produce different error codes. Refusal and error envelopes are not treated as a successful proposal. HTTP 401, 403, and 500 each consume one attempt with no recorded delay. The positive retry cases reach the executor exactly once after a permitted generation retry.

The test module also uses pytest monkeypatch to replace the default async HTTP transport method with a failure. pytest restores the replacement after each test. This catches accidentally using that default HTTPX path in these adapter tests; it is not a process-wide network sandbox.

The shared exercise helper always exits the provider context and asserts that its client is closed. The cancellation test additionally observes transport closure. Exceptions are injected directly, so these tests verify error classification and cleanup paths, not the actual duration of HTTPX timeouts or socket behavior.

Controlled HTTP responses and exceptions exercise serialization, envelope handling, retries, exhaustion, deferred waits, and owned-resource cancellation.test_adapter.py

import asyncio
import json

import httpx
import pytest

from application import AppError, TicketRequest, Trace, run
from fakes import ExecutorFake, SleeperFake
from http_adapter import HttpProvider
from test_application import ALLOWED, PROPOSAL, REQUEST

OK = {"status": "ok", "content": json.dumps(PROPOSAL)}


@pytest.fixture(autouse=True)
def block_default_network(monkeypatch):
    async def fail(*args, **kwargs):
        raise AssertionError("unexpected default HTTP transport")
    monkeypatch.setattr(httpx.AsyncHTTPTransport, "handle_async_request", fail)


def exercise(*outcomes):
    calls, sleeper, executor, trace = [], SleeperFake(), ExecutorFake(), Trace()

    def handler(request):
        calls.append(request)
        outcome = outcomes[len(calls) - 1]
        if isinstance(outcome, Exception):
            raise outcome
        return outcome

    async def scenario():
        async with HttpProvider(
            transport=httpx.MockTransport(handler), sleeper=sleeper,
        ) as provider:
            try:
                receipt = await run(REQUEST, ALLOWED, provider, executor, trace)
                return receipt, provider
            except AppError as error:
                return error.code, provider

    result, provider = asyncio.run(scenario())
    assert provider.client.is_closed
    return result, calls, sleeper.calls, executor, trace


def test_adapter_serializes_only_contract_fields():
    secret = "fixture-private-note-not-for-provider"
    calls = []

    def handler(request):
        calls.append(request)
        return httpx.Response(200, json=OK)

    async def scenario():
        async with HttpProvider(transport=httpx.MockTransport(handler)) as provider:
            request = TicketRequest.model_validate(REQUEST | {"private_note": secret})
            assert await provider.propose(request) == OK["content"]
        assert provider.client.is_closed

    asyncio.run(scenario())
    assert len(calls) == 1
    assert calls[0].method == "POST"
    assert str(calls[0].url) == "https://provider.example/propose"
    assert calls[0].headers["content-type"] == "application/json"
    assert json.loads(calls[0].content) == REQUEST
    assert secret.encode() not in calls[0].content


@pytest.mark.parametrize("status", [401, 403, 500])
def test_terminal_http_status_does_not_retry(status):
    result, calls, delays, executor, trace = exercise(httpx.Response(status))
    assert result == ("provider_auth" if status in (401, 403) else "provider_error")
    assert len(calls) == 1 and delays == [] and executor.calls == []
    assert trace.phase == "failed" and trace.receipt is None


@pytest.mark.parametrize("response,code", [
    (httpx.Response(200, content=b'{"status":'), "provider_protocol"),
    (httpx.Response(200, json={"status": "ok", "content": 123}), "provider_protocol"),
    (httpx.Response(200, json={"status": "refused", "content": "No"}), "provider_refused"),
    (httpx.Response(200, json={"status": "error", "content": "Upstream"}), "provider_error"),
    (httpx.Response(200, json={"status": "ok", "content": '{"ticket_id":'}), "invalid_output"),
])
def test_wire_and_generated_output_failures(response, code):
    result, calls, delays, executor, trace = exercise(response)
    assert result == code and len(calls) == 1 and delays == []
    assert executor.calls == [] and trace.receipt is None


@pytest.mark.parametrize("first", [
    httpx.Response(429), httpx.Response(503),
    httpx.ConnectTimeout("injected connect timeout"),
    httpx.ReadTimeout("injected read timeout"),
])
def test_read_only_generation_retries_with_injected_time(first):
    result, calls, delays, executor, trace = exercise(first, httpx.Response(200, json=OK))
    assert result == "receipt-001" and len(calls) == 2 and delays == [0.25]
    assert len(executor.calls) == 1 and trace.phase == "succeeded"


@pytest.mark.parametrize("status", [429, 503])
def test_retry_budget_exhaustion_has_exact_counts(status):
    result, calls, delays, executor, trace = exercise(
        *(httpx.Response(status) for _ in range(3)),
    )
    assert result == "provider_exhausted" and len(calls) == 3
    assert delays == [0.25, 0.5] and executor.calls == []
    assert trace.phase == "failed" and trace.receipt is None


def test_retry_after_delta_is_respected_without_sleeping():
    result, calls, delays, _, _ = exercise(
        httpx.Response(429, headers={"Retry-After": "2"}),
        httpx.Response(200, json=OK),
    )
    assert result == "receipt-001" and len(calls) == 2 and delays == [2.0]


@pytest.mark.parametrize("value", ["30", "not-a-delay", "Wed, 16 Sep 2026 12:00:00 GMT"])
def test_unsupported_retry_after_defers_instead_of_retrying_early(value):
    result, calls, delays, executor, _ = exercise(
        httpx.Response(503, headers={"Retry-After": value}),
    )
    assert result == "provider_deferred" and len(calls) == 1
    assert delays == [] and executor.calls == []


def test_cancellation_before_dispatch_closes_owned_http_resources():
    async def scenario():
        entered = asyncio.Event()
        executor, trace = ExecutorFake(), Trace()

        async def handler(request):
            entered.set()
            await asyncio.Event().wait()
            raise AssertionError("unreachable")

        class ClosingTransport(httpx.MockTransport):
            closed = False

            async def aclose(self):
                self.closed = True
                await super().aclose()

        transport = ClosingTransport(handler)
        provider = HttpProvider(transport=transport)

        async def owned_run():
            async with provider:
                await run(REQUEST, ALLOWED, provider, executor, trace)

        task = asyncio.create_task(owned_run())
        await asyncio.wait_for(entered.wait(), timeout=2)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        assert task.cancelled() and provider.client.is_closed and transport.closed
        assert trace.phase == "cancelled" and trace.receipt is None
        assert executor.calls == []
    asyncio.run(scenario())


@pytest.mark.parametrize("failure", [
    httpx.WriteTimeout("injected write timeout"),
    httpx.ConnectError("injected connection failure"),
])
def test_other_transport_errors_are_terminal_under_this_policy(failure):
    result, calls, delays, executor, trace = exercise(failure)
    assert result == "provider_error" and len(calls) == 1
    assert delays == [] and executor.calls == []
    assert trace.phase == "failed" and trace.receipt is None


def test_last_attempt_can_succeed_after_two_different_failures():
    result, calls, delays, executor, trace = exercise(
        httpx.Response(429), httpx.Response(503), httpx.Response(200, json=OK),
    )
    assert result == "receipt-001" and len(calls) == 3
    assert delays == [0.25, 0.5] and len(executor.calls) == 1
    assert trace.phase == "succeeded"

Control retry time and preserve the attempt budget

The default budget is three total generation attempts, with fallback waits of 0.25 and 0.5 seconds. The suite asserts those requested values through SleeperFake; no retry test waits for those durations. Separate cases cover 429, 503, connect timeout, and read timeout followed by success. Exhaustion checks three calls, two delays, provider_exhausted, zero executor calls, and no receipt. Another test succeeds on the third and final attempt.

A 429 response indicates rate limiting and may include Retry-After. The sample supports small integer delay values up to two seconds. It uses the larger of the fallback and the requested delay. When another attempt is available, dates, malformed values, and larger waits produce provider_deferred without sleeping or retrying early; no scheduler is implemented to resume them.

HTTP semantics distinguish retry timing from whether an operation may be repeated. This tutorial does not retry 401/403 with unchanged credentials, and its permission to repeat generation does not transfer to executor writes. In a real adapter, decide retryable operations and failure classes from its external contract before adding jitter, elapsed-time budgets, or queue-based deferral.

A connect timeout and a timeout waiting for a response can carry different information about what reached a remote server. Here both are retryable only because of the fictional generation contract. A write timeout and other request errors are terminal under this deliberately narrow policy. Status-based backoff is not evidence of idempotency.

Keep ambiguous writes, cancellation, and diagnostics honest

The lost-acknowledgement test appends a simulated write before raising TimeoutError. run returns no success receipt, records ambiguous, and raises effect_unknown. It does not ask the executor to repeat. The caller must reconcile the operation elsewhere. There is no local deduplication layer, operation lookup API, or exactly-once behavior in this implementation.

The LLM-to-API gateway guide develops authorization, operation identity, and external-effect architecture in more depth. Here the handoff is a concrete regression assertion: one observed executor entry despite a lost receipt. Local state alone cannot make a remote write atomic.

Python recommends allowing task cancellation to propagate after cleanup. Before dispatch, the test waits until the provider transport is entered, cancels the task, and asserts cancelled, no receipt, zero executor calls, and client/transport closure. After dispatch, it waits until the fake has recorded the write, cancels, and asserts ambiguous plus executor cleanup. Cancellation is not reported as rollback.

Events synchronize the cancellation tests at the intended boundary. Their two-second wait_for limits are deadlock guards, not retry sleeps or performance assertions. The context owns the provider client; the runner records its state and re-raises CancelledError. Repeated cancellation during cleanup, process termination, durable recovery, and concurrent invocations are outside these tests.

The diagnostic test checks a synthetic private sentinel against captured ticket_app logs, str(error), and a formatted traceback. The application logs stable error codes and suppresses validation exception chaining instead of logging raw provider content. This is bounded evidence for that path. Debugger locals, dumps, third-party logging, tracing exporters, and other fixtures need their own controls. The private_note field is omitted from HTTP serialization; repr=False alone would not prevent model_dump from including it.

Separate local checks, real integrations, and model evaluation

Keep this suite cheap and credential-free in local development and CI. For each real provider, write a separate adapter against official documentation and run opt-in conformance tests against a controlled account or sandbox. Verify authentication, request shape, response envelopes, refusal forms, limits, and documented retry semantics there. Do not relabel these fictional transport fixtures as live-provider evidence.

The real executor needs its own adapter tests and integration evidence: what counts as an acknowledgement, how to query an uncertain operation, and which failures establish that no write occurred. If the API supports idempotency keys, test its actual scope, retention, payload matching, and replay behavior. A local cache is insufficient evidence for any of those remote semantics.

For whether a proposed closure is justified by the ticket evidence, use a separate LLM evaluation pipeline with representative cases and appropriate grading. That work owns semantic quality. Keep parser, policy, side-effect, and retry assertions exact here; record model-quality changes in the evaluation suite.

Evidence still needed when replacing the tutorial doubles.
Change	Next evidence
Replace the fictional generation protocol	Official provider contract plus opt-in integration and conformance tests.
Connect a real ticket executor	Acknowledgement, reconciliation, permissions, and documented idempotency behavior.
Persist and distribute runs	Crash recovery, concurrent claims, state versions, and durable operation tracking.
Change model or prompt	Task-specific semantic evaluation, with local contract tests still running.

Run the complete suite and interpret the observed result

Create a Python virtual environment, install requirements.txt, and keep application.py, http_adapter.py, fakes.py, test_application.py, and test_adapter.py together. The command below runs synchronous pytest functions that use asyncio.run internally, so pytest-asyncio is not required.

The example was executed on Windows with Python 3.14.7, pytest 9.1.1, HTTPX 0.28.1, and Pydantic 2.13.5. The displayed output is from that successful execution of these files. Its elapsed time is an observation from one run, not a benchmark or an expected timing assertion.

On macOS or Linux, create the same .venv and use .venv/bin/python in place of the Windows interpreter path. Dependency installation requires package access; the test run itself uses fakes and controlled transports. Pins identify the tested direct dependencies rather than claiming a complete cross-platform lockfile.

A passing result establishes the exercised contracts under these fixtures. It does not eliminate hallucinations, prove provider conformance, or establish deployment readiness. Keep the files together when adapting the example, and add assertions for any new boundary before relying on it.

Direct dependency versions used for the recorded run.requirements.txt

pytest==9.1.1
httpx==0.28.1
pydantic==2.13.5

Run from the directory containing the six saved files; activation is optional.

python -m venv .venv
.\.venv\Scripts\python.exe -m pip install -r requirements.txt

Observed pytest output from the actual tutorial suite. Disabling bytecode and pytest caching keeps generated files out of the example directory.

.\.venv\Scripts\python.exe -B -m pytest -q -p no:cacheprovider --color=no

Expected output

.............................................                            [100%]
45 passed in 1.76s

FAQ

Should I assert the exact text generated by the model?

Assert exact text when it is a fixed fixture used to exercise your own parser or routing rule. Live model wording belongs in a task-specific evaluation with suitable acceptance criteria. A model can change its wording without breaking an application contract, and a schema-valid answer can still be wrong.

Can these tests run without an API key or internet access?

Yes, after installing the pinned dependencies. Application tests use recording fakes; adapter tests use HTTPX transports. The provider.example address labels a fictional protocol. No real provider or external ticket system is exercised.

Why use recording fakes instead of patching every method?

A small fake makes received arguments, scripted outcomes, and unexpected extra calls visible in ordinary Python. It is easy to assert that the executor was never called. Signature-aware mocks can still be useful for a legacy seam, but neither approach supplies independent evidence about a remote provider.

Why not retry the ticket action after a timeout?

The write might already have been applied. This sample records an ambiguous outcome and requires reconciliation outside the runner. A local cache cannot atomically commit both local state and a remote write. Any later retry policy needs evidence from the external API, including its operation lookup or documented idempotency semantics.

Does a green suite mean this application is ready to deploy?

It establishes only the assertions exercised by these files and dependencies. Real authentication, provider conformance, an actual executor adapter, durable recovery, concurrent workers, response-size limits, operational telemetry, and model quality still need separate implementation and evidence.

Sources

Primary and authoritative sources reviewed for this article.

pytest: parametrizing tests
Supports running the same contract assertion against separately reported input and HTTP failure cases.
pytest: monkeypatching modules and environments
Supports scoped replacement of dependency references and automatic restoration; explicit injection remains the main example design.
Python unittest.mock: autospeccing
Supports the limited comparison with mocks that check callable signatures; it does not establish remote behavior.
HTTPX: custom and mock transports
Supports exercising the concrete HTTP adapter with controlled responses and a transport that records closure.
Pydantic: strict mode
Supports strict model configuration and the warning that strict behavior depends on the field type and validation entry point.
RFC 9110: HTTP semantics
Sections 9.2.2, 10.2.3, 15.5.2, 15.5.4, and 15.6.4 inform the bounded discussion of retries, Retry-After, 401, 403, and 503. The tutorial policy is application-specific.
RFC 6585 section 4: 429 Too Many Requests
Supports rate-limit status semantics and optional Retry-After; the status alone does not establish that repeating an operation is safe.
Python asyncio: task cancellation
Supports propagating CancelledError after recording state and allowing the resource-owning context to clean up.

Conclusion

Make each uncertain boundary replaceable, then assert what the surrounding software owes its caller. Record calls, reject invalid proposals, prove that denied actions never reach the executor, inspect retry delays without waiting, and preserve ambiguity when a write may have escaped. Carry these contracts into integration tests and maintain model evaluations alongside them. The useful result is a set of inspectable behaviors with explicit limits.