Testing utilities¶

The anyinfer.testing package is public on purpose: a third-party adapter certifies itself by running the same conformance suite the built-in adapters run. Guide: the conformance suite.

Fake providers¶

In-process fake servers that speak the real wire dialects, so examples and tests run without credentials or a network.

anyinfer.testing.FakeOpenAIServer ¶

FakeOpenAIServer(
    responses: Sequence[FakeResponse]
    | FakeResponse
    | None = None,
    *,
    models: Sequence[str] = (
        "fake-model-small",
        "fake-model-large",
    ),
    chunk_size: int = 4,
)

A configurable in-process OpenAI-compatible endpoint.

Parameters:

Name	Type	Description	Default
`responses`	`Sequence[FakeResponse] \| FakeResponse \| None`	Responses to serve, one per request, in order. The last is reused once exhausted, so a single-element list serves every request.	`None`
`models`	`Sequence[str]`	Model ids reported by `GET /models`.	`('fake-model-small', 'fake-model-large')`
`chunk_size`	`int`	Characters per streamed text delta.	`4`

Attributes:

Name	Type	Description
`requests`	`list[dict[str, Any]]`	Every request body received, for assertions.

call_count `property` ¶

call_count: int

How many generation requests have been served.

transport ¶

transport() -> httpx2.MockTransport

Build an httpx2 transport that routes to this fake.

next_response ¶

next_response() -> FakeResponse

The response for the next generation call.

anyinfer.testing.FakeOllamaServer ¶

FakeOllamaServer(
    responses: Sequence[FakeResponse]
    | FakeResponse
    | None = None,
    *,
    models: Sequence[str] = ("qwen3:8b", "qwen2.5:3b"),
    loaded: Mapping[str, int] | None = None,
    chunk_size: int = 4,
)

A configurable in-process Ollama server speaking the native NDJSON dialect.

Parameters:

Name	Type	Description	Default
`responses`	`Sequence[FakeResponse] \| FakeResponse \| None`	Responses to serve, one per request, in order. The last is reused once exhausted.	`None`
`models`	`Sequence[str]`	Models reported by `/api/tags`.	`('qwen3:8b', 'qwen2.5:3b')`
`loaded`	`Mapping[str, int] \| None`	`model -> size_vram` entries reported by `/api/ps`, for GPU-spill tests.	`None`
`chunk_size`	`int`	Characters per streamed text delta.	`4`

Attributes:

Name	Type	Description
`requests`	`list[dict[str, Any]]`	Every request body received, for assertions.

call_count `property` ¶

call_count: int

How many generation requests have been served.

transport ¶

transport() -> httpx2.MockTransport

Build an httpx2 transport that routes to this fake.

anyinfer.testing.FakeGeminiServer ¶

FakeGeminiServer(
    responses: Sequence[FakeResponse]
    | FakeResponse
    | None = None,
    *,
    models: Sequence[str] = (
        "gemini-2.5-flash",
        "gemini-2.5-pro",
    ),
    chunk_size: int = 4,
)

A configurable in-process Gemini endpoint speaking the native protocol.

Parameters:

Name	Type	Description	Default
`responses`	`Sequence[FakeResponse] \| FakeResponse \| None`	Responses to serve, one per request, in order. The last is reused once exhausted.	`None`
`models`	`Sequence[str]`	Model ids reported by `GET /models`.	`('gemini-2.5-flash', 'gemini-2.5-pro')`
`chunk_size`	`int`	Characters per streamed text part.	`4`

Attributes:

Name	Type	Description
`requests`	`list[dict[str, Any]]`	Every request body received, for assertions.

call_count `property` ¶

call_count: int

How many generation requests have been served.

transport ¶

transport() -> httpx2.MockTransport

Build an httpx2 transport that routes to this fake.

anyinfer.testing.FakeResponse `dataclass` ¶

FakeResponse(
    text: str = "Hello from the fake provider.",
    reasoning: str = "",
    tool_calls: tuple[tuple[str, str, str], ...] = (),
    finish_reason: str = "stop",
    usage: Mapping[str, Any] | None = (
        lambda: {
            "prompt_tokens": 11,
            "completion_tokens": 7,
            "total_tokens": 18,
        }
    )(),
    status: int = 200,
    error_message: str = "fake provider error",
    headers: Mapping[str, str] = dict(),
    malformed_sse: bool = False,
    ignore_stream: bool = False,
    omit_usage_chunk: bool = False,
)

A scripted response the fake server should produce.

Attributes:

Name	Type	Description
`text`	`str`	Assistant text to emit, chunked across deltas when streaming.
`reasoning`	`str`	Thinking text to emit before the answer. Only dialects with a reasoning channel surface it (Ollama's `thinking` field, Gemini's `thought`-flagged parts).
`tool_calls`	`tuple[tuple[str, str, str], ...]`	Tool calls to emit, as `(id, name, arguments_json)` triples.
`finish_reason`	`str`	Finish reason to report.
`usage`	`Mapping[str, Any] \| None`	Usage block to report, or `None` to omit it entirely.
`status`	`int`	HTTP status; `>= 400` produces an error body instead of a completion.
`error_message`	`str`	Message for error responses.
`headers`	`Mapping[str, str]`	Extra response headers (e.g. `retry-after`).
`malformed_sse`	`bool`	Emit an unparseable SSE data field, to exercise error handling.
`ignore_stream`	`bool`	Answer a streaming request with a buffered JSON body.
`omit_usage_chunk`	`bool`	Stream without a terminal usage chunk.

anyinfer.testing.chunk_text ¶

chunk_text(text: str, size: int = 4) -> list[str]

Split text into fixed-size fragments, mimicking token-level streaming.

anyinfer.testing.sse_lines ¶

sse_lines(
    payloads: Iterable[Any], *, done: bool = True
) -> bytes

Encode payloads as an SSE body.

anyinfer.testing.ndjson_lines ¶

ndjson_lines(payloads: Iterable[Any]) -> bytes

Encode payloads as an NDJSON body (Ollama's framing).

Cassettes¶

Record/replay of real provider exchanges.

anyinfer.testing.Cassette ¶

Cassette(path: Path)

A file of recorded interactions.

load ¶

load() -> None

Read interactions from disk.

save ¶

save() -> None

Write interactions to disk, redacting secrets first.

append ¶

append(interaction: Interaction) -> None

Record an interaction.

anyinfer.testing.CassetteTransport ¶

CassetteTransport(
    cassette: Cassette,
    *,
    record: bool = False,
    inner: AsyncBaseTransport | None = None,
)

Bases: AsyncBaseTransport

Replays a cassette, or records live traffic into one.

Parameters:

Name	Type	Description	Default
`cassette`	`Cassette`	The cassette to read or write.	required
`record`	`bool`	When `True`, forward requests to `inner` and record the results.	`False`
`inner`	`AsyncBaseTransport \| None`	The transport used while recording. Required in record mode.	`None`

handle_async_request `async` ¶

handle_async_request(request: Request) -> httpx2.Response

Serve one request from the cassette, or record it live.

anyinfer.testing.Interaction `dataclass` ¶

Interaction(
    method: str,
    url: str,
    request_body: str,
    status: int,
    headers: dict[str, str],
    body: str,
)

One recorded request/response exchange.

Attributes:

Name	Type	Description
`method`	`str`	HTTP method of the recorded request; replay matches on it.
`url`	`str`	Full request URL; replay matches on its path, and redaction scrubs it before it reaches disk.
`request_body`	`str`	The request body as text, redacted at save time.
`status`	`int`	HTTP status code of the recorded response.
`headers`	`dict[str, str]`	Response headers. Secret-bearing headers are replaced wholesale at save time; the rest pass through redaction.
`body`	`str`	The response body as text, redacted at save time and replayed verbatim.

to_json ¶

to_json() -> dict[str, Any]

Serialize for storage.

from_json `classmethod` ¶

from_json(data: dict[str, Any]) -> Interaction

Deserialize from storage.

Conformance¶

The parametrized suite behind the conformance matrix.

anyinfer.testing.conformance.run_conformance `async` ¶

run_conformance(
    harness: ConformanceHarness,
    *,
    only: Sequence[str] | None = None,
) -> list[CaseResult]

Run the suite against one adapter.

Parameters:

Name	Type	Description	Default
`harness`	`ConformanceHarness`	The adapter under test.	required
`only`	`Sequence[str] \| None`	Restrict the run to these case names.	`None`

Returns:

Type	Description
`list[CaseResult]`	One `CaseResult` per case, in matrix order.

anyinfer.testing.conformance.ConformanceHarness `dataclass` ¶

ConformanceHarness(
    provider_id: str,
    model: str,
    build_client: Callable[[str], Awaitable[AsyncClient]],
    supports: Capabilities = Capabilities(),
)

Everything the suite needs to exercise one adapter.

Attributes:

Name	Type	Description
`provider_id`	`str`	The provider under test.
`model`	`str`	Model id to send.
`build_client`	`Callable[[str], Awaitable[AsyncClient]]`	Builds a client whose scripted responses match `scenario`. The suite passes a scenario name so the harness can program its fake or select its cassette.
`supports`	`Capabilities`	Declared capabilities; unsupported cases are skipped.

target `property` ¶

target: str

The target string for this harness.

anyinfer.testing.conformance.Capabilities `dataclass` ¶

Capabilities(
    list_models: bool = True,
    health: bool = True,
    non_streaming: bool = True,
    streaming: bool = True,
    ttft: bool = True,
    usage: bool = True,
    tools: bool = True,
    reasoning: bool = True,
    structured_output: bool = True,
    repair: bool = True,
    retry_after: bool = True,
    error_mapping: bool = True,
    byte_cap: bool = True,
)

What a provider claims to support, so unsupported cases skip honestly.

Each flag gates at least one case of the conformance matrix. Setting a flag to False is a documented ➖, not a pass.

Attributes:

Name	Type	Description
`list_models`	`bool`	Model discovery returns the provider's models.
`health`	`bool`	The provider answers a health probe.
`non_streaming`	`bool`	Whole-response generation, including finish-reason normalization.
`streaming`	`bool`	Incremental generation with the event-ordering guarantees.
`ttft`	`bool`	Time to first token is measurable on streams.
`usage`	`bool`	Token usage is reported, including usage that trails the finish reason.
`tools`	`bool`	Tool calls surface completely, streaming and non-streaming.
`reasoning`	`bool`	Reasoning streams on its own channel, excluded from the answer text.
`structured_output`	`bool`	Schema-constrained generation yields a validated value.
`repair`	`bool`	An invalid structured value can be repaired within the attempt budget.
`retry_after`	`bool`	Rate limiting surfaces as a retryable, recorded attempt.
`error_mapping`	`bool`	Provider failures map to typed errors with a correct retry flag.
`byte_cap`	`bool`	An oversized response is rejected rather than silently truncated.

anyinfer.testing.conformance.ConformanceCase `dataclass` ¶

ConformanceCase(
    name: str,
    scenario: str,
    requires: str,
    run: Callable[
        [AsyncClient, ConformanceHarness], Awaitable[None]
    ],
)

One named check.

Attributes:

Name	Type	Description
`name`	`str`	Matrix row name.
`scenario`	`str`	Scenario key handed to the harness's client factory.
`requires`	`str`	Capability flag gating this case.
`run`	`Callable[[AsyncClient, ConformanceHarness], Awaitable[None]]`	The check itself; raises `AssertionError` on failure.

anyinfer.testing.conformance.CaseResult `dataclass` ¶

CaseResult(
    name: str,
    passed: bool,
    skipped: bool = False,
    detail: str = "",
)

The outcome of one conformance case.

Attributes:

Name	Type	Description
`name`	`str`	The case's row name in the conformance matrix.
`passed`	`bool`	Whether the check succeeded. Also `False` for skipped cases; check `skipped` first.
`skipped`	`bool`	The harness declared the capability unsupported, so the case did not run.
`detail`	`str`	Why the case failed (truncated), or why it was skipped; empty on a pass.

symbol `property` ¶

symbol: str

Matrix symbol: ✅ pass, ➖ declared-unsupported, ❌ failure.

anyinfer.testing.conformance.CONFORMANCE_CASES `module-attribute` ¶

CONFORMANCE_CASES: tuple[ConformanceCase, ...] = (
    ConformanceCase(
        "list_models",
        "default",
        "list_models",
        _case_list_models,
    ),
    ConformanceCase(
        "health", "default", "health", _case_health
    ),
    ConformanceCase(
        "non_streaming",
        "default",
        "non_streaming",
        _case_non_streaming,
    ),
    ConformanceCase(
        "streaming", "default", "streaming", _case_streaming
    ),
    ConformanceCase(
        "event_ordering",
        "default",
        "streaming",
        _case_event_ordering,
    ),
    ConformanceCase("ttft", "default", "ttft", _case_ttft),
    ConformanceCase(
        "usage", "default", "usage", _case_usage
    ),
    ConformanceCase(
        "usage_survives_streaming",
        "default",
        "usage",
        _case_usage_survives_streaming,
    ),
    ConformanceCase(
        "tool_calls", "tools", "tools", _case_tool_calls
    ),
    ConformanceCase(
        "streaming_tool_calls",
        "tools",
        "tools",
        _case_streaming_tool_calls,
    ),
    ConformanceCase(
        "reasoning",
        "reasoning",
        "reasoning",
        _case_reasoning,
    ),
    ConformanceCase(
        "structured_output",
        "structured",
        "structured_output",
        _case_structured_output,
    ),
    ConformanceCase(
        "schema_repair", "repair", "repair", _case_repair
    ),
    ConformanceCase(
        "error_mapping",
        "auth_error",
        "error_mapping",
        _case_error_mapping,
    ),
    ConformanceCase(
        "retry_after",
        "rate_limited",
        "retry_after",
        _case_retry_after,
    ),
    ConformanceCase(
        "byte_cap", "oversized", "byte_cap", _case_byte_cap
    ),
    ConformanceCase(
        "unknown_finish_reason",
        "odd_finish",
        "non_streaming",
        _case_unknown_finish_reason,
    ),
)

Every conformance case, in matrix order.

anyinfer.testing.conformance.PROBE_SCHEMA `module-attribute` ¶

PROBE_SCHEMA = {
    "type": "object",
    "properties": {"answer": {"type": "string"}},
    "required": ["answer"],
    "additionalProperties": False,
}

The schema every structured-output probe requests.

anyinfer.testing.conformance.PROBE_TOOL `module-attribute` ¶

PROBE_TOOL = ToolSpec(
    name="lookup",
    description="Look up a value by key.",
    parameters={
        "type": "object",
        "properties": {"key": {"type": "string"}},
        "required": ["key"],
    },
)

The tool every tool-calling probe advertises.

anyinfer.testing.conformance.matrix_row ¶

matrix_row(
    provider_id: str, results: Sequence[CaseResult]
) -> str

Render results as one Markdown conformance-matrix row.

Provider documentation pages embed this so a page cannot overstate what the suite actually verified.

anyinfer.testing.conformance.results_to_json ¶

results_to_json(
    provider_id: str, results: Sequence[CaseResult]
) -> str

Serialize results for the docs build.

Testing utilities¶

Fake providers¶

anyinfer.testing.FakeOpenAIServer ¶

call_count property ¶

transport ¶

next_response ¶

anyinfer.testing.FakeOllamaServer ¶

call_count property ¶

transport ¶

anyinfer.testing.FakeGeminiServer ¶

call_count property ¶

transport ¶

anyinfer.testing.FakeResponse dataclass ¶

anyinfer.testing.chunk_text ¶

anyinfer.testing.sse_lines ¶

anyinfer.testing.ndjson_lines ¶

Cassettes¶

anyinfer.testing.Cassette ¶

load ¶

save ¶

append ¶

anyinfer.testing.CassetteTransport ¶

handle_async_request async ¶

anyinfer.testing.Interaction dataclass ¶

to_json ¶

from_json classmethod ¶

Conformance¶

anyinfer.testing.conformance.run_conformance async ¶

anyinfer.testing.conformance.ConformanceHarness dataclass ¶

target property ¶

anyinfer.testing.conformance.Capabilities dataclass ¶

anyinfer.testing.conformance.ConformanceCase dataclass ¶

anyinfer.testing.conformance.CaseResult dataclass ¶

symbol property ¶

anyinfer.testing.conformance.CONFORMANCE_CASES module-attribute ¶

anyinfer.testing.conformance.PROBE_SCHEMA module-attribute ¶

anyinfer.testing.conformance.PROBE_TOOL module-attribute ¶

anyinfer.testing.conformance.matrix_row ¶

anyinfer.testing.conformance.results_to_json ¶

call_count `property` ¶

call_count `property` ¶

call_count `property` ¶

anyinfer.testing.FakeResponse `dataclass` ¶

handle_async_request `async` ¶

anyinfer.testing.Interaction `dataclass` ¶

from_json `classmethod` ¶

anyinfer.testing.conformance.run_conformance `async` ¶

anyinfer.testing.conformance.ConformanceHarness `dataclass` ¶

target `property` ¶

anyinfer.testing.conformance.Capabilities `dataclass` ¶

anyinfer.testing.conformance.ConformanceCase `dataclass` ¶

anyinfer.testing.conformance.CaseResult `dataclass` ¶

symbol `property` ¶

anyinfer.testing.conformance.CONFORMANCE_CASES `module-attribute` ¶

anyinfer.testing.conformance.PROBE_SCHEMA `module-attribute` ¶

anyinfer.testing.conformance.PROBE_TOOL `module-attribute` ¶