Testing
grelmicro gives you four tools for tests: run the app on in-process stores, swap one backend for a test, drive time by hand, and record the calls a pattern makes.
Declare the test environment
Tests wire memory backends, and a memory backend behind a Lock is what the
backend check reports. Declare the
environment once in conftest.py and the report stops, in the suite where it
has nothing to say. pytest reads conftest.py before your tests:
import os
os.environ.setdefault("GREL_ENVIRONMENT", "test")
Grelmicro(environment="test") does the same for one app. Keep calling
micro.check_backends() for the app you deploy: it answers for production
whatever this process declares.
Test the app you ship
Enter micro.fake() before the test client starts. The client runs your
app's lifespan, the lifespan opens micro, and micro opens on in-process
stores:
from collections.abc import Iterator
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from grelmicro import Grelmicro
from grelmicro.coordination import Lock
from grelmicro.providers.postgres import PostgresProvider
# app.py
micro = Grelmicro(uses=[PostgresProvider("postgresql://app:secret@db/app")])
app = FastAPI()
micro.install(app)
@app.post("/checkout")
async def checkout() -> dict[str, str]:
async with Lock("cart"):
return {"status": "done"}
# conftest.py
@pytest.fixture
def client() -> Iterator[TestClient]:
with micro.fake(), TestClient(app) as client:
yield client
# test_checkout.py
def test_checkout(client: TestClient) -> None:
assert client.post("/checkout").json() == {"status": "done"}
The test runs the app you deploy, with every component and name it really
registers. Coordination, Cache, RateLimiterComponent and
CircuitBreakerComponent open on memory, so the Lock above is a real lock
that needs no database. The Postgres they would have used is never opened, so
the suite connects to nothing and /readyz does not probe it.
A fixture is a pytest function that prepares a test. Do not open micro in the
fixture as well. install(app) already opens it
when the client starts, and a second open raises.
A Provider stays open when a component that is not faked still uses it, such
as the Postgres an Outbox stores in. A test that reaches a closed Provider
directly, postgres.client for example, is told so. Pass
micro.fake(keep=[postgres]) to open the real one for that test.
An app with no HTTP
With no framework to open it, the fixture opens micro itself, after the fake:
from collections.abc import AsyncIterator
from unittest.mock import AsyncMock
import pytest
from grelmicro import Grelmicro
from grelmicro.coordination import Coordination, Lock, LockBackend
from grelmicro.providers.postgres import PostgresProvider
# app.py
micro = Grelmicro(uses=[PostgresProvider("postgresql://app:secret@db/app")])
async def reserve(sku: str) -> None:
async with Lock(f"stock:{sku}"):
...
# conftest.py
@pytest.fixture
async def running() -> AsyncIterator[Grelmicro]:
async with micro.fake(), micro:
yield micro
# test_stock.py
async def test_reserve(running: Grelmicro) -> None:
await reserve("sku-1")
async def test_reserve_takes_the_lock(running: Grelmicro) -> None:
backend = AsyncMock(spec=LockBackend)
async with running.override(Coordination(lock=backend)):
await reserve("sku-1")
backend.acquire.assert_awaited()
Swap one backend
micro.override(...) replaces a component on the open app for a block and
puts the original back on exit. The last test above uses it to check that
reserve takes the lock. Reach for fake() when the test is about your code,
and for override() when it is about the calls to a backend.
Drive time with VirtualClock
Time-dependent patterns (retry backoff, circuit breaker cooldown, rate limiter
refill) read time through grelmicro's clock seam. Install a VirtualClock and
advance it by hand so tests never wait real seconds:
from grelmicro.clock import VirtualClock
async def test_cooldown() -> None:
async with VirtualClock() as clock:
...
await clock.advance(30) # cooldown elapses, no real wait
With no clock installed, the seam forwards to real time, so production pays nothing.
Record calls
record(backend) instruments a backend in place and returns a CallLog. The
backend keeps its real behavior while the log captures every call for assertions:
from grelmicro import Grelmicro
from grelmicro.coordination import Coordination
from grelmicro.providers.memory import MemoryProvider
from grelmicro.testing import record
async def test_login_takes_the_lock() -> None:
backend = MemoryProvider().lock_backend()
log = record(backend)
micro = Grelmicro(uses=[Coordination(lock=backend)])
async with micro:
await login("u1")
assert log.count("acquire", name="user:u1") == 1
Test an authenticated app
AuthenticatedRequests takes any verifier that answers verify(token). In a
test, hand it a FakeVerifier: each token is a name for the claims it carries,
built with fake_claims(subject, *scopes), so no key is generated and no token
is signed:
from http import HTTPStatus
from fastapi import FastAPI
from fastapi.testclient import TestClient
from grelmicro import Grelmicro
from grelmicro.http import AuthenticatedRequests, ErrorResponses
from grelmicro.integrations.fastapi import CurrentPrincipal
from grelmicro.testing import FakeVerifier, fake_claims
app = FastAPI()
@app.get("/orders")
async def orders(principal: CurrentPrincipal) -> dict[str, str | None]:
return {"subject": principal.subject}
verifier = FakeVerifier(alice=fake_claims("alice", "orders:read"))
micro = Grelmicro(uses=[ErrorResponses(), AuthenticatedRequests(verifier)])
micro.install(app)
def test_orders_needs_a_token() -> None:
with TestClient(app) as client:
assert client.get("/orders").status_code == HTTPStatus.UNAUTHORIZED
assert client.get(
"/orders", headers={"Authorization": "Bearer alice"}
).json() == {"subject": "alice"}
A token the verifier does not know is refused the way a bad one is in
production, so the 401 and its challenge are tested too. To test the verifier
itself, sign a real token as
Test with a real token shows.
app.dependency_overrides changes what a dependency returns, never what a
route declares, so the response cache still answers a route declaring
CachedResponse(). To reach the handler on every request, leave
CachedResponses() out of the test's uses, or name the path in its
exclude=.
Going deeper
The Testing architecture page covers override
restrictions, how VirtualClock interacts with each backend, and the full
CallLog API.