# `sandloom.sdk.testing`

The testing kit: every helper of the framework's test harness and the 1.0 names.

Public API 1.0 (spec §2.2, Appendix C).

`from sandloom.sdk.testing import <name>`. Signatures in a terminal: `sandloom docs <name>`. Beginner names are the ones generated code and the examples use; any other name is advanced (mark its import line `# sandloom: advanced <reason>` in generated code and examples).

## Start with (beginner)

| Name | Kind | Summary |
| --- | --- | --- |
| [`as_user`](<#as_user>) | function | `(user_id, headers)` with a freshly minted access token. |
| [`deliver`](<#deliver>) | async function | Deliver one event to every consumer of its channel; returns how many. |
| [`fake_access_fact`](<#fake_access_fact>) | function | Answer the access fact `name` with `value` inside the block. |
| [`fake_contract`](<#fake_contract>) | function | Serve contract `ref` from `impl` for the block: |
| [`published`](<#published>) | function | The events of `event`'s channel published by the code under test: |

## Advanced

| Name | Kind | Summary |
| --- | --- | --- |
| [`apply_worker_env`](<#apply_worker_env>) | function | Point `environ` (default `os.environ`) at this worker's backends. |
| [`assert_contract_conformance`](<#assert_contract_conformance>) | async function | Assert `contract_name` behaves identically locally and over RPC. |
| [`assert_route_snapshot`](<#assert_route_snapshot>) | function | Fail unless each module's routes equal its committed `routes.snapshot.json`. |
| [`bind_contract_session`](<#bind_contract_session>) | function | Run contract calls in this block on `session` (TEST SEAM). |
| [`bind_test_uow`](<#bind_test_uow>) | function | Bind `session` as the active UoW for `database` in this context. |
| [`block_mark_key`](<#block_mark_key>) | function | The cache key of `user_id`'s subject-state record (the block lives there). |
| [`boot_module_alone`](<#boot_module_alone>) | async function | `create_app` with only `module` loaded, then run its lifespan. |
| [`BootOutcome`](<#bootoutcome>) | class | What booting one module alone did. |
| [`capture_effects`](<#capture_effects>) | function | Record `publish_event` / `record_audit` and typed `Event.publish()` calls made from `modules`. |
| [`Captured`](<#captured>) | callable | dict() -\> new empty dictionary dict(mapping) -\> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable)... |
| [`chain_coverage_problems`](<#chain_coverage_problems>) | function | Tables / indexes a model declares that its owning chain does not create. |
| [`check_contract_conformance`](<#check_contract_conformance>) | async function | Run every case locally and over RPC; return the mismatches. |
| [`committing_uow`](<#committing_uow>) | function | Run the block in a REAL Unit of Work that commits, like `db_session()`. |
| [`ContractCase`](<#contractcase>) | class | One call to compare: `method(*args, **kwargs)` under `tenant`. |
| [`ContractConformanceError`](<#contractconformanceerror>) | exception | The local and the RPC path disagreed for at least one case. |
| [`email_of`](<#email_of>) | function | The deterministic `email` claim minted for a test user. |
| [`FakeClock`](<#fakeclock>) | class | The runtime's injectable clock (naive UTC). |
| [`flush_owned_redis`](<#flush_owned_redis>) | async function | `FLUSHDB` a test Redis database this run owns, keeping its marker. |
| [`get_db`](<#get_db>) | function | Request-scoped DB session governed by a UnitOfWork (the canonical path). |
| [`Headers`](<#headers>) | callable | dict() -\> new empty dictionary dict(mapping) -\> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable)... |
| [`in_process_rpc_client`](<#in_process_rpc_client>) | function | An HTTP client whose requests hit `core`'s internal RPC routes in-process (authenticated and bound under `settings`, default the core's). |
| [`InMemoryCache`](<#inmemorycache>) | class | Dict-backed `CacheProvider` (incl. `cas` and an atomic `set_max`) for unit tests. |
| [`keyed_rpc_settings`](<#keyed_rpc_settings>) | function | A copy of `cfg` keyed as `service_id` calling ITSELF. |
| [`migrate_module`](<#migrate_module>) | function | Move `package`'s own chain in `url` to `revision`. |
| [`migration_chains`](<#migration_chains>) | function | The core chain plus the chain of every loaded module that has one. |
| [`MigrationChain`](<#migrationchain>) | class | One Alembic chain: the kernel's core chain or a module's own. |
| [`model_tables`](<#model_tables>) | function | `{table: owning chain}` for every ORM table of the core and the loaded modules — the owner is the package that DECLARES the model. |
| [`ModuleBootError`](<#modulebooterror>) | exception | The module started alone, but logged errors while doing so. |
| [`ModuleRegistrationDraft`](<#moduleregistrationdraft>) | class | A buffer of one module's registrations, applied atomically on commit. |
| [`open_saga_harness`](<#open_saga_harness>) | function | The fixture body: point `default` at the test database, yield a harness. |
| [`Outcome`](<#outcome>) | class | A normalized call outcome (comparable across the two paths). |
| [`override_instance`](<#override_instance>) | function | Configure `capability`/`name` with `driver` + `options` in the block. |
| [`probe`](<#probe>) | function | One readiness check: `True` once THIS spawn answers with its nonce. |
| [`record_revocation_checks`](<#record_revocation_checks>) | function | Record the `sub` of every access-token revocation check. |
| [`recording_uow`](<#recording_uow>) | function | Bind an in-memory UoW that records audits/events (unit tests). |
| [`RecordingSession`](<#recordingsession>) | class | Stand-in UoW session for `recording_uow` (only `info`). |
| [`register_event_schemas`](<#register_event_schemas>) | function | Register schemas for `channels` (a list = permissive object schemas). |
| [`registration_surface`](<#registration_surface>) | function | Run `register` against a recording `ModuleAPI` stand-in. |
| [`RegistrationSurface`](<#registrationsurface>) | class | What one `register(core)` call registered, in call order. |
| [`reload_test_env`](<#reload_test_env>) | function | `load_dotenv(path, override=True)` then re-apply the worker rewrite. |
| [`ReservedPort`](<#reservedport>) | class | A loopback port held by a listening socket owned by the test process. |
| [`revocation_mark_key`](<#revocation_mark_key>) | function | The cache key of `user_id`'s subject-state record. |
| [`route_policy`](<#route_policy>) | function | The policy the framework applies to `route` (resolved live). An app asserts deny-by-default with it: |
| [`RoutePolicy`](<#routepolicy>) | class | What the framework applies to one route (ADR-0009). |
| [`rpc_client_for`](<#rpc_client_for>) | function | (client, base\_url) of the RPC path for `transport` (see module doc). |
| [`rpc_env`](<#rpc_env>) | function | `rpc_settings` as environment variables (maps JSON-encoded). |
| [`rpc_keys`](<#rpc_keys>) | callable | Fresh keys for `service_id` (generated once per process). |
| [`rpc_proxy_for`](<#rpc_proxy_for>) | function | A generated RPC proxy for `contract_name` served by `core` via `client`. |
| [`rpc_request_headers`](<#rpc_request_headers>) | function | Authenticated headers of ONE raw internal-RPC request (tests). |
| [`rpc_response_headers`](<#rpc_response_headers>) | function | The binding headers a provider keyed with `settings` puts on its response to `request_headers` (tests faking a provider's answer). |
| [`rpc_settings`](<#rpc_settings>) | function | `Settings` field values keying `service_id` for internal RPC. |
| [`RpcKeys`](<#rpckeys>) | class | One service's ephemeral internal-RPC keys (tests only). |
| [`run_on_commit`](<#run_on_commit>) | async function | Treat a test-owned UoW session as committed: run its `before_commit` checks (a failing one propagates and no `on_commit` callback runs —... |
| [`run_sql`](<#run_sql>) | function | Run `sql` in its own committed transaction; the rows it returned. |
| [`SagaHarness`](<#sagaharness>) | class | The `saga_harness` fixture's object (see the module docstring). |
| [`SagaRun`](<#sagarun>) | class | What one `run()` / `drive()` observed. |
| [`served_rpc`](<#served_rpc>) | function | Serve `core`'s internal RPC routes (or `app`) on a loopback socket. |
| [`set_environment`](<#set_environment>) | function | Run the rest of the test with `settings.environment == name`. |
| [`SimulatedCrash`](<#simulatedcrash>) | exception | The process "died": the transition's UoW rolls back (not an Exception, so the step classification never sees it). |
| [`spawn_served`](<#spawn_served>) | function | Start `app` (`"module:attr"`) in a child serving on `port`'s socket. |
| [`SpawnedServer`](<#spawnedserver>) | class | A child serving on a `ReservedPort`, identified by its nonce. |
| [`SpawnedServerError`](<#spawnedservererror>) | exception | The spawned server exited, or its port is answered by another process. |
| [`stop_revocation_listener`](<#stop_revocation_listener>) | async function | Drop every per-process copy of a subject's state (test teardown). |
| [`tables_created_by`](<#tables_created_by>) | function | Tables a chain's revisions create (`op.create_table(<name>, ...)`). |
| [`temporary_database`](<#temporary_database>) | function | Create an empty database on the test server; yield its sync URL. |
| [`TestResourceOwnershipError`](<#testresourceownershiperror>) | exception | The test database / Redis index belongs to another run. |
| [`UNIQUE`](<#unique>) | constant |  |
| [`upgrade_core_chain`](<#upgrade_core_chain>) | function | Apply the core chain (outbox / audit / inbox / command log) to `url`. |
| [`use_security_cache`](<#use_security_cache>) | function | Resolve the kernel's revocation / block state to `provider`. |
| [`wait_revocation_listener_subscribed`](<#wait_revocation_listener_subscribed>) | async function | Return once no per-process copy of a subject's state can be stale. |
| [`worker_env_values`](<#worker_env_values>) | function | `dotenv_values(path)` (`None` dropped), the `SANDLOOM_TEST_<VAR>` overrides applied, rewritten for this worker. Missing file -\> `{}` base. |

## Beginner names

### `as_user`

```
def as_user(
    user_id: UUID | None = None,
    tenant_id: UUID | None = None,
    *,
    email: str | None = '',
    email_verified: bool | None = True,
    tenant: bool = True,
    **claims: Any,
) -> tuple[UUID, dict[str, str]]
```

`(user_id, headers)` with a freshly minted access token.

Today's `make_user_headers`: `user_id` / `tenant_id` default to random UUIDs; the token carries `tenant_id` and the signed `tenant_ids` allowlist; `tenant=False` mints a tenant-less token; any extra keyword becomes a claim (`roles=["admin"]`):

```
_, headers = as_user(tenant_id=uuid4())
await client.post("/api/v1/notes/", json={...}, headers=headers)
```

### `deliver`

```
async def deliver(
    event: str | Event,
    payload: dict[str, Any] | None = None,
    *,
    times: int = 1,
    event_id: str | None = None,
    session: Any = None,
    core: Any = None,
) -> int
```

Deliver one event to every consumer of its channel; returns how many.

Today's `deliver_event` (same parameters and result) that also takes an `Event` instance. The envelope id is stable across `times` (a redelivery): a `durable=True` consumer applies it once:

```
await deliver(NoteCreated(note_id=n, tenant_id=t), times=2)
```

### `fake_access_fact`

```
def fake_access_fact(
    name: str,
    value: bool | Callable[[Any], Any] = True,
    *,
    deny_category: str | None = None,
    core: Any = None,
) -> Iterator[None]
```

Answer the access fact `name` with `value` inside the block.

### `fake_contract`

```
def fake_contract(
    ref: ContractRef[P] | str,
    impl: object,
) -> AbstractContextManager[None]
```

Serve contract `ref` from `impl` for the block:

```
with fake_contract(STATS_QUOTA_V1, AllowAll()):
    response = await client.post(...)
```

### `published`

```
def published(event: type[E] | str) -> list[E]
```

The events of `event`'s channel published by the code under test:

```
assert len(published(NoteCreated)) == 1
```

With a channel name, every recorded event of that channel.

## Advanced names

### `apply_worker_env`

```
def apply_worker_env(
    environ: MutableMapping[str, str] | None = None,
    *,
    create_database: bool = True,
) -> dict[str, str]
```

Point `environ` (default `os.environ`) at this worker's backends.

Idempotent: a value this function already wrote is left alone, a value put back by a `.env.test` reload is rewritten again. Creates the worker database on first use (`create_database`). Returns what changed.

### `assert_contract_conformance`

```
async def assert_contract_conformance(
    core: Any,
    contract_name: str,
    cases: Sequence[ContractCase],
    *,
    transport: str = 'asgi',
) -> None
```

Assert `contract_name` behaves identically locally and over RPC.

### `assert_route_snapshot`

```
def assert_route_snapshot(app: Any, *modules: str) -> None
```

Fail unless each module's routes equal its committed `routes.snapshot.json`.

The module's approval of its routes' resolved policy (SANDLOOM-T001): the snapshot lists method, path, auth and policy of every route the module mounts (kernel-mounted routes excluded). Regenerate with `sandloom routes snapshot --write` and review the diff. With no `modules` every loaded module is checked. A module test pins its own:

```
def test_routes_are_approved() -> None:
    from sandloom.main import app

    assert_route_snapshot(app, "sandloom.notes")
```

### `bind_contract_session`

```
def bind_contract_session(session: Any) -> Iterator[Any]
```

Run contract calls in this block on `session` (TEST SEAM).

The contract runtime normally opens the provider's own Unit of Work. A test that owns an uncommitted transaction (the `get_db` override pattern) binds its session here so contracts read the test's data; the framework binds it automatically on the dependency-override path. Never use it in production code: it reintroduces the shared-transaction semantics the runtime exists to remove.

### `bind_test_uow`

```
def bind_test_uow(
    session: Any,
    *,
    database: str = 'default',
    outbox: OutboxStore | None = None,
    audit: AuditStore | None = None,
) -> Iterator[Any]
```

Bind `session` as the active UoW for `database` in this context.

Uses the relational (SQLAlchemy) outbox/audit stores unless explicit ones are given. The caller owns the transaction (commit/rollback as the test needs); nothing is dispatched to the event bus.

### `block_mark_key`

```
def block_mark_key(user_id: Any) -> str
```

The cache key of `user_id`'s subject-state record (the block lives there).

Writing `blocked_value` to it blocks the subject on the very next authenticated request, exactly like a committed `block_user` (minus its audit / revocation / event); deleting it unblocks (`unblock_user` never deletes: it swaps the value for a revocation).

### `boot_module_alone`

```
async def boot_module_alone(module: str, *, lifespan: bool = True) -> BootOutcome
```

`create_app` with only `module` loaded, then run its lifespan.

Returns `started=True` when the app built and (with `lifespan`) started and shut down cleanly; `started=False` with the missing contracts when the loader refused with `ContractBindingError`. Anything else raises (an error logged during the lifespan: `ModuleBootError`). Needs the test database / cache the harness configures (the `db_engine` fixture builds the schema).

### `BootOutcome`

```
class BootOutcome(
    module: str,
    started: bool,
    missing_contracts: frozenset[str] = <factory>,
) -> None
```

What booting one module alone did.

### `capture_effects`

```
def capture_effects(monkeypatch: Any, *modules: ModuleType) -> Captured
```

Record `publish_event` / `record_audit` and typed `Event.publish()` calls made from `modules`.

Patches the two names in each module (where the service imported them from `sandloom.core.di`), so the calls never reach a UoW: events are recorded as `(channel, payload)` and audits as `(action, kwargs)` — the exact keyword arguments, so `kwargs["required"]` can be asserted. Use it when the service runs on a session that is not bound as the UoW (committed sessions of a race test, a direct service call).

### `Captured`

```
def Captured(*args, **kwargs)
```

dict() -\> new empty dictionary dict(mapping) -\> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -\> new dictionary initialized as if via: d = {} for k, v in iterable: d\[k\] = v dict(\*\*kwargs) -\> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

### `chain_coverage_problems`

```
def chain_coverage_problems(core: Any = None) -> list[str]
```

Tables / indexes a model declares that its owning chain does not create.

The harness builds tables from the models, production from the chains: a table or index on the model only passes the suite and is missing in production (SANDLOOM-T002). Each problem names the table, the model's owning module and that module's chain directory.

### `check_contract_conformance`

```
async def check_contract_conformance(
    core: Any,
    contract_name: str,
    cases: Sequence[ContractCase],
    *,
    transport: str = 'asgi',
) -> list[tuple[ContractCase, Outcome, Outcome]]
```

Run every case locally and over RPC; return the mismatches.

### `committing_uow`

```
def committing_uow(
    session_factory: Any,
    *,
    database: str = 'default',
) -> AsyncIterator[Any]
```

Run the block in a REAL Unit of Work that commits, like `db_session()`.

`session_factory` is an `async_sessionmaker` on the test engine (e.g. `async_sessionmaker(db_engine, expire_on_commit=False)`). The session is bound as the UoW (outbox / audit rows join its transaction); on a clean exit it COMMITS and then runs the queued `on_commit` callbacks; an exception rolls back and skips them — the request / worker sequence, without the post-commit bus dispatch:

```
with pytest.raises(Boom):
    async with committing_uow(sessions) as db:
        await Service.remove(..., db)
        raise Boom          # rolled back: no post-commit effect ran
```

### `ContractCase`

```
class ContractCase(
    method: str,
    args: tuple[Any, ...] = (),
    kwargs: Mapping[str, Any] = <factory>,
    tenant: UUID | None = None,
    label: str = '',
) -> None
```

One call to compare: `method(*args, **kwargs)` under `tenant`.

#### `ContractCase.describe`

```
def describe(self) -> str
```

### `ContractConformanceError`

```
class ContractConformanceError(AssertionError)
```

The local and the RPC path disagreed for at least one case.

### `email_of`

```
def email_of(user_id: uuid.UUID) -> str
```

The deterministic `email` claim minted for a test user.

### `FakeClock`

```
class FakeClock(start: datetime | None = None) -> None
```

The runtime's injectable clock (naive UTC).

#### `FakeClock.advance`

```
def advance(self, delta: timedelta) -> datetime
```

#### `FakeClock.advance_to`

```
def advance_to(self, when: datetime) -> datetime
```

### `flush_owned_redis`

```
async def flush_owned_redis(client: Any, *, owner: str | None = None) -> None
```

`FLUSHDB` a test Redis database this run owns, keeping its marker.

`client` is an asyncio Redis client (bytes or decoded responses). Raises `TestResourceOwnershipError` when another run marked it.

### `get_db`

```
def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]
```

Request-scoped DB session governed by a UnitOfWork (the canonical path).

The whole request runs in one transaction: commit when the request succeeds, rollback on any exception (including HTTPException). Services must use `flush()` — never `commit()` — so the UoW alone owns the transaction boundary. Events published via `publish_event` are written to the transactional outbox in the same transaction and dispatched to the bus only after a successful commit.

Test overrides registered for the raw provider (`sandloom.core._internal.db_provider.get_db`) bypass the UoW entirely: the test owns the session and its transaction.

### `Headers`

```
def Headers(*args, **kwargs)
```

dict() -\> new empty dictionary dict(mapping) -\> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -\> new dictionary initialized as if via: d = {} for k, v in iterable: d\[k\] = v dict(\*\*kwargs) -\> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

### `in_process_rpc_client`

```
def in_process_rpc_client(core: Any, settings: Any = None) -> httpx.AsyncClient
```

An HTTP client whose requests hit `core`'s internal RPC routes in-process (authenticated and bound under `settings`, default the core's).

### `InMemoryCache`

```
class InMemoryCache(CacheProvider)(*, broken: bool = False) -> None
```

Dict-backed `CacheProvider` (incl. `cas` and an atomic `set_max`) for unit tests.

TTLs are accepted and ignored. `broken=True` makes every operation raise `ConnectionError` (a cache outage). Install it where the code under test resolves the cache, e.g. `monkeypatch.setattr(cache_store, "cache", lambda name="default": mem)` (`sandloom.core.idempotency.cache_store`) for the idempotency / contract-dedupe store.

#### `InMemoryCache.features`

```
def features(self) -> frozenset[str]
```

Optional capability features THIS instance actually supports.

A subset of the owning `CapabilitySpec`'s declared features. A module may require a feature; the registry resolves against the driver's advertised set, so it can refuse a backend that cannot do what the module needs (e.g. a graph store without `cypher`). Default: none.

#### `InMemoryCache.get`

```
async def get(self, key: str) -> str | None
```

Return the value for `key`, or `None` if absent/expired.

#### `InMemoryCache.set`

```
async def set(self, key: str, value: str, *, ttl_seconds: int | None = None) -> None
```

Store `value` under `key` (overwriting), optionally with a TTL.

#### `InMemoryCache.add`

```
async def add(self, key: str, value: str, *, ttl_seconds: int | None = None) -> bool
```

Atomically store `value` ONLY if `key` is absent (SET NX).

Returns `True` when this call created the key, `False` when it already existed. The atomic primitive behind reserve-or-conflict guards (idempotency keys).

#### `InMemoryCache.delete`

```
async def delete(self, key: str) -> None
```

Remove `key` (no error if it does not exist).

#### `InMemoryCache.incr`

```
async def incr(self, key: str, *, ttl_seconds: int | None = None) -> int
```

Atomically increment the integer at `key` and return the new value.

When the key is first created (result `1`) and `ttl_seconds` is given, the expiry is set in the same atomic step — the primitive behind fixed-window rate limiting (no counter can leak without a TTL).

#### `InMemoryCache.compare_and_set`

```
async def compare_and_set(
    self,
    key: str,
    expected: str,
    value: str,
    *,
    ttl_seconds: int | None = None,
) -> bool
```

Atomically replace `key` with `value` ONLY if it equals `expected`.

Returns `True` when swapped, `False` when the key is absent or holds something else. The primitive behind owner-checked state transitions (an idempotency reservation must only be finalised by its owner). Optional: drivers advertise it via the `cas` feature; the default raises `NotImplementedError` so callers can detect its absence.

#### `InMemoryCache.compare_and_delete`

```
async def compare_and_delete(self, key: str, expected: str) -> bool
```

Atomically delete `key` ONLY if it equals `expected` (`cas`).

#### `InMemoryCache.set_max`

```
async def set_max(self, key: str, value: str, *, ttl_seconds: int | None = None) -> str
```

Atomic monotonic write.

Reads through `get` first, so a subclass's fault injection (an outage simulated in `get`) applies, then decides and writes on the store with no await in between — atomic even when `get` yields.

#### `InMemoryCache.health`

```
async def health(self) -> HealthResult
```

Liveness probe for this instance. Default: always healthy.

### `keyed_rpc_settings`

```
def keyed_rpc_settings(
    cfg: Any,
    contracts: Iterable[str],
    service_id: str = 'sandloom-conformance',
) -> Any
```

A copy of `cfg` keyed as `service_id` calling ITSELF.

The in-process RPC harness serves and calls with it: one service id on both sides, allowed exactly `contracts`.

### `migrate_module`

```
def migrate_module(
    url: str,
    package: str,
    revision: str = 'head',
    *,
    downgrade: bool = False,
    chain: str | None = None,
) -> None
```

Move `package`'s own chain in `url` to `revision`.

`package` is the module's import name (`"sandloom_billing"`); its chain lives in `<package>/migrations` with the version table `alembic_version_<chain>` (`chain` defaults to the package's last component, like `ModuleDescriptor.migrations_chain`). `downgrade=True` walks DOWN to `revision` instead of up.

### `migration_chains`

```
def migration_chains(core: Any = None) -> list[MigrationChain]
```

The core chain plus the chain of every loaded module that has one.

Derived from the loaded descriptors (`migrations_path`, `migration_chain_key`), so a new module's chain is covered without editing any list. `core` defaults to the composition root's.

### `MigrationChain`

```
class MigrationChain(
    key: str,
    module: str | None,
    script_location: Path,
    version_table: str,
) -> None
```

One Alembic chain: the kernel's core chain or a module's own.

#### `MigrationChain.versions`

```
versions  (property)
```

### `model_tables`

```
def model_tables(core: Any = None) -> dict[str, MigrationChain]
```

`{table: owning chain}` for every ORM table of the core and the loaded modules — the owner is the package that DECLARES the model.

### `ModuleBootError`

```
class ModuleBootError(AssertionError)
```

The module started alone, but logged errors while doing so.

### `ModuleRegistrationDraft`

```
class ModuleRegistrationDraft() -> None
```

A buffer of one module's registrations, applied atomically on commit.

Module loading is transactional: instead of mutating the kernel as `register()` runs (then snapshotting/restoring private kernel state if it fails — fragile and tied to FastAPI/ServiceRegistry internals), every write a module makes is appended here. `commit` first `preflight`s the whole draft against live kernel state (duplicate services / health checks, breaking or conflicting event schemas, malformed names) and only then `apply`s it. If `register()` raises first, commit never runs; if preflight fails, NOTHING is applied. A half-registered module can no longer leave an orphan service or schema in the live inventory.

#### `ModuleRegistrationDraft.commit`

```
def commit(self, facade: CoreFacade) -> None
```

Validate the whole draft, then apply it (atomic: all or nothing).

#### `ModuleRegistrationDraft.preflight`

```
def preflight(self, facade: CoreFacade) -> None
```

Check the draft against live state; raise before applying anything.

#### `ModuleRegistrationDraft.apply`

```
def apply(self, facade: CoreFacade) -> None
```

Apply every buffered registration to `facade` (routers last).

Runs only after `preflight` passed, so the validated operations (services, schemas, health) should not fail partway. As a strict all-or-nothing backstop the whole apply is wrapped in a snapshot/restore: if ANY step still raises (e.g. `include_router` hits a FastAPI route conflict preflight cannot foresee), the kernel is rewound to exactly its pre-apply state so a half-registered module leaves no orphan service, route or schema behind.

### `open_saga_harness`

```
def open_saga_harness(core: Any) -> AsyncIterator[SagaHarness]
```

The fixture body: point `default` at the test database, yield a harness.

### `Outcome`

```
class Outcome(ok: bool, value: Any = None, error: tuple[Any, ...] | None = None) -> None
```

A normalized call outcome (comparable across the two paths).

### `override_instance`

```
def override_instance(
    core: Any,
    capability: str,
    name: str = 'default',
    *,
    driver: str,
    **options: Any,
) -> AsyncIterator[Any]
```

Configure `capability`/`name` with `driver` + `options` in the block.

Yields the core's infrastructure registry. On exit the provider built for the override (if any) is closed and the previous instance configuration and provider are put back — so a lazily built, process-wide provider (e.g. the `default` engine) survives the test untouched:

```
async with override_instance(core, "cache", driver="redis",
                             url=os.environ["CACHE_URL"]):
    ...
```

### `probe`

```
def probe(server: SpawnedServer, timeout: float = 1.0) -> bool
```

One readiness check: `True` once THIS spawn answers with its nonce.

`False` while it is not serving yet. Raises `SpawnedServerError` when the child has exited (e.g. a startup or bind failure) or when the port answers with anything but this spawn's nonce — a different process is serving there, and no request must be sent to it.

### `record_revocation_checks`

```
def record_revocation_checks(monkeypatch: pytest.MonkeyPatch) -> list[str]
```

Record the `sub` of every access-token revocation check.

Wraps the kernel's single revocation lookup (the one every framework token guard reaches through the per-request memo): the returned list grows by one subject per actual cache lookup, so a request whose guards share the memo adds exactly one entry.

### `recording_uow`

```
def recording_uow(*, database: str = 'default') -> Iterator[RecordingSession]
```

Bind an in-memory UoW that records audits/events (unit tests).

### `RecordingSession`

```
class RecordingSession(
    info: dict[str, Any] = <factory>,
    audits: list[dict[str, Any]] = <factory>,
    events: list[tuple[str, dict[str, Any], str]] = <factory>,
) -> None
```

Stand-in UoW session for `recording_uow` (only `info`).

#### `RecordingSession.audit_actions`

```
def audit_actions(self) -> list[str]
```

### `register_event_schemas`

```
def register_event_schemas(
    channels: Mapping[str, dict[str, Any] | None] | Iterable[str],
    *,
    target: Any = None,
) -> None
```

Register schemas for `channels` (a list = permissive object schemas).

`target` is where they go: a `CoreFacade` (`register_event_schema`), an event bus (its schema registry) or a schema registry; default: the active application's event bus. Re-registering an identical schema is a no-op.

### `registration_surface`

```
def registration_surface(
    register: Callable[[Any], Any],
    *,
    api: Any | None = None,
) -> RegistrationSurface
```

Run `register` against a recording `ModuleAPI` stand-in.

By default nothing is registered anywhere (a pure stub: read methods such as `get_service` return `None`). Pass `api=ModuleAPI(core)` to also forward every call to a real core. Side effects of `register` itself (e.g. installing a process-wide client) still happen — restore them in the test if they matter.

### `RegistrationSurface`

```
class RegistrationSurface(descriptor: Any, calls: list[Call] = <factory>) -> None
```

What one `register(core)` call registered, in call order.

#### `RegistrationSurface.of`

```
def of(self, method: str) -> list[Call]
```

The recorded calls of one `ModuleAPI` method, in order.

### `reload_test_env`

```
def reload_test_env(path: str | os.PathLike[str]) -> None
```

`load_dotenv(path, override=True)` then re-apply the worker rewrite.

Use instead of a bare `load_dotenv(..., override=True)` in tests: the reload would otherwise put a parallel worker back on the shared base database / Redis.

### `ReservedPort`

```
class ReservedPort(host: str = '127.0.0.1', backlog: int = 128) -> None
```

A loopback port held by a listening socket owned by the test process.

While open, no other process can bind the port; children inherit the very socket. A test that must make the port REFUSE connections for a while (a "service down" window) `close` it and `reopen` the same port afterwards — that re-bind fails loudly (`OSError`) if another process took the port meanwhile, it never silently shares it.

#### `ReservedPort.reopen`

```
def reopen(self) -> None
```

Re-reserve the SAME port after `close` (no-op while open).

#### `ReservedPort.url`

```
url  (property)
```

#### `ReservedPort.fileno`

```
def fileno(self) -> int
```

#### `ReservedPort.closed`

```
closed  (property)
```

#### `ReservedPort.close`

```
def close(self) -> None
```

Release the port (idempotent). Do it after the last child stopped.

### `revocation_mark_key`

```
def revocation_mark_key(user_id: Any) -> str
```

The cache key of `user_id`'s subject-state record.

A revocation is the value `v1:after:<epoch seconds>` (see `revoked_after_value`): the subject's access tokens issued at or before that instant are revoked. The same key holds a block (`block_mark_key`).

### `route_policy`

```
def route_policy(route: Any) -> RoutePolicy
```

The policy the framework applies to `route` (resolved live). An app asserts deny-by-default with it:

```
for route in app.routes:
    policy = route_policy(route)
    assert policy.auth != "public" or policy.public_reason
```

### `RoutePolicy`

```
class RoutePolicy(
    always_strict: bool,
    declares_budget: bool,
    staleness_budget: float,
    auth: str = 'user',
    public_reason: str | None = None,
    effect: str = 'read',
    risk: tuple[str, ...] = (),
) -> None
```

What the framework applies to one route (ADR-0009).

`auth` is `"user"`, `"service"` or `"public"` (deny-by-default: a public route always carries `public_reason`). `effect` is `"read"` or `"write"` (by method) and `risk` the risks resolved from their sources (an admin gate, a money or security effect, `sensitive()`), sorted. Every route is strict: the subject-state record is read live on every authenticated request (`always_strict` is True, `declares_budget` False and `staleness_budget` 0.0 for every route).

### `rpc_client_for`

```
def rpc_client_for(
    core: Any,
    transport: str,
    settings: Any = None,
) -> AsyncIterator[tuple[Any, str]]
```

(client, base\_url) of the RPC path for `transport` (see module doc).

### `rpc_env`

```
def rpc_env(
    service_id: str,
    *,
    peers: Iterable[str] = (),
    allowed_calls: Mapping[str, Any] | None = None,
) -> dict[str, str]
```

`rpc_settings` as environment variables (maps JSON-encoded).

### `rpc_keys`

```
def rpc_keys(service_id: str) -> RpcKeys
```

Fresh keys for `service_id` (generated once per process).

### `rpc_proxy_for`

```
def rpc_proxy_for(
    core: Any,
    contract_name: str,
    client: Any,
    base_url: str = 'http://sandloom-conformance',
    *,
    settings: Any = None,
) -> Any
```

A generated RPC proxy for `contract_name` served by `core` via `client`.

`settings` (default: the core's) key the call; the binding pins their service id — a core calling itself is its own provider.

### `rpc_request_headers`

```
def rpc_request_headers(
    settings: Any,
    target: str,
    body: bytes = b'',
    *,
    http_method: str = 'POST',
    provider: str | None = None,
    sign: bool = False,
) -> dict[str, str]
```

Authenticated headers of ONE raw internal-RPC request (tests).

`target` is `"<contract>/<method>"` (a call) or `"<contract>"` (the describe GET); the request is MACed with `settings`' pair key towards `provider` (default: `settings`' own service id — a core calling itself) under a fresh nonce, and — with `sign` — also carries the command signature. The bound tenant (`bind_tenant`) travels in the context, as on a real call.

### `rpc_response_headers`

```
def rpc_response_headers(
    settings: Any,
    request_headers: Mapping[str, str],
    status_code: int,
    body: bytes,
) -> dict[str, str]
```

The binding headers a provider keyed with `settings` puts on its response to `request_headers` (tests faking a provider's answer).

The caller is the one the request's pair-MAC header names; the response is keyed with the pair key towards it, exactly as a real provider does once that request verified.

### `rpc_settings`

```
def rpc_settings(
    service_id: str,
    *,
    peers: Iterable[str] = (),
    allowed_calls: Mapping[str, Any] | None = None,
) -> dict[str, Any]
```

`Settings` field values keying `service_id` for internal RPC.

Its own Ed25519 and X25519 keys plus the public keys of `service_id` and every peer (`rpc_keys`), and the `RPC_ALLOWED_CALLS` map. Apply with `monkeypatch.setattr(settings, name, value)` per item, or `settings.model_copy(update=...)`; `rpc_env` gives env names.

### `RpcKeys`

```
class RpcKeys(
    service_id: str,
    kid: str,
    private_key: str,
    public_key: str,
    dh_private_key: str,
    dh_public_key: str,
) -> None
```

One service's ephemeral internal-RPC keys (tests only).

### `run_on_commit`

```
async def run_on_commit(session: Any = None, *, database: str | None = None) -> None
```

Treat a test-owned UoW session as committed: run its `before_commit` checks (a failing one propagates and no `on_commit` callback runs — like a real UoW that refuses to commit), then its `on_commit` callbacks.

`bind_test_uow` / `recording_uow` never commit (the test owns the transaction), so post-commit side effects (e.g. `revoke_user_tokens` queued via `on_commit`) stay pending. Call this where the test treats the work as committed — e.g. after the `yield` of a `get_db` override, which is skipped when the endpoint raised (the request "rolled back"):

```
async def override_get_db():
    with bind_test_uow(test_db):
        yield test_db
    await run_on_commit(test_db)
```

Without `session` it takes the UoW currently bound for `database` (`bind_test_uow` / `recording_uow` / an ambient test UoW); with none bound there is nothing pending — `on_commit` outside a UoW runs at once.

### `run_sql`

```
def run_sql(url: str, sql: str, params: Any = None) -> list[Any]
```

Run `sql` in its own committed transaction; the rows it returned.

### `SagaHarness`

```
class SagaHarness(core: Any) -> None
```

The `saga_harness` fixture's object (see the module docstring).

#### `SagaHarness.register`

```
def register(
    self,
    definition: Saga[Any],
    *,
    module: str = 'saga_tests',
) -> RegisteredSaga
```

Register a definition for this test (checked like at startup).

#### `SagaHarness.fake_contract`

```
def fake_contract(
    self,
    name: str,
    *,
    idempotency_key: str = 'operation_key',
    **methods: Callable[..., Any],
) -> type
```

Register `name` as a command contract made of plain functions.

Each function takes the business arguments plus the idempotency key (`operation_key` unless renamed) and is wrapped in `@contract_method(command=True)` — its calls go through the durable command log like a real provider's. `fake_calls["name.method"]` records the executions (a replay does not execute the body).

#### `SagaHarness.close`

```
def close(self) -> None
```

#### `SagaHarness.fail_step`

```
def fail_step(
    self,
    name: str,
    *,
    times: int = 1,
    error: str = 'transient',
    direction: Direction = 'do',
    status: int | None = None,
    exception: BaseException | None = None,
) -> None
```

Fail the next `times` calls of `name` (`direction`).

`error`: `transient` (503 before the call), `refused` (422 before the call), `timeout` (the provider runs and commits, then the caller sees a timeout — ambiguous), `misconfigured` (401), `conflict` (409 `ContractIdempotencyConflict`) — or pass the `exception` to raise (before the call).

#### `SagaHarness.crash_before`

```
def crash_before(
    self,
    name: str,
    *,
    direction: Direction = 'do',
    times: int = 1,
) -> None
```

The process dies before the step call: nothing ran, the UoW rolls back.

#### `SagaHarness.crash_after`

```
def crash_after(self, name: str, *, direction: Direction = 'do', times: int = 1) -> None
```

The provider committed, then the process died before our commit.

#### `SagaHarness.deliver_twice`

```
def deliver_twice(self, enabled: bool = True) -> None
```

Deliver every nudge twice (the duplicate must be a no-op).

#### `SagaHarness.start`

```
async def start(
    self,
    definition: Saga[Any],
    *,
    saga_id: object,
    input: BaseModel | Mapping[str, Any],
    tenant_id: uuid.UUID | None = None,
) -> SagaHandle
```

`start_saga` in a committing UoW (the caller's business change).

#### `SagaHarness.run`

```
async def run(
    self,
    definition: Saga[Any],
    *,
    saga_id: object | None = None,
    input: BaseModel | Mapping[str, Any],
    tenant_id: uuid.UUID | None = None,
    max_transitions: int = 500,
    stop_when: Callable[[InstanceRow], bool] | None = None,
) -> SagaRun
```

Start `definition` and drive it until it settles (terminal/parked).

#### `SagaHarness.drive`

```
async def drive(
    self,
    definition: Saga[Any],
    saga_id: object,
    *,
    max_transitions: int = 500,
    stop_when: Callable[[InstanceRow], bool] | None = None,
) -> SagaRun
```

Drive an existing instance until it settles (or `stop_when`).

#### `SagaHarness.deliver`

```
async def deliver(self, reg: RegisteredSaga, row: InstanceRow, version: int) -> str
```

Deliver one nudge through the real advance path (its own UoW).

#### `SagaHarness.instance`

```
async def instance(
    self,
    reg: RegisteredSaga | Saga[Any],
    saga_id: object,
) -> InstanceRow
```

#### `SagaHarness.assert_compensations_idempotent`

```
async def assert_compensations_idempotent(
    self,
    definition: Saga[Any],
    *,
    input: BaseModel | Mapping[str, Any] | None = None,
    tenant_id: uuid.UUID | None = None,
) -> None
```

Every compensation runs once per applied step, survives a repeat and accepts "nothing to undo".

For each compensatable step `k`: the next step refuses, so steps `0..k` apply and are compensated (`compensated`, one `undo` per applied step, in reverse). Then each compensation is called again with its key (a replay must succeed) and on a fresh saga id whose forward call never ran (nothing to undo — it must not fail).

#### `SagaHarness.assert_resolves_before_compensating`

```
async def assert_resolves_before_compensating(
    self,
    definition: Saga[Any],
    *,
    input: BaseModel | Mapping[str, Any] | None = None,
    tenant_id: uuid.UUID | None = None,
) -> None
```

An ambiguous step is re-issued (same key) before anything is undone.

For every command step up to the pivot: its first call commits at the provider but times out for the caller; the saga deadline then passes. The runtime must re-issue the forward call (`<step>:do` again, same key: the command log replays) and then compensate it as APPLIED — never compensate blindly, never skip it.

### `SagaRun`

```
class SagaRun(
    definition: Saga[Any],
    instance: InstanceRow,
    calls: list[str],
    keys: list[str],
    steps: list[dict[str, Any]],
    transitions: list[str] = <factory>,
    crashes: int = 0,
) -> None
```

What one `run()` / `drive()` observed.

#### `SagaRun.status`

```
status  (property)
```

#### `SagaRun.park_reason`

```
park_reason  (property)
```

#### `SagaRun.results`

```
results  (property)
```

#### `SagaRun.saga_id`

```
saga_id  (property)
```

#### `SagaRun.step_row`

```
def step_row(self, step: str, direction: Direction = 'do') -> dict[str, Any]
```

#### `SagaRun.assert_keys_derived`

```
def assert_keys_derived(self) -> None
```

Every command call carried `saga:<type>:<id>:<step>:do|undo`.

### `served_rpc`

```
def served_rpc(core: Any, app: Any = None, settings: Any = None) -> AsyncIterator[str]
```

Serve `core`'s internal RPC routes (or `app`) on a loopback socket.

A uvicorn server (httptools, as deployed) runs in the CURRENT event loop and context; yields its base URL.

### `set_environment`

```
def set_environment(
    monkeypatch: pytest.MonkeyPatch,
    name: Literal['dev', 'test', 'prod'],
) -> None
```

Run the rest of the test with `settings.environment == name`.

Behaviour gated on `settings.is_prod` / `is_dev` (startup guards, fail-closed defaults) follows; other settings are left as they are.

### `SimulatedCrash`

```
class SimulatedCrash(BaseException)
```

The process "died": the transition's UoW rolls back (not an Exception, so the step classification never sees it).

### `spawn_served`

```
def spawn_served(
    port: ReservedPort,
    app: str,
    *,
    env: dict[str, str],
    log_level: str = 'warning',
    label: str = '',
    **popen_kwargs: Any,
) -> SpawnedServer
```

Start `app` (`"module:attr"`) in a child serving on `port`'s socket.

### `SpawnedServer`

```
class SpawnedServer(
    proc: subprocess.Popen[bytes],
    url: str,
    nonce: str,
    label: str = '',
) -> None
```

A child serving on a `ReservedPort`, identified by its nonce.

#### `SpawnedServer.stop`

```
def stop(self, timeout: float = 10.0) -> None
```

### `SpawnedServerError`

```
class SpawnedServerError(AssertionError)
```

The spawned server exited, or its port is answered by another process.

### `stop_revocation_listener`

```
async def stop_revocation_listener() -> None
```

Drop every per-process copy of a subject's state (test teardown).

There are none (ADR-0009: every check reads the record live), so this returns at once.

### `tables_created_by`

```
def tables_created_by(chain: MigrationChain) -> set[str]
```

Tables a chain's revisions create (`op.create_table(<name>, ...)`).

Read from the revisions' syntax tree: the name is a string literal, the `table_name=` keyword, or a constant assigned a string literal in the revision (`_TABLE = "x"`; `op.create_table(_TABLE, ...)`).

### `temporary_database`

```
def temporary_database(base_url: str | None = None) -> Iterator[str]
```

Create an empty database on the test server; yield its sync URL.

`base_url` (default: the test `DATABASE_URL`) names the server; the database is dropped (`WITH (FORCE)`) when the block exits.

### `TestResourceOwnershipError`

```
class TestResourceOwnershipError(RuntimeError)
```

The test database / Redis index belongs to another run.

### `UNIQUE`

```
UNIQUE: str = '__sandloom_conformance_unique__'
```

*(no docstring)*

### `upgrade_core_chain`

```
def upgrade_core_chain(url: str) -> None
```

Apply the core chain (outbox / audit / inbox / command log) to `url`.

### `use_security_cache`

```
def use_security_cache(
    monkeypatch: pytest.MonkeyPatch,
    provider: CacheProvider,
) -> CacheProvider
```

Resolve the kernel's revocation / block state to `provider`.

The revocation marks and the emergency block marks are read and written through `sandloom.core.infra.access.cache` at call time; for the rest of the test every such lookup (any instance name) returns `provider`.

### `wait_revocation_listener_subscribed`

```
async def wait_revocation_listener_subscribed(timeout: float = 5.0) -> None
```

Return once no per-process copy of a subject's state can be stale.

Every authenticated request reads the subject-state record live (ADR-0009): there are no per-process copies and no invalidation listener, so this is always already true and returns at once.

### `worker_env_values`

```
def worker_env_values(path: str | os.PathLike[str]) -> dict[str, str]
```

`dotenv_values(path)` (`None` dropped), the `SANDLOOM_TEST_<VAR>` overrides applied, rewritten for this worker. Missing file -\> `{}` base.

## Examples and related guides

- [Testing examples and fixtures](<https://sandloom.dev/docs/reference/testing/index.md>)
- [Python API overview](<https://sandloom.dev/docs/reference/sdk/index.md>)