sandloom.sdk.testing.fixtures¶

The harness fixtures, for from sandloom.sdk.testing.fixtures import * conftests.

The star import also brings the kit's autouse fixture (_sandloom_sdk_kit): published(E) records, and Contract(REF) resolves through the test application, exactly as under the sandloom.sdk.testing.plugin.

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

from sandloom.sdk.testing.fixtures 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).

Advanced¶

Advanced names¶

_sandloom_isolate_module_globals¶

@pytest.fixture
def _sandloom_isolate_module_globals(
    request: pytest.FixtureRequest,
) -> Iterator[None]

Start every test module from the session baseline (the harness app's runtime) and put the process's runtime bindings back to it afterwards.

The baseline is the same whatever ran before: the harness app is built before the first module (harness_app), so a test that resolves the process's active core / infra / event bus without asking for sandloom_app passes alone, first or after any other file. A module-scoped fixture that built its own app (create_app binds it as the process's active core / infra / event bus) keeps it for its own tests only; the next module never resolves db_session() through it.

_sandloom_isolate_process_globals¶

@pytest.fixture
def _sandloom_isolate_process_globals() -> Iterator[None]

Restore the process's runtime bindings after every test.

create_app rebinds the active core, infrastructure registry, event bus and the settings' module lists (sandloom.core.testing.isolation); a test that builds an app must not hand that state to the next test. What a test registers on the bound core (a channel, a service, a startup task, a declared operation, ...) goes too (CoreFacade.restore_registrations), and the process caches it warmed are emptied (PROCESS_CACHES).

_sandloom_release_default_pools¶

@pytest.fixture
def _sandloom_release_default_pools() -> AsyncIterator[None]

Return the default engines' pooled connections after every test.

Pooled asyncpg connections belong to the loop of the test that opened them; pytest-asyncio gives each test its own loop, so a later test reusing one fails with "attached to a different loop". Disposing on the test's own loop closes them cleanly; the engines survive and reconnect next time.

_sandloom_sdk_kit¶

@pytest.fixture
def _sandloom_sdk_kit(
    request: _pytest.fixtures.FixtureRequest,
) -> collections.abc.Iterator[None]

Record published events; bind Contract resolution to the app.

ambient_uow¶

@pytest.fixture
def ambient_uow() -> AsyncIterator[Any]

Bind an in-memory recording UoW for the whole test.

async_client¶

@pytest.fixture
def async_client(client: Any) -> Any

(no docstring)

client¶

@pytest.fixture
def client(
    sandloom_app: Any,
    db: Any,
    request: pytest.FixtureRequest,
) -> AsyncIterator[Any]

HTTP client on the app; every request runs on the test's db.

Shared-session semantics: every request uses the test's ONE db session (bound as the UoW), through DbSession and through db() alike (sandloom.core.testing._uow.SharedRequestUnit), so nothing commits and a request that fails with a 4xx does NOT roll its writes back: the rows a failed request flushed stay visible to the test (and to later requests) until the test ends. Assert rollback behaviour with real_uow_client, whose requests commit or roll back exactly as in production (both paths).

An endpoint that raises an unhandled exception re-raises it here (sandloom.core.testing.server_errors.UnhandledServerError: type, message, fix) instead of returning a bare 500 internal_error; mark a test @pytest.mark.sandloom_allow_server_errors to get the 500 response.

COMPOSITION_ROOT_INI¶

COMPOSITION_ROOT_INI: str = 'sandloom_composition_root'

(no docstring)

db¶

@pytest.fixture
def db(db_engine: Any) -> AsyncIterator[Any]

A session on the test database, rolled back after the test.

db_engine¶

@pytest.fixture
def db_engine(sandloom_app: Any) -> AsyncIterator[Any]

Engine on the test database; every table present and empty.

db_uow¶

@pytest.fixture
def db_uow(db: Any) -> AsyncIterator[Any]

db bound as the UoW for the whole test (outbox / audit rows join it).

ENV_FILE¶

ENV_FILE: str = '.env.test'

(no docstring)

env_starts¶

def env_starts(config: pytest.Config) -> list[Path]

Where to look for .env.test: the rootdir, then each test path.

find_env_file¶

def find_env_file(starts: Iterable[Path]) -> Path | None

The nearest .env.test walking up from each start (first hit wins).

load_test_env¶

def load_test_env(starts: Iterable[Path], *, warn_if_late: bool = False) -> Path

Load the nearest .env.test (sandloom.core.testing.env.load_test_env); a missing file is a pytest usage error.

make_headers¶

@pytest.fixture
def make_headers(
    sandloom_test_env: Path,
) -> Callable[..., tuple[Any, dict[str, str]]]

Factory: make_headers(user_id=None, tenant_id=None, **claims).

real_uow_client¶

@pytest.fixture
def real_uow_client(
    sandloom_app: Any,
    sandloom_core: Any,
    db_engine: Any,
    request: pytest.FixtureRequest,
) -> AsyncIterator[Any]

HTTP client whose requests run through the REAL request UoW.

No get_db override: the app's default database (and cache) are pointed at the test DATABASE_URL / CACHE_URL, so every request commits or rolls back exactly as in production — a 4xx rolls its writes back (committed rows are emptied by the next test's db_engine). An unhandled endpoint error re-raises like with client (@pytest.mark.sandloom_allow_server_errors keeps the 500 response).

revoked_tokens¶

@pytest.fixture
def revoked_tokens(monkeypatch: pytest.MonkeyPatch) -> RevokedTokens

Record revoke_user_tokens(user_id) calls instead of running them.

Keeps API tests deterministic (a token minted in the same second as a revocation would otherwise flip to 401) and lets the test assert who was revoked; enforcement itself is covered by the core auth tests. A RevokedTokens list: .available = False simulates an outage.

RevokedTokens¶

class RevokedTokens(list)(iterable=(), /)

The user ids revoke_user_tokens was called for (revoked_tokens).

Set available = False to simulate an unreachable revocation store: every later call records nothing and reports the mark as NOT written, so an immediate revocation (require_immediate=True) refuses the change (503 / RevocationUnavailableError).

saga_harness¶

@pytest.fixture
def saga_harness(
    sandloom_core: Any,
    db_engine: Any,
) -> AsyncIterator[Any]

Drive the REAL saga protocol synchronously (sandloom.core.testing.sagas).

The app's default database points at the test database; every transition commits for real (rows are emptied by the next test's db_engine). run() / fail_step / crash_after / deliver_twice / clock.advance / fake_contract.

sandloom_app¶

@pytest.fixture
def sandloom_app(
    pytestconfig: pytest.Config,
    sandloom_test_env: Path,
) -> Any

The application's ASGI app, built by its composition root.

sandloom_cache¶

@pytest.fixture
def sandloom_cache(sandloom_core: Any) -> AsyncIterator[Any]

The app's default cache on the test CACHE_URL, emptied first.

Yields the CacheProvider the app itself resolves (core.cache(), idempotency, rate limits, token revocation). The Redis database is FLUSHDB'd before the test so keys never leak between tests — so CACHE_URL must name a disposable test database. Also needed where the app was imported before .env.test was loaded (its settings snapshot names another cache), as in the framework's own suite.

sandloom_core¶

@pytest.fixture
def sandloom_core(sandloom_app: Any) -> Any

The application's CoreFacade.

sandloom_test_env¶

@pytest.fixture
def sandloom_test_env(request: pytest.FixtureRequest) -> Path

Load the nearest .env.test (rootdir first, then the test paths).

scratch_database¶

@pytest.fixture
def scratch_database(sandloom_test_env: Path) -> Iterator[str]

A throw-away database on the test server (sync URL), dropped after.

For a module's migration tests (sandloom.core.testing.migrations): walk its chain up and down without touching the shared test database.

test_db¶

@pytest.fixture
def test_db(db: Any) -> Any

(no docstring)

test_engine¶

@pytest.fixture
def test_engine(db_engine: Any) -> Any

(no docstring)

user_headers¶

@pytest.fixture
def user_headers(
    make_headers: Callable[..., tuple[Any, dict[str, str]]],
) -> dict[str, str]

Headers of a default user (random user id and tenant).

Examples and related guides¶

• Testing examples and fixtures
• Python API overview