Skip to content

Test Your Application Offline

Your application's inference code has behavior worth testing: it falls back when a provider is down, it repairs a malformed structured answer, it reduces a corpus to fit a budget. Testing that normally means either mocking the library, which tests your mocks, or calling a real provider from CI, which is slow, costs money, and fails for reasons that have nothing to do with your change.

AnyInfer ships the third option. anyinfer.testing gives you a provider whose behavior you declare, and pytest fixtures that wire it to a real client. Everything runs in-process: no sockets, no credentials, no network, and the same result on every machine.

pip install anyinfer     # the fixtures come with it — no extra to install

Declare a Provider, Get a Real Client

The fixtures are available as soon as anyinfer is installed. anyinfer_scripted builds a provider; anyinfer_client builds a client wired to it.

from anyinfer.testing import ScriptedModel


def test_summarizer_returns_the_models_answer(anyinfer_client, anyinfer_scripted):
    provider = anyinfer_scripted([ScriptedModel("small", text="A one-sentence summary.")])
    client = anyinfer_client(provider)

    result = client.generate("Summarize this", target=provider.target("small"))

    assert result.text == "A one-sentence summary."

Everything between your call and that assertion is the real library: the router resolved the target, the adapter spoke the wire dialect, the core measured the timings.

Prove Your Fallback Chain Works

A scripted model can be told to fail. Failures are consumed in order, then the model answers normally, so "fails once, then succeeds" is one line, enough to exercise the fallback chain you configured.

from anyinfer.testing import ScriptedFailure, ScriptedModel


def test_retries_a_transient_failure(anyinfer_client, anyinfer_scripted):
    provider = anyinfer_scripted(
        [ScriptedModel("flaky", failures=(ScriptedFailure(status=503, retry_after_s=0.0),))]
    )
    client = anyinfer_client(provider)

    result = client.generate("hi", target=provider.target("flaky"))

    assert [attempt.outcome for attempt in result.attempts] == ["retried", "ok"]

retry_after_s=0.0 advertises the header without making your suite wait for it.

Prove Your Repair Budget Converges

malformed-json answers with something that will not validate, so the repair loop runs for real:

from anyinfer.testing import ScriptedFailure, ScriptedModel

SCHEMA = {
    "type": "object",
    "properties": {"answer": {"type": "string"}},
    "required": ["answer"],
}


def test_repairs_an_invalid_structured_answer(anyinfer_client, anyinfer_scripted):
    provider = anyinfer_scripted(
        [
            ScriptedModel(
                "structured",
                structured={"answer": "valid on the second try"},
                failures=(ScriptedFailure(kind="malformed-json"),),
            )
        ]
    )
    client = anyinfer_client(provider)

    result = client.generate(
        "extract", target=provider.target("structured"), schema=SCHEMA, repair={"max_attempts": 1}
    )

    assert result.structured == {"answer": "valid on the second try"}
    assert result.repair_attempts == 1

The Failures You Can Script

kind What the provider does What it lets you test
status Returns an HTTP error, optionally with Retry-After Retry, backoff, the attempt trail
truncate Cuts the stream mid-event Partial-response handling, teardown
malformed-json Answers with something that fails validation Schema validation and the repair loop
timeout Raises a read timeout Your timeout handling, without waiting
refusal Finishes with content_filter Your content-policy fallback

Assert on Telemetry

anyinfer_events collects the same typed event stream your observers consume in production. It is payload-free: it never captures prompt or response text, so adding it to a suite cannot start logging user content.

from anyinfer.events.telemetry import RetryScheduled


def test_emits_a_retry_event(anyinfer_client, anyinfer_scripted, anyinfer_events):
    provider = anyinfer_scripted(
        [ScriptedModel("flaky", failures=(ScriptedFailure(retry_after_s=0.0),))]
    )
    anyinfer_client(provider).generate("hi", target=provider.target("flaky"))

    assert anyinfer_events.of_type(RetryScheduled)

Model Capabilities Are Declarable Too

A scripted model states what it supports. Declaring a model without JSON support is how you test what your code does on the weakest structured-output mechanism:

from anyinfer.types.capabilities import Feature, ModelCapabilities, Sourced

ScriptedModel(
    "plain",
    structured={"answer": "ok"},
    capabilities=ModelCapabilities(
        context_window=Sourced(8_192, "catalog"),
        features=Sourced(Feature.STREAMING | Feature.SYSTEM_PROMPT, "catalog"),
    ),
)

The result still validates (client-side validation is always authoritative), but result.structured_mechanism reports prompt instead of json_schema, which is what your production code will see against a model that cannot do better.

Regression-Test Inference Behavior

When the route, repair budget, cache placement, or context policy is the contract you care about, compare a golden run manifest instead of asserting on model prose or unstable timing:

def test_answer_path(anyinfer_golden_manifest):
    result = application.answer("hi")
    anyinfer_golden_manifest(result.manifest, "answer-path")

The fixture removes request IDs and timings before comparing manifests/answer-path.json beside your test. Run pytest --update-manifests only after an intentional behavior change, then review the JSON diff. The complete fallback-and-repair example runs offline in AnyInfer's own suite.

The Fixtures

Fixture What it gives you
anyinfer_scripted Factory for scripted providers, registered for this test only
anyinfer_client Factory for sync clients, closed automatically
anyinfer_async_client The same, asynchronous
anyinfer_events A payload-free telemetry collector
anyinfer_registry The per-test provider registry, if you need it directly
anyinfer_cassette Resolves a cassette stored beside your test file
anyinfer_recording Whether this run is recording cassettes
anyinfer_golden_manifest Compare a normalized run manifest with a checked-in golden

Each test gets its own provider registry, so two tests may register the same provider id without depending on execution order. Every fixture and scripted type is specified in the testing API reference.

Recording Real Traffic

In order to test against what a provider actually sent, record it once and replay it forever:

def test_against_recorded_traffic(anyinfer_cassette, anyinfer_recording):
    cassette = anyinfer_cassette("summarize")
    ...

Run your suite with ANYINFER_RECORD_CASSETTES=1 to record; unset it to replay. Recorded bodies pass through the redaction registry before reaching disk, so a cassette you commit alongside a test cannot carry a registered credential.

Key Takeaways

  • A scripted provider exercises the real router, adapter layer, and core, so the suite tests the library's behavior rather than your mocks.
  • Failures are consumed in order before the model answers normally, which makes "fails once, then succeeds" a one-line declaration.
  • anyinfer_events is payload-free by construction; adding telemetry assertions cannot start logging user content.
  • Golden run manifests pin routing and policy decisions without asserting on prose; refresh them with pytest --update-manifests only after an intentional change.
  • Recorded cassettes pass through redaction before reaching disk, so committing one cannot leak a registered credential.

See Also