Testing Utilities¶

The anyinfer.testing package is public on purpose, for two audiences. An application tests its own routing, repair, and reduction logic against a scripted provider (guide: test your application offline). A third-party adapter certifies itself by running the same conformance suite the built-in adapters run (guide: the conformance suite).

Scripted Providers¶

A provider whose behavior is declared per model, including the failures that are otherwise unreachable without a real outage.

anyinfer.testing.assert_manifest_matches ¶

assert_manifest_matches(
    manifest: RunManifest | Mapping[str, Any],
    path: Path | str,
    *,
    update: bool = False,
) -> None

Assert a manifest matches its golden file, normalizing volatility away first.

A missing golden is written rather than failed: the first run of a new test records what the behaviour is, and the diff in review is where it gets agreed to. A golden that exists and disagrees is a failure, reported as a field-by-field difference rather than two walls of JSON.

Parameters:

Name	Type	Description	Default
`manifest`	`RunManifest \| Mapping[str, Any]`	The manifest the run produced.	required
`path`	`Path \| str`	Where the golden lives. Parent directories are created as needed.	required
`update`	`bool`	Rewrite the golden instead of comparing. Wired to the pytest plugin's `--update-manifests` flag.	`False`

Raises:

Type	Description
`AssertionError`	If the golden exists and the run no longer matches it.

anyinfer.testing.ScriptedProvider ¶

ScriptedProvider(
    provider_id: str = "scripted",
    models: Sequence[ScriptedModel] | None = None,
    *,
    aliases: Sequence[str] = (),
    locality: Literal[
        "hosted", "local", "remote"
    ] = "local",
    base_url: str = "http://scripted.invalid/v1",
    display_name: str | None = None,
)

A registered provider whose behaviour is declared, not coded.

Parameters:

Name	Type	Description	Default
`provider_id`	`str`	Id to register under, and the left half of every target it serves.	`'scripted'`
`models`	`Sequence[ScriptedModel] \| None`	The models this provider serves. At least one is required.	`None`
`aliases`	`Sequence[str]`	Additional names resolving to this provider.	`()`
`locality`	`Literal['hosted', 'local', 'remote']`	`"local"` (the default) keeps cost at a genuine zero and hardware detection honest; pass `"hosted"` when a test needs pricing to apply.	`'local'`
`base_url`	`str`	Endpoint recorded in settings. Never contacted — the transport intercepts every request, so it uses an unroutable `.invalid` host.	`'http://scripted.invalid/v1'`
`display_name`	`str \| None`	Human-readable name for UIs and error messages. Defaults to naming the provider as scripted, which is the right answer in a test and the wrong one in an application that ships a scripted provider as a visible offline mode.	`None`

Attributes:

Name	Type	Description
`requests`	`list[dict[str, Any]]`	Every request body received, oldest first, across all models.

models `property` ¶

models: tuple[ScriptedModel, ...]

The models this provider serves, in declaration order.

target ¶

target(model_id: str | None = None) -> str

The target string for one of this provider's models.

Parameters:

Name	Type	Description	Default
`model_id`	`str \| None`	Which model; defaults to the first declared.	`None`

Returns:

Type	Description
`str`	A `provider:model` target.

descriptor ¶

descriptor() -> ProviderDescriptor

The declarative descriptor this provider registers.

register ¶

register(
    registry: ProviderRegistry | None = None,
) -> ProviderRegistry

Register this provider, replacing any earlier registration of the same id.

Parameters:

Name	Type	Description	Default
`registry`	`ProviderRegistry \| None`	Where to register. Defaults to the process-wide registry, which is convenient in a one-off script and wrong in a test suite — the pytest fixtures hand each test its own registry so parallel tests cannot collide.	`None`

Returns:

Type	Description
`ProviderRegistry`	The registry that was written to.

settings ¶

settings(**overrides: Any) -> ProviderSettings

Provider settings wired to this provider's in-process transport.

transport ¶

transport() -> httpx2.MockTransport

An httpx2 transport serving this provider's scripted behaviour.

reset ¶

reset() -> None

Rewind every model's failure script and forget recorded requests.

call_count ¶

call_count(model_id: str | None = None) -> int

How many generation calls were served, in total or for one model.

with_model ¶

with_model(model: ScriptedModel) -> ScriptedProvider

Return this provider with one model's declaration replaced or added.

Convenience for a test that needs a variant of an otherwise shared provider without rebuilding the whole table.

anyinfer.testing.ScriptedModel `dataclass` ¶

ScriptedModel(
    id: str,
    text: str = "Scripted answer.",
    structured: Mapping[str, Any] | None = None,
    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,
        }
    )(),
    chunk_size: int = 4,
    answer_after_tools: str | None = None,
    failures: tuple[ScriptedFailure, ...] = (),
    capabilities: ModelCapabilities = DEFAULT_SCRIPTED_CAPABILITIES,
)

One model a scripted provider serves, and how it behaves.

Attributes:

Name	Type	Description
`id`	`str`	Model id, as it appears after the colon in a target.
`text`	`str`	Assistant text to answer with. Ignored when `structured` is set.
`structured`	`Mapping[str, Any] \| None`	An object to answer with, serialized as JSON. The convenient way to script a request that carries a schema.
`tool_calls`	`tuple[tuple[str, str, str], ...]`	Tool calls to emit, as `(id, name, arguments_json)` triples.
`finish_reason`	`str`	Normalized finish reason to report.
`usage`	`Mapping[str, Any] \| None`	Usage block to report, or `None` to report none at all, which is how a test reaches the estimated-usage path.
`chunk_size`	`int`	Characters per streamed delta. Smaller values produce more events.
`answer_after_tools`	`str \| None`	Text to answer with once the conversation already carries a tool result. Without it, a model scripted to call a tool calls it on every round and the loop never converges, which is a test that only ever proves the round budget works. Set it and the model behaves like a real one: ask, then answer.
`failures`	`tuple[ScriptedFailure, ...]`	Failures consumed in order before any success. A model with two failures and a retry budget of one will exhaust the budget; that is the point.
`capabilities`	`ModelCapabilities`	What this model claims to support. Defaults to `DEFAULT_SCRIPTED_CAPABILITIES`; declare narrower capabilities to force a weaker structured-output mechanism.

answer_text `property` ¶

answer_text: str

The body this model answers with, structured content winning over prose.

anyinfer.testing.ScriptedFailure `dataclass` ¶

ScriptedFailure(
    kind: FailureKind = "status",
    status: int = 503,
    retry_after_s: float | None = None,
    message: str = "scripted failure",
)

One scripted failure, consumed by the next call to its model.

Attributes:

Name	Type	Description
`kind`	`FailureKind`	Which failure to produce; see `FailureKind`.
`status`	`int`	HTTP status for `status` failures.
`retry_after_s`	`float \| None`	`Retry-After` seconds to advertise, when the failure carries one. Use `0.0` in tests that assert retry behaviour without waiting for it.
`message`	`str`	Error text the provider reports. Bounded and inert — it is test data, not a template.

anyinfer.testing.FailureKind `module-attribute` ¶

FailureKind = Literal[
    "status",
    "truncate",
    "malformed-json",
    "timeout",
    "refusal",
]

How a scripted call fails.

Each kind reaches a different part of the core, and each is otherwise reachable only by provoking a real provider into misbehaving:

status An HTTP error, optionally carrying Retry-After. Exercises status classification, backoff, and the retry event. truncate A stream cut mid-event. Exercises stream teardown and partial-result handling. malformed-json A body that will not validate against the requested schema. Exercises validation and the bounded repair loop. timeout A read timeout, raised rather than slept, so the case is deterministic. refusal A completed response reporting content_filter. Exercises the content-policy chain.

anyinfer.testing.DEFAULT_SCRIPTED_CAPABILITIES `module-attribute` ¶

DEFAULT_SCRIPTED_CAPABILITIES = ModelCapabilities(
    context_window=Sourced(32768, "default"),
    features=Sourced(
        Feature.STREAMING
        | Feature.TOOLS
        | Feature.SYSTEM_PROMPT
        | Feature.JSON_MODE,
        "default",
    ),
)

Capabilities a scripted model claims when it declares none of its own.

Deliberately default provenance: a scripted provider is not a source of truth about what any real model supports, and a test that needs a trusted window should say so by declaring one.

anyinfer.testing.VOLATILE_FIELDS `module-attribute` ¶

VOLATILE_FIELDS: Mapping[str, tuple[str, ...]] = {
    "": ("request_id", "anyinfer_version"),
    "attempts": (
        "first_token_ms",
        "total_ms",
        "queued_ms",
        "retry_delay_s",
        "paced_s",
    ),
    "timing": (
        "first_token_ms",
        "total_ms",
        "output_tokens_per_s",
        "phases",
    ),
}

Which fields normalize() drops, keyed by the facet they live on.

Everything here is a wall-clock or a per-process identifier: true of the run, and false of the next identical one. Nothing about a decision is in this list, because a decision that changed is precisely what a golden manifest exists to catch.

anyinfer.testing.normalize ¶

normalize(
    manifest: RunManifest | Mapping[str, Any],
) -> dict[str, Any]

Strip the fields that differ between two identical runs.

Parameters:

Name	Type	Description	Default
`manifest`	`RunManifest \| Mapping[str, Any]`	A `RunManifest` or its `RunManifest.to_dict` form.	required

Returns:

Type	Description
`dict[str, Any]`	JSON-safe data with every volatile field removed, ready to compare or write.

Fake Embedding and Rerank Providers¶

An in-process fake implementing EmbedsText/ReranksText directly; there is no wire dialect to reproduce for these operations, so this fake needs no mock transport.

anyinfer.testing.FakeEmbeddingRerankProvider ¶

FakeEmbeddingRerankProvider(
    provider_id: str = "fake-embed",
    *,
    embedding_dimensions: Mapping[str, int] | None = None,
    rerank_models: Sequence[str] = (),
    embedding_failures: Mapping[
        str, Sequence[ScriptedEmbeddingFailure]
    ]
    | None = None,
    normalized: bool = True,
    locality: Literal[
        "hosted", "local", "remote"
    ] = "local",
    embedding_capabilities: Mapping[
        str, EmbeddingCapabilities
    ]
    | None = None,
    declared_embedding: Mapping[str, EmbeddingCapabilities]
    | None = None,
    rerank_capabilities: Mapping[str, RerankCapabilities]
    | None = None,
    pricing: Mapping[str, Pricing] | None = None,
)

An in-process fake supporting EmbedsText and ReranksText, or both.

Parameters:

Name	Type	Description	Default
`provider_id`	`str`	Id to register under, and the left half of every target it serves.	`'fake-embed'`
`embedding_dimensions`	`Mapping[str, int] \| None`	Vector length `embed()` produces, keyed by model id. A model id absent here does not embed.	`None`
`rerank_models`	`Sequence[str]`	Model ids that support reranking. A model id absent here does not rerank.	`()`
`embedding_failures`	`Mapping[str, Sequence[ScriptedEmbeddingFailure]] \| None`	Failures consumed in order before any success, keyed by model id.	`None`
`normalized`	`bool`	Whether vectors this provider produces are reported as unit-normalized.	`True`
`locality`	`Literal['hosted', 'local', 'remote']`	Recorded on the descriptor.	`'local'`
`embedding_capabilities`	`Mapping[str, EmbeddingCapabilities] \| None`	Static `EmbeddingCapabilities` recorded on the descriptor, keyed by model id — the way tests declare a verified batch limit (`max_batch_inputs`) for core-owned batching to resolve.	`None`
`declared_embedding`	`Mapping[str, EmbeddingCapabilities] \| None`	`EmbeddingCapabilities` reported through `list_models()`, keyed by model id — the other way a provider states model-level facts, used by listings that tag their own models and by local engines reading a pinned catalog row. Distinct from `embedding_capabilities` on purpose: a descriptor table is keyed by ids known when the code was written, and a provider whose ids come from the machine it runs on cannot use one.	`None`
`rerank_capabilities`	`Mapping[str, RerankCapabilities] \| None`	Static `RerankCapabilities` recorded on the descriptor, keyed by model id (`max_documents` drives rerank batching).	`None`
`pricing`	`Mapping[str, Pricing] \| None`	Trusted per-model pricing recorded on the descriptor (provenance `"catalog"`), so cost computation and spend ceilings can be tested offline.	`None`

Attributes:

Name	Type	Description
`embed_requests`	`list[EmbeddingWireRequest]`	Every `EmbeddingWireRequest` received, oldest first.
`rerank_requests`	`list[RerankWireRequest]`	Every `RerankWireRequest` received, oldest first.

operations ¶

operations() -> frozenset[InferenceOperation]

Which operations this provider declares, for building its descriptor.

descriptor ¶

descriptor() -> ProviderDescriptor

The declarative descriptor this provider registers.

register ¶

register(registry: ProviderRegistry) -> ProviderRegistry

Register this provider's descriptor, replacing any earlier registration.

list_models `async` ¶

list_models() -> Sequence[DiscoveredModel]

Enumerate the models this fake serves, with any facts they declare.

health `async` ¶

health() -> Health

Always healthy — this fake never simulates transport-level outages at health time.

aclose `async` ¶

aclose() -> None

No resources to release.

embed `async` ¶

embed(req: EmbeddingWireRequest) -> EmbeddingWireResult

Return deterministic pseudo-embeddings, or raise a scripted failure.

rerank `async` ¶

rerank(req: RerankWireRequest) -> RerankWireResult

Return documents ranked by deterministic lexical overlap with the query.

anyinfer.testing.ScriptedEmbeddingFailure `dataclass` ¶

ScriptedEmbeddingFailure(
    kind: _EmbeddingFailureKind = "status",
    retry_after_s: float | None = None,
    message: str = "scripted failure",
)

One scripted failure, consumed by the next call to the model it is attached to.

Attributes:

Name	Type	Description
`kind`	`_EmbeddingFailureKind`	`"status"` raises a generic retryable `ProviderError`; `"rate-limit"` raises `RateLimitError` with `retry_after_s`; `"transport"` raises `TransportError`.
`retry_after_s`	`float \| None`	Advertised retry delay for a `rate-limit` failure.
`message`	`str`	Error text reported. Test data only.

Fake MCP Server¶

An in-process Model Context Protocol server, for testing a tool loop fed by MCP tool sources.

anyinfer.testing.FakeMCPServer ¶

FakeMCPServer(
    tools: Sequence[FakeMCPTool] | None = None,
    *,
    protocol_version: str = PROTOCOL_VERSION,
    page_size: int = 0,
)

A scripted MCP server that answers over an in-process transport.

Parameters:

Name	Type	Description	Default
`tools`	`Sequence[FakeMCPTool] \| None`	The tools to advertise.	`None`
`protocol_version`	`str`	Version to answer the handshake with. Override it to test a client's version negotiation.	`PROTOCOL_VERSION`
`page_size`	`int`	Advertise tools across several `tools/list` pages, to exercise pagination. `0` means one page.	`0`

Attributes:

Name	Type	Description
`calls`	`list[tuple[str, Mapping[str, Any]]]`	Every `tools/call` received, as `(name, arguments)` pairs.

transport ¶

transport() -> _FakeMCPTransport

An anyinfer.mcp.MCPTransport serving this fake.

handle ¶

handle(message: Mapping[str, Any]) -> dict[str, Any] | None

Answer one JSON-RPC message, or None for a notification.

anyinfer.testing.FakeMCPTool `dataclass` ¶

FakeMCPTool(
    name: str,
    description: str = "A fake tool.",
    parameters: Mapping[str, Any] = dict(),
    result: str = "fake tool result",
    is_error: bool = False,
    annotations: Mapping[str, Any] = dict(),
    blocks: tuple[Mapping[str, Any], ...] = (),
)

One tool a fake server advertises, and what calling it produces.

Attributes:

Name	Type	Description
`name`	`str`	The tool's name, un-namespaced.
`description`	`str`	What the model is told it does.
`parameters`	`Mapping[str, Any]`	JSON Schema for its arguments; an empty object accepts anything.
`result`	`str`	Text the tool returns.
`is_error`	`bool`	Whether the call reports its own failure, which a loop feeds back to the model rather than raising.
`annotations`	`Mapping[str, Any]`	Behavioural hints to advertise, in the protocol's own spelling (`readOnlyHint` and friends).
`blocks`	`tuple[Mapping[str, Any], ...]`	Overrides `result` with raw content blocks, for testing what happens to non-text content.

pytest Fixtures¶

Registered automatically when anyinfer is installed; see the guide for the full table.

anyinfer.testing.plugin.EventCollector ¶

EventCollector()

Collects telemetry events for assertions.

A collector, not an assertion library: it hands back what happened and leaves the judging to the test. Registered payload-free unless a test explicitly opts in, so a suite cannot start capturing prompt text by accident.

on_event ¶

on_event(event: TelemetryEvent) -> None

Record one event.

of_type ¶

of_type(*types: type) -> list[Any]

Every recorded event that is an instance of any of types.

request_ids ¶

request_ids() -> list[str]

Correlation ids seen, in first-seen order.

clear ¶

clear() -> None

Forget everything recorded so far.

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,
    reasoning_field: str | None = None,
    dimensions: int = 8,
)

Bases: _FakeServerBase

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`
`reasoning_field`	`str \| None`	Name of the dialect's reasoning key (DeepSeek and xAI both send `reasoning_content`). `None` — the plain OpenAI shape — omits reasoning entirely, so an adapter with no reasoning channel cannot accidentally pass the reasoning case.	`None`
`dimensions`	`int`	Width of the vectors `POST /embeddings` returns.	`8`

Attributes:

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

next_response ¶

next_response() -> FakeResponse

The response for the next generation call.

anyinfer.testing.FakeResponsesServer ¶

FakeResponsesServer(
    responses: Sequence[FakeResponse]
    | FakeResponse
    | None = None,
    *,
    models: Sequence[str] = ("gpt-5", "gpt-5-mini"),
    chunk_size: int = 4,
    dimensions: int = 8,
)

Bases: _FakeServerBase

A configurable in-process OpenAI Responses API endpoint.

Not the same dialect as FakeOpenAIServer, which speaks /chat/completions. The Responses protocol is typed events — response.output_text.delta, response.output_item.added, response.completed — rather than choice deltas, and a finish reason is derived from the terminal response object's status and incomplete_details rather than sent as a field. Modelling that difference is the point: an adapter that quietly treated one as the other would pass a chat-shaped fake.

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`.	`('gpt-5', 'gpt-5-mini')`
`chunk_size`	`int`	Characters per streamed text delta.	`4`
`dimensions`	`int`	Width of the vectors `POST /embeddings` returns.	`8`

Attributes:

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

next_response ¶

next_response() -> FakeResponse

The response for the next generation call.

anyinfer.testing.FakeAnthropicServer ¶

FakeAnthropicServer(
    responses: Sequence[FakeResponse]
    | FakeResponse
    | None = None,
    *,
    models: Sequence[str] = (
        "claude-sonnet-4-5",
        "claude-opus-4-1",
    ),
    chunk_size: int = 4,
    page_size: int = 1,
    batch_polls_before_done: int = 1,
    batch_failures: int = 0,
)

Bases: _FakeServerBase

A configurable in-process Messages API endpoint.

Two properties of the real API shape this fake. It always streams — the adapter sends stream: true unconditionally, so there is no buffered branch to model — and it has no response-format field, so a schema is emulated as a single forced tool call and the structured answer arrives as a tool_use block rather than as text.

That second point is why this fake reads the request before answering. A scenario says "return this JSON"; whether that JSON belongs in a text_delta or in the input of a tool_use block is a property of what was asked, not of the scenario. The tool name is echoed from the request's own tool_choice rather than hardcoded, so the fake cannot drift from whatever the core decides to call the schema.

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 /v1/models`.	`('claude-sonnet-4-5', 'claude-opus-4-1')`
`chunk_size`	`int`	Characters per streamed text delta.	`4`
`batch_polls_before_done`	`int`	How many status polls a submitted batch reports `in_progress` for before it ends. At least one, so a caller must always poll at least once — a fake that ended immediately could never exercise the poll-then-fetch shape that is the whole point of a deferred API.	`1`
`batch_failures`	`int`	How many of a batch's lines come back as per-line errors. A batch is not all-or-nothing, and a result type that could not carry a partial failure would force the whole job to be discarded over one bad request.	`0`
`page_size`	`int`	Model ids per listing page. The listing is cursor-paginated, and an adapter that ignores `has_more` silently reports only the first page -- which looks like a working discovery call.	`1`

Attributes:

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

next_response ¶

next_response() -> FakeResponse

The response for the next generation call.

anyinfer.testing.FakeBedrockServer ¶

FakeBedrockServer(
    responses: Sequence[FakeResponse]
    | FakeResponse
    | None = None,
    *,
    models: Sequence[str] = (
        "anthropic.claude-sonnet-4-5-v1:0",
        "amazon.titan-embed-text-v2:0",
    ),
    chunk_size: int = 4,
    dimensions: int = 8,
    batch_polls_before_done: int = 1,
    batch_failures: int = 0,
)

Bases: _FakeServerBase

A configurable in-process Bedrock endpoint spanning its four actions.

Bedrock is not one API. Generation is /model/{id}/converse buffered or /model/{id}/converse-stream in AWS's binary event framing; Titan embeddings are /model/{id}/invoke on the same runtime host; rerank is a different service (bedrock-agent-runtime's POST /rerank); and discovery is a different host again (the control plane's /foundation-models). Routing all four through one fake is what lets the shared suite treat Bedrock as one provider the way a caller does.

Two Converse details the fake models deliberately. Usage arrives only in the terminal metadata event, so a stream that ends at messageStop reports no tokens — an adapter reading usage from the wrong event silently loses it. And a schema is emulated as a forced tool call, so the structured answer arrives as a toolUse block rather than as text.

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 the control plane's `/foundation-models`.	`('anthropic.claude-sonnet-4-5-v1:0', 'amazon.titan-embed-text-v2:0')`
`chunk_size`	`int`	Characters per streamed text delta.	`4`
`dimensions`	`int`	Width of the vectors `/invoke` returns.	`8`
`batch_polls_before_done`	`int`	Status polls a submitted job reports in progress for before it finishes. At least one, so a caller must always poll.	`1`
`batch_failures`	`int`	How many records come back carrying an `error` rather than a `modelOutput`. Bedrock puts both in the same object, unlike OpenAI's two files, and an adapter that assumed one or the other would drop them.	`0`

Attributes:

Name	Type	Description
`requests`	`list[dict[str, Any]]`	Every request body received, for assertions.
`objects`	`dict[str, bytes]`	Every S3 object written, keyed by URL — a batch is staged in the caller's own bucket rather than uploaded over the API, so the fake has to stand in for S3 as well as for Bedrock.

next_response ¶

next_response() -> FakeResponse

The response for the next 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,
    embed_scenario: str | None = None,
)

Bases: _FakeServerBase

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.

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,
    batch_polls_before_done: int = 1,
    batch_failures: int = 0,
)

Bases: _FakeServerBase

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.

anyinfer.testing.FakeRetrievalServer ¶

FakeRetrievalServer(
    responses: Sequence[FakeResponse]
    | FakeResponse
    | None = None,
    *,
    dimensions: int = 8,
    rerank_key: str = "data",
    top_n_key: str = "top_n",
    models_status: int = 404,
)

Bases: _FakeServerBase

An in-process endpoint for retrieval-only providers (Voyage, Jina).

These providers speak a narrow dialect: POST /embeddings in the OpenAI shape, a POST /rerank that differs only in which key holds the ranking, and no listing route at all. Sharing one fake keeps their conformance rows honest about the same scenarios every other adapter answers.

Parameters:

Name	Type	Description	Default
`responses`	`Sequence[FakeResponse] \| FakeResponse \| None`	Scenario script, as for the other fakes. Only `status`, `headers`, and `error_message` apply — there is no text to generate.	`None`
`dimensions`	`int`	Width of the vectors `POST /embeddings` returns.	`8`
`rerank_key`	`str`	Key holding the ranking. Voyage uses `data`; Jina uses `results`.	`'data'`
`top_n_key`	`str`	Request key carrying the truncation count. Voyage spells it `top_k`; Jina sends a plain `top_n`. A fake that accepted either would let an adapter send the wrong one and still pass.	`'top_n'`
`models_status`	`int`	Status for `GET /models`. These providers document no listing route, so the honest default is a 404 that still proves reachability.	`404`

Attributes:

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

next_response ¶

next_response() -> FakeResponse

The response for the next retrieval call.

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,
    logprobs: tuple[tuple[str, float], ...] = (),
    top_logprobs: tuple[tuple[str, float], ...] = (),
)

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.
`logprobs`	`tuple[tuple[str, float], ...]`	Report these per-token log-probabilities, as `(token, logprob)` pairs, but only when the request asked for them. A fake that answered with log-probabilities nobody requested would let an adapter that never sends the field pass a test for sending it.
`top_logprobs`	`tuple[tuple[str, float], ...]`	Alternatives to attach to each reported token, as `(token, logprob)` pairs. Sent only when the request asked for a positive count.

anyinfer.testing.scenario_responses ¶

scenario_responses(
    scenario: str,
    *,
    text: str = "Hello from the fake provider.",
    reasoning: str = "Let me think.",
    probe_answer: str = _PROBE_ANSWER,
) -> list[FakeResponse]

The canonical response programme for one conformance scenario.

Every harness answers the same nine scenarios with the same shapes: a tool call for tools, an invalid-then-valid pair for repair, a 401 for auth_error, and so on. Only the wire encoding differs per dialect, and the fake server classes already own that. Programming the scenarios here means a new adapter's harness declares its dialect and its capabilities — never a ninth copy of the same if/elif chain, which is exactly where a subtly weaker probe slips into one provider's column unnoticed.

Parameters:

Name	Type	Description	Default
`scenario`	`str`	A key from `CONFORMANCE_SCENARIOS`. An unrecognized key is treated as `default`, so a harness never has to enumerate them.	required
`text`	`str`	Assistant text for the scenarios that just need a successful answer.	`'Hello from the fake provider.'`
`reasoning`	`str`	Thinking text for the `reasoning` scenario. Dialects without a reasoning channel ignore it and declare `reasoning=False` instead.	`'Let me think.'`
`probe_answer`	`str`	The JSON answer satisfying `PROBE_SCHEMA`.	`_PROBE_ANSWER`

Returns:

Type	Description
`list[FakeResponse]`	Responses to serve in order. The fake servers reuse the last one once exhausted,
`list[FakeResponse]`	so a single-element list answers every request in the scenario.

anyinfer.testing.CONFORMANCE_SCENARIOS `module-attribute` ¶

CONFORMANCE_SCENARIOS: tuple[str, ...] = (
    "default",
    "tools",
    "reasoning",
    "structured",
    "repair",
    "auth_error",
    "rate_limited",
    "oversized",
    "odd_finish",
)

Every scenario key run_conformance hands to a harness's client factory.

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

anyinfer.testing.eventstream_frame ¶

eventstream_frame(event_type: str, payload: Any) -> bytes

Encode one application/vnd.amazon.eventstream frame.

AWS frames its Converse stream in a binary envelope with two CRC32s rather than in SSE, so a fake that returned JSON lines would exercise a decoder the adapter does not have. The checksums are computed, not stubbed: an adapter that skips validating them should not be able to pass by being handed frames that were never valid.

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`

aclose `async` ¶

aclose() -> None

Close the recording-mode inner transport, when one was opened.

AsyncBaseTransport.aclose is a no-op by default; recording opens a real httpx2.AsyncHTTPTransport in __init__ and nothing else was closing its connection pool, which leaked a socket per test under this project's filterwarnings = ["error"] gate. Replay mode has no inner to close.

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.

Recording¶

Turning a live run against the developer's own account into committable cassettes. The audit is a second, independent pass over the saved bytes: redaction removes secrets it was told about, and this looks for credential shapes it was not.

anyinfer.testing.audit_cassette ¶

audit_cassette(
    source: Cassette | Path | Iterable[Interaction],
) -> list[AuditFinding]

Findings across a whole cassette.

A Path is read from disk, which is the form that matters: it audits the bytes that would actually be committed, after Cassette.save() has applied redaction, rather than the in-memory objects redaction was handed.

Parameters:

Name	Type	Description	Default
`source`	`Cassette \| Path \| Iterable[Interaction]`	A cassette, a path to a saved cassette file, or interactions directly.	required

Returns:

Type	Description
`list[AuditFinding]`	Every finding, in cassette order.

anyinfer.testing.audit_interaction ¶

audit_interaction(
    interaction: Interaction, index: int = 0
) -> list[AuditFinding]

Findings for one recorded interaction.

Parameters:

Name	Type	Description	Default
`interaction`	`Interaction`	The interaction as it would be written to disk.	required
`index`	`int`	Its position in the cassette, for the report.	`0`

Returns:

Type	Description
`list[AuditFinding]`	Every credential-shaped survivor, in scan order. An empty list means nothing
`list[AuditFinding]`	matched — not that the interaction is provably clean.

anyinfer.testing.AuditFinding `dataclass` ¶

AuditFinding(
    interaction: int, where: str, shape: str, excerpt: str
)

One credential-shaped string that survived redaction.

Attributes:

Name	Type	Description
`interaction`	`int`	Index of the interaction it appeared in.
`where`	`str`	Which part carried it — `url`, `request_body`, `body`, or a header name.
`shape`	`str`	Name of the pattern that matched, e.g. `"bearer-token"`.
`excerpt`	`str`	A short, masked excerpt for the report. Never the full value: a finding printed to a terminal or a CI log must not itself become the leak.

str ¶

__str__() -> str

One line for a terminal report.

anyinfer.testing.SECRET_SHAPES `module-attribute` ¶

SECRET_SHAPES: tuple[tuple[str, Pattern[str]], ...] = (
    (
        "vendor-api-key",
        re.compile(
            "\\b(?:sk-ant-|sk-|xai-|gsk_|AIza|co-|r8_)[A-Za-z0-9_\\-]{16,}"
        ),
    ),
    (
        "bearer-token",
        re.compile("\\bBearer\\s+[A-Za-z0-9._\\-]{16,}"),
    ),
    (
        "jwt",
        re.compile(
            "\\beyJ[A-Za-z0-9_\\-]{8,}\\.[A-Za-z0-9_\\-]{8,}\\.[A-Za-z0-9_\\-]{8,}"
        ),
    ),
    (
        "aws-access-key-id",
        re.compile("\\b(?:AKIA|ASIA)[0-9A-Z]{16}\\b"),
    ),
    (
        "private-key",
        re.compile(
            "-----BEGIN (?:[A-Z ]+ )?PRIVATE KEY-----"
        ),
    ),
    (
        "credential-field",
        re.compile(
            '"(?:[a-z_]*(?:api[_-]?key|secret|token|password|credential)[a-z_]*)"\\s*:\\s*"(?!\\[redacted\\])[^"]{12,}"',
            re.IGNORECASE,
        ),
    ),
)

Credential shapes an audit looks for, independently of what redaction knew about.

Each is a public convention — a documented key prefix, a standard token encoding — which is precisely why it can be matched without knowing the secret. Anything genuinely opaque and unprefixed cannot be found this way, which is why audit_cassette reports findings for a human rather than claiming a cassette is clean.

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(),
    covered_elsewhere: frozenset[str] = frozenset(),
    embedding_model: str | None = None,
    rerank_model: str | None = None,
    retry: Retry = Retry(
        max_attempts=2, backoff_base_s=0.0
    ),
)

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.
`covered_elsewhere`	`frozenset[str]`	Capability names that are implemented and tested, but in a dedicated module instead of this harness — usually because exercising them here would need a second fake with a different shape. Named explicitly so the matrix can distinguish "this adapter cannot" from "the proof is next door"; a name here without the matching `supports` flag left `False` is a contradiction, so the annotation cannot quietly overstate coverage.
`retry`	`Retry`	The policy the rate-limiting cases route through. It defaults to no backoff because those cases assert the attempt trail -- that a 429 is recorded as a retried attempt and typed retryable -- and never assert how long the client waited. Sleeping the real backoff proves nothing and, at three cases per adapter across every preset, was the single largest block of wall time in the suite. A live-mode harness that talks to a real rate-limited endpoint should override this with a policy that honors `Retry-After`.

target `property` ¶

target: str

The target string for this harness.

embedding_target `property` ¶

embedding_target: str

The embedding target, falling back to the generation model.

rerank_target `property` ¶

rerank_target: str

The rerank target, falling back to the generation model.

retry_route ¶

retry_route(target: str) -> Route

A route to target carrying this harness's retry policy.

The rate-limiting cases state their policy here rather than passing a bare target= and relying on the client to supply one. A client whose default route sets a policy would pass it down either way, but a harness is free to build a client without a default route at all, and these three cases are the ones where the difference is measured in seconds of real sleeping.

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,
    cancellation: bool = False,
    embedding: bool = False,
    rerank: bool = False,
)

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.
`cancellation`	`bool`	Abandoning a stream mid-flight releases its upstream connection and leaves the client usable. Defaults to `False`: this is a case a harness opts into, because a fake that cannot be held open mid-stream would fail it for a reason that says nothing about the adapter.
`embedding`	`bool`	`embed()` returns ordered, uniform, finite vectors. Defaults to `False` — most adapters generate only, and an operation nobody declared must skip rather than fail.
`rerank`	`bool`	`rerank()` returns descending, identity-preserving rankings. Defaults to `False` for the same reason.

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,
    covered_elsewhere: 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.
`covered_elsewhere`	`bool`	The capability is implemented, but its coverage lives in a dedicated test module rather than this shared harness. Refines `skipped` rather than replacing it — the case still did not run here, so every failure check stays correct — but the matrix draws it differently, because collapsing both onto `➖` reported working adapters as unable.
`detail`	`str`	Why the case failed (truncated), or why it was skipped; empty on a pass.

symbol `property` ¶

symbol: str

The cell this result draws in the conformance matrix.

✅ pass · 🔗 implemented, covered by dedicated tests · ➖ 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(
        "cancellation",
        "streaming",
        "cancellation",
        _case_cancellation,
    ),
    ConformanceCase(
        "unknown_finish_reason",
        "odd_finish",
        "non_streaming",
        _case_unknown_finish_reason,
    ),
    ConformanceCase(
        "embedding", "default", "embedding", _case_embedding
    ),
    ConformanceCase(
        "embedding_duplicates",
        "default",
        "embedding",
        _case_embedding_duplicates,
    ),
    ConformanceCase(
        "rerank", "default", "rerank", _case_rerank
    ),
    ConformanceCase(
        "rerank_top_n",
        "default",
        "rerank",
        _case_rerank_top_n,
    ),
    ConformanceCase(
        "rerank_duplicate_text",
        "default",
        "rerank",
        _case_rerank_duplicate_text,
    ),
    ConformanceCase(
        "embedding_normalization_probe",
        "default",
        "embedding",
        _case_embedding_normalization_probe,
    ),
    ConformanceCase(
        "embedding_byte_cap",
        "oversized",
        "embedding",
        _case_embedding_byte_cap,
    ),
    ConformanceCase(
        "rerank_byte_cap",
        "oversized",
        "rerank",
        _case_rerank_byte_cap,
    ),
    ConformanceCase(
        "embedding_retry_after",
        "rate_limited",
        "embedding",
        _case_embedding_retry_after,
    ),
    ConformanceCase(
        "rerank_retry_after",
        "rate_limited",
        "rerank",
        _case_rerank_retry_after,
    ),
)

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¶

Scripted Providers¶

anyinfer.testing.assert_manifest_matches ¶

anyinfer.testing.ScriptedProvider ¶

models property ¶

target ¶

descriptor ¶

register ¶

settings ¶

transport ¶

reset ¶

call_count ¶

with_model ¶

anyinfer.testing.ScriptedModel dataclass ¶

answer_text property ¶

anyinfer.testing.ScriptedFailure dataclass ¶

anyinfer.testing.FailureKind module-attribute ¶

anyinfer.testing.DEFAULT_SCRIPTED_CAPABILITIES module-attribute ¶

anyinfer.testing.VOLATILE_FIELDS module-attribute ¶

anyinfer.testing.normalize ¶

Fake Embedding and Rerank Providers¶

anyinfer.testing.FakeEmbeddingRerankProvider ¶

operations ¶

descriptor ¶

register ¶

list_models async ¶

health async ¶

aclose async ¶

embed async ¶

rerank async ¶

anyinfer.testing.ScriptedEmbeddingFailure dataclass ¶

Fake MCP Server¶

anyinfer.testing.FakeMCPServer ¶

transport ¶

handle ¶

anyinfer.testing.FakeMCPTool dataclass ¶

pytest Fixtures¶

anyinfer.testing.plugin.EventCollector ¶

on_event ¶

of_type ¶

request_ids ¶

clear ¶

Fake Providers¶

anyinfer.testing.FakeOpenAIServer ¶

next_response ¶

anyinfer.testing.FakeResponsesServer ¶

next_response ¶

anyinfer.testing.FakeAnthropicServer ¶

next_response ¶

anyinfer.testing.FakeBedrockServer ¶

next_response ¶

anyinfer.testing.FakeOllamaServer ¶

anyinfer.testing.FakeGeminiServer ¶

anyinfer.testing.FakeRetrievalServer ¶

next_response ¶

anyinfer.testing.FakeResponse dataclass ¶

anyinfer.testing.scenario_responses ¶

anyinfer.testing.CONFORMANCE_SCENARIOS module-attribute ¶

anyinfer.testing.chunk_text ¶

anyinfer.testing.sse_lines ¶

anyinfer.testing.ndjson_lines ¶

anyinfer.testing.eventstream_frame ¶

Cassettes¶

anyinfer.testing.Cassette ¶

load ¶

save ¶

append ¶

anyinfer.testing.CassetteTransport ¶

aclose async ¶

handle_async_request async ¶

anyinfer.testing.Interaction dataclass ¶

to_json ¶

from_json classmethod ¶

Recording¶

anyinfer.testing.audit_cassette ¶

anyinfer.testing.audit_interaction ¶

anyinfer.testing.AuditFinding dataclass ¶

__str__ ¶

anyinfer.testing.SECRET_SHAPES module-attribute ¶

Conformance¶

models `property` ¶

anyinfer.testing.ScriptedModel `dataclass` ¶

answer_text `property` ¶

anyinfer.testing.ScriptedFailure `dataclass` ¶

anyinfer.testing.FailureKind `module-attribute` ¶

anyinfer.testing.DEFAULT_SCRIPTED_CAPABILITIES `module-attribute` ¶

anyinfer.testing.VOLATILE_FIELDS `module-attribute` ¶

list_models `async` ¶

health `async` ¶

aclose `async` ¶

embed `async` ¶

rerank `async` ¶

anyinfer.testing.ScriptedEmbeddingFailure `dataclass` ¶

anyinfer.testing.FakeMCPTool `dataclass` ¶

anyinfer.testing.FakeResponse `dataclass` ¶

anyinfer.testing.CONFORMANCE_SCENARIOS `module-attribute` ¶

aclose `async` ¶

handle_async_request `async` ¶

anyinfer.testing.Interaction `dataclass` ¶

from_json `classmethod` ¶

anyinfer.testing.AuditFinding `dataclass` ¶

str ¶

anyinfer.testing.SECRET_SHAPES `module-attribute` ¶

anyinfer.testing.conformance.run_conformance `async` ¶

anyinfer.testing.conformance.ConformanceHarness `dataclass` ¶

target `property` ¶

embedding_target `property` ¶

rerank_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` ¶