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

Advanced¶

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 preflights the whole draft against live kernel state (duplicate services / health checks, breaking or conflicting event schemas, malformed names) and only then applys 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
• Python API overview