Testing budget enforcement

import pytest
from openai import OpenAI
from solwyn import BudgetExceededError
from solwyn.testing import FakeControlPlane

def test_deny_handler():
    plane = FakeControlPlane()
    with OpenAI(api_key="test") as provider:
        with plane.wrap(provider) as client:
            with pytest.raises(BudgetExceededError):
                client.chat.completions.create(model="solwyn-test/deny", messages=[])
        # Closing the wrapper forwards to the provider client's close seam.
        assert provider.is_closed()

SDK v0.6.0+: solwyn.testing.FakeControlPlane is an in-process double for Solwyn Cloud. It plugs into the same control-plane transport seam production uses, speaks the same request, response, and Pydantic wire models, and never opens a socket. You create and own the plane, wrap your provider client with it, script what the control plane should say, and inspect what the SDK actually sent afterwards.

Three things the double deliberately does not do:

• It never prices anything. Solwyn Cloud owns pricing; the SDK reports token counts and the double records them. A scripted denial tests your handling of a denial, not your budget arithmetic. remaining_budget is a static number you set, never decremented by calls, though a denial floors it at zero (a run_stopped denial always reports 0).
• It does not simulate providers. Mock the provider yourself whenever a test can reach provider dispatch. A test that only exercises a hard denial never needs to, because the denial happens during preflight before any provider request; an alert_only verdict warns and still dispatches, so those tests do need a provider mock.
• It does not reset itself. Build a fresh FakeControlPlane() per test.

Everything in this guide is available from solwyn.testing (FakeControlPlane, MAGIC_MODELS). The contract pack and the pytest plugin live at solwyn.testing.contract and solwyn.testing.pytest_plugin and are imported by full path.

Wrapping a client

plane.wrap(provider, **solwyn_kwargs) returns a normal Solwyn wrapper; plane.wrap_async(provider, **solwyn_kwargs) returns an AsyncSolwyn. The plane's transport serves both sync and async traffic, so the same plane can back both kinds of client in one test.

from openai import AsyncOpenAI
from solwyn.testing import FakeControlPlane

async def test_async_denial():
    plane = FakeControlPlane()
    async with plane.wrap_async(AsyncOpenAI(api_key="test"), fail_open=False) as client:
        with pytest.raises(BudgetExceededError):
            await client.chat.completions.create(model="solwyn-test/deny", messages=[])
    assert len(plane.checks) == 1

That second recipe carries the first one's imports (pytest, BudgetExceededError) and needs an async pytest plugin to run -- pytest-asyncio under --asyncio-mode=auto, or anyio. Neither comes with the SDK install, and nothing else in this guide needs one.

Any Solwyn constructor keyword or SolwynConfig field is accepted -- fail_open, lease_enabled, budget_mode, tags, on_unmetered, the velocity_* settings, and so on -- and your keywords always win over the wrapper's defaults. Three keywords are reserved because the plane supplies them: api_key, api_url, and control_plane_transport. Passing any of them raises TypeError.

The wrapper applies two defaults of its own:

lease_enabled stays at its production default (True). Pass lease_enabled=False when a test should stay on the per-call check path.

Closing the wrapper shuts down the SDK's control-plane resources and then forwards to the wrapped provider client's own close seam, which is why the recipes nest both context managers and assert provider.is_closed().

If you need to wire a client by hand instead of through wrap, the plane exposes plane.api_key (a deterministic project-key-shaped string), plane.api_url (http://control-plane.invalid), and plane.transport, which is also accepted by Solwyn(control_plane_transport=...).

Magic models

Magic models are reserved model names that script a deterministic control-plane verdict. They are validated before provider dispatch (including entries in a fallback chain and on media surfaces), so a magic hard denial never produces a provider request. An alert_only verdict -- solwyn-test/deny-alert, or any denial on a mode="alert_only" plane -- warns and still dispatches, so mock the provider for those.

Note that the plane's default mode is hard_deny, the opposite of the SDK's budget_mode default. The plane models what the server reports, so a test that wants alert-only behavior constructs FakeControlPlane(mode="alert_only") or uses solwyn-test/deny-alert.

A model name that starts with solwyn-test/ but is not one of the seven raises RuntimeError at the wrapper. The three run-scoped models raise RuntimeError when used outside a solwyn.run(...) scope, because the verdicts they script only exist for a run. When a fallback chain contains more than one magic model, the first applicable trigger in chain order wins.

When scripts overlap, the plane resolves each request in a fixed order: a scripted transport outage wins over everything (including a slow window); then an unknown path returns 404 and is recorded in plane.unmatched_requests; then an endpoint refusal (refuse_checks, refuse_leases, read_only); then the verdict, where a run stop outranks a queued denial, which outranks a deny_run, which outranks a magic model; and finally allow. An outage therefore tests unreachable posture without a scripted denial leaking through, while a reachable endpoint refusal wins over the normal verdict.

Scripting scenarios

Scenario windows are context managers. requests=None (the default for all but reject_ingest) leaves the window open until exit; a count bounds it. A request that does not match the window's path never consumes a slot.

Two more methods exercise late settlement without a context manager: plane.expire_reservations() forgets every active reservation so a later confirm meets a 404, and plane.expire_leases() expires every held lease so the SDK follows its renewal and outage ladder.

Scripting denials and stops

Test fail-open posture

Mock the provider separately with the HTTP dialect used by the installed OpenAI SDK, and assert both that dispatch proceeded and that the control-plane warning surfaced.

import logging
from openai import OpenAI
from solwyn.testing import FakeControlPlane

try:
    import httpx2 as provider_httpx
except ImportError:
    import httpx as provider_httpx

def test_fail_open_provider_proceeds(caplog):
    plane = FakeControlPlane()
    provider_requests = []

    def handle_provider(request):
        provider_requests.append(request)
        return provider_httpx.Response(
            200,
            json={
                "id": "chatcmpl-test", "object": "chat.completion", "created": 0,
                "model": "gpt-5.5", "choices": [{"index": 0,
                    "message": {"role": "assistant", "content": "served"},
                    "finish_reason": "stop"}],
                "usage": {"prompt_tokens": 2, "completion_tokens": 1, "total_tokens": 3},
            },
        )

    transport = provider_httpx.MockTransport(handle_provider)
    with provider_httpx.Client(transport=transport) as provider_http_client:
        with OpenAI(
            base_url="https://provider.test/v1",
            api_key="test",
            http_client=provider_http_client,
        ) as provider:
            with (
                plane.wrap(provider, fail_open=True, lease_enabled=False) as client,
                caplog.at_level(logging.WARNING),
                plane.outage(),
            ):
                response = client.chat.completions.create(model="gpt-5.5", messages=[])
            # Closing the wrapper forwards to the provider client's close seam.
            assert provider.is_closed()
    assert len(provider_requests) == 1
    assert response.choices[0].message.content == "served"
    assert "budget check failed" in caplog.text.lower()

The httpx2 import dance is there because some provider SDKs (Anthropic 1.x among them) ship on the separate httpx2 distribution rather than httpx. The provider-side mock transport has to be built with whichever module the installed provider SDK actually uses; the fallback keeps the recipe working when httpx2 is not installed. The Solwyn core never imports httpx2 itself.

Run a deny, outage, recovery game day

Compose scenarios on one caller-owned plane to prove that a known hard denial is preserved during an outage and cleared only by a recovered allow verdict.

import pytest
from openai import OpenAI
from solwyn import BudgetExceededError
from solwyn.testing import FakeControlPlane

try:
    import httpx2 as provider_httpx
except ImportError:
    import httpx as provider_httpx

def test_deny_outage_recovery():
    plane = FakeControlPlane()
    provider_requests = []

    def handle_provider(request):
        provider_requests.append(request)
        return provider_httpx.Response(
            200,
            json={
                "id": "chatcmpl-test", "object": "chat.completion", "created": 0,
                "model": "gpt-5.5", "choices": [{"index": 0,
                    "message": {"role": "assistant", "content": "served"},
                    "finish_reason": "stop"}],
                "usage": {"prompt_tokens": 2, "completion_tokens": 1, "total_tokens": 3},
            },
        )

    transport = provider_httpx.MockTransport(handle_provider)
    with provider_httpx.Client(transport=transport) as provider_http_client:
        with OpenAI(
            base_url="https://provider.test/v1",
            api_key="test",
            http_client=provider_http_client,
        ) as provider:
            with plane.wrap(provider, fail_open=True, lease_enabled=False) as client:
                with pytest.raises(BudgetExceededError):
                    client.chat.completions.create(model="solwyn-test/deny", messages=[])
                with plane.outage(), pytest.raises(BudgetExceededError):
                    client.chat.completions.create(model="gpt-5.5", messages=[])
                recovered = client.chat.completions.create(model="gpt-5.5", messages=[])
            # Closing the wrapper forwards to the provider client's close seam.
            assert provider.is_closed()
    assert len(provider_requests) == 1
    assert recovered.choices[0].message.content == "served"

The middle call is the interesting one: the plane is unreachable and the client is fail_open=True, yet the SDK still raises, because it preserves the last authoritative hard deny through an outage. Only the recovered allow after the window closes clears it. The same composition -- plane.outage(), plane.expire_leases(), plane.refuse_leases(...), plane.slow(...) -- reproduces the refusal, expiry, and outage stages of the lease outage ladder deterministically. The drawdown stages need the plane's grant knobs instead -- granted_tokens, headroom_share_tokens, final_grant, refresh_interval_s, and lease_length_s, all constructor keywords. The default grant (granted_tokens=200_000, refreshed every 30.0 s inside a 90.0 s lease) is more than a test will spend, so a drawdown test constructs something like FakeControlPlane(granted_tokens=30, refresh_interval_s=60.0, lease_length_s=120.0). The SDK repository's tests/unit/testing_double/test_gameday_recipes.py walks the full refusal, breaker, reporter, lease-drawdown, and recovery ladder if you want a template.

Simulating an operator kill

plane.stop_run(run_id) is the dashboard kill switch: from the next request on, every check, lease grant, and lease renewal naming that run is denied and the wrapper raises RunStoppedError -- which is not a BudgetExceededError, so a budget-denial handler cannot swallow it. The stop stays terminal through a control-plane outage, and plane.clear_stop(run_id) lifts it.

import pytest
import solwyn
from openai import OpenAI
from solwyn import RunStoppedError
from solwyn.testing import FakeControlPlane

def test_operator_kill_stops_the_run():
    plane = FakeControlPlane()
    with OpenAI(api_key="test") as provider:
        with plane.wrap(provider) as client, solwyn.run("nightly-report") as run_id:
            plane.stop_run(run_id, reason="operator_stop")
            with pytest.raises(RunStoppedError) as stopped:
                client.chat.completions.create(model="gpt-5.5", messages=[])
        assert stopped.value.agent_run_id == run_id
        assert stopped.value.reason == "operator_stop"
        assert [receipt.deny_reason for receipt in plane.denial_receipts] == ["operator_stop"]

plane.denial_receipts holds the content-free evidence the SDK reported for every call it refused -- who denied it (deny_source), why (deny_reason), and under which period -- with plane.aggregate_replays holding the folded aggregates the SDK replays after an ingest rejection. The receipt fields are documented under Denial receipts.

Two magic models cover the same ground without a run id in hand: solwyn-test/kill allows the first call in a run and then stops the run on every channel, exactly as stop_run does; solwyn-test/deny-stopped returns a run_stopped denial on each call without the plane holding a stop.

Inspecting what the SDK sent

The plane records every request it routes, as the SDK's own Pydantic wire models, in arrival order. Routing is what gets recorded: a request answered by an endpoint refusal (refuse_checks -- including its 422 unknown-model shape -- refuse_leases, or read_only) is turned away before the recording handler runs, and a request to a path the plane does not route lands in plane.unmatched_requests instead.

Records contain what the SDK sends -- token counts, model, provider, run id, tags, receipt fields -- and nothing else. There is no cost field on a MetadataEvent because the SDK never computes cost.

plane.reset_recording() clears the recording lists and nothing else: scripted denials, stops, open windows, held leases, and reservations all survive it. Because the plane is a plain object, the isolation primitive is a new FakeControlPlane() per test.

For contract-style tests that need the raw status, headers, and response model, plane.handle(method, path, body) answers a request directly. The routed paths are GET /health and POST on /api/v1/budgets/check, /api/v1/budgets/confirm, /api/v1/budgets/lease, /api/v1/budgets/lease/renew, /api/v1/budgets/lease/surrender, /api/v1/metadata/ingest, /api/v1/untracked-surfaces, and /api/v1/projects/*/providers/breaker-reports.

The plane is thread-safe. Reporter flush threads and lease renewal workers hit it concurrently in a normal test, and every effect is decided under a lock before any delay is served. State is per process: a forked child works on a copy, and the parent's recordings never see the child's traffic.

Opt-in pytest fixtures

Fixtures never auto-register. Enable them in the test module (or your own conftest.py) and request both fixtures when you want to script and inspect the same plane. solwyn_test_client is the normal Solwyn wrapper around a private denial-only dispatch sentinel; it does not simulate provider responses.

import pytest
from solwyn import BudgetExceededError

pytest_plugins = ["solwyn.testing.pytest_plugin"]

def test_denial_fixture(solwyn_control_plane, solwyn_test_client):
    with pytest.raises(BudgetExceededError):
        solwyn_test_client.chat.completions.create(
            model="solwyn-test/deny", messages=[]
        )
    assert len(solwyn_control_plane.checks) == 1

The plugin has no pytest11 entry point, so installing the SDK never changes the behavior of a test suite that did not ask for it.

The contract pack

solwyn.testing.contract is the pack the SDK uses to prove that the double and the live API answer the same wire shapes. It is lane-agnostic: each function takes an httpx.Client and an API key, and raises AssertionError (with a bounded body preview) on the first mismatch. Against the double, the client is httpx.Client(transport=plane.transport, base_url=plane.api_url).

The pack does not arm scenarios for you. Before assert_check_contract, queue the four denials in order (plane.deny_next(period=...) and plane.deny_run(...)); before assert_lease_contract, arm one refuse_leases(status=409, code="lease_holder_cap_exceeded", requests=1) window and one deny_next(period="monthly", scope="lease"); before assert_run_control_contract, call plane.stop_run(stopped_run_id). Every function mutates state on whichever plane it talks to, so run the pack against a fresh plane -- or, in the live lane, against disposable projects.

Bringing your own transport

FakeControlPlane is built on a public seam you can use directly: Solwyn(..., control_plane_transport=...) and AsyncSolwyn(..., control_plane_transport=...) accept any object implementing solwyn.ControlPlaneTransport. The protocol has exactly two members: a plain-callable handle_request and a coroutine handle_async_request. Sync components need only the first; async components need both, because they perform blocking drains at interpreter exit. An httpx.MockTransport qualifies for the sync client.

The SDK never closes an injected transport -- not on close() and not at interpreter exit -- and reuses the same instance for normal traffic, interpreter-exit delivery, lease surrender, and fork recovery. A stateful transport you inject must therefore be fork-safe itself.

Limits

• Scenario errors are guaranteed loud only through wrap, wrap_async, or a raw httpx client. A BudgetEnforcer you construct directly converts transport exceptions into its fail-open outage posture instead.
• refuse_leases(409, "lease_holder_cap_exceeded") refuses grants only.
• Stops are stickier than denials: reset_recording() and clear_denials() leave them in place; only clear_stop() lifts one.
• Unmatched paths answer 404 silently unless you use the solwyn_control_plane fixture or inspect plane.unmatched_requests.
• price_hints is the one price-adjacent knob, and it is a passthrough: the plane serves the mapping you construct it with, only to requests that opted in. Nothing is derived from it.

Related

• Run control -- what RunStoppedError means and how local velocity detection works
• Budget enforcement -- the posture and outage ladder the recipes above reproduce
• Error handling -- exception hierarchy and the silent-error classes worth asserting on
• Spend delivery -- how confirms, ingest events, and denial receipts leave your process
• Coverage controls -- on_unmetered and the advisory reports plane.untracked_reports records
• Troubleshooting -- shutdown and teardown issues in async test suites