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_eventsis 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-manifestsonly after an intentional change. - Recorded cassettes pass through redaction before reaching disk, so committing one cannot leak a registered credential.