Tests

Writing tests (tests/unit/, tests/integration/, tests/e2e/)

Tests without rules drift into ceremony — "test exists, test passed, shipping". Real bugs this repo shipped because nobody had a test that exercised the exact scenario:

1. UserResult.model_validate(orm_user) in update.py / create.py / permission.py — crashed with MissingGreenlet on PATCH/POST. Not caught because the e2e tests only hit GET.
2. UpdateUserRequest had only password field while AdminUpdateUser contract had password + active + role_uuid — PATCH silently dropped active and role_uuid (Pydantic extra="ignore"), returned 200, and did nothing. Not caught because no test verified that the mutation persisted via a follow-up GET.
3. select_many without loads=role triggered MissingGreenlet. Caught only because an explicit test_select_users_omits_role_by_default test existed. Other endpoints had no such test.

Every "why didn't the tests catch this" answer has the same root: the scenario was never tested. This skill lists the scenarios that must exist.

Pick the right layer — don't cross them

| Layer | What it tests | What it uses | Typical runtime | |---|---|---|---| | tests/unit/ | Pure logic: use cases, validators, helpers, contracts | MagicMock/AsyncMock DBGateway, FakeHasher, FakeJWT, FakeStrCache from tests/fakes.py | /.py there is a tests/unit/test___usecase.py` file. No exceptions. If a use case has no unit test, it is not considered finished.

Fixtures

Shared fakes live in tests/fakes.py (FakeHasher, FakeJWT, FakeStrCache). Shared fixtures live in tests/unit/conftest.py (fake_cache, fake_hasher, fake_jwt, auth_service, services, fake_database, stub_user). Reuse them — do not redefine.

Patterns

1. Mock DBGateway.. to return the expected ORM stub, capture call kwargs:

@pytest.fixture
def update_repo_call(fake_database: MagicMock) -> Any:
    captured: dict[str, Any] = {}

    def _stub(user_stub: MagicMock) -> None:
        async def _update(uuid_arg: Any, /, **kwargs: Any) -> MagicMock:
            captured["uuid"] = uuid_arg
            captured["kwargs"] = kwargs
            return MagicMock(result=lambda: user_stub)

        fake_database.user.update = AsyncMock(side_effect=_update)

    _stub.captured = captured  # type: ignore[attr-defined]
    return _stub

2. ORM stub must implement .as_dict() to mimic Base.as_dict():

def _make_orm_user(login: str = "alice", active: bool = True) -> MagicMock:
    user = MagicMock()
    user.uuid = uuid.uuid4()
    user.login = login
    user.active = active
    user.as_dict.return_value = {
        "uuid": user.uuid,
        "login": login,
        "active": active,
    }
    return user

Do not add role to as_dict.return_value unless you are specifically testing the eager-loaded case — mimicking the production behavior where unloaded relationships simply aren't in __dict__.

3. Assert what the use case forwards to the repository, not just the return value:

async def test_update_user_passes_password_hashed(
    use_case: UpdateUserUseCase,
    update_repo_call: Any,
    fake_hasher: FakeHasher,
) -> None:
    update_repo_call(_make_orm_user())

    await use_case(UpdateUserRequest(user_uuid=uuid.uuid4(), password="plain"))

    assert update_repo_call.captured["kwargs"]["password"] == fake_hasher.hash_password("plain")

4. exclude_unset=True coverage — when client omits a field, verify it is not forwarded:

async def test_update_user_omits_unset_fields(
    use_case: UpdateUserUseCase, update_repo_call: Any
) -> None:
    update_repo_call(_make_orm_user())

    await use_case(UpdateUserRequest(user_uuid=uuid.uuid4(), active=False))

    kwargs = update_repo_call.captured["kwargs"]
    assert kwargs == {"active": False}
    assert "password" not in kwargs
    assert "role_uuid" not in kwargs

This is the test that would have caught the AdminUpdateUser → UpdateUserRequest field-mismatch bug.

Mandatory scenarios per use case

For every use case, unit tests must cover:

• Happy path — the primary success case returning the expected Result type.
• Each if / raise branch — if the use case raises NotFoundError / ForbiddenError / ConflictError, there is a test that triggers each raise.
• Kwargs forwarding — assert the exact kwargs passed to each repo method. Prevents silent field drops.
• Return type matches declaration — if the method is typed -> UserResult, assert isinstance(result, UserResult) and at least one field value.
• exclude_unset=True / exclude_none=True semantics — if the use case selectively forwards fields, test both "included" and "omitted".
• Pagination propagation (for SelectMany* use cases) — test limit=None → repo called with limit=None; test default → repo called with default; test strict limits → forwarded as-is.

Anti-patterns

| ❌ | ✅ | |---|---| | Testing use_case(request) and only asserting result.uuid is not None | Assert the captured repo kwargs against an exact expected dict | | Using a real DBGateway and a real session in a unit test | That's integration. Move the test. | | Adding role to the ORM stub's .as_dict() return when testing "no loads" scenarios | Mimic real behavior — unloaded relations are absent from __dict__ | | Hardcoding uuid.uuid4() literals as strings in assertions | Capture the generated UUID from the request, compare by reference | | Testing Pydantic Field(...) validators via the use case | Test contracts/requests directly in test_.py — don't go through the use case |

Integration tests — repositories

For every repository method that has non-trivial logic (select_many with filters, select with multiple where branches, @on_integrity wrapped methods, Query Objects), there is an integration test. Trivial insert / plain SELECT by uuid are covered by the repo's unit-typed _crud helper and don't need per-column tests.

Fixtures

• database: DBGateway (function scope) from root tests/conftest.py — opens a fresh session against the shared testcontainer Postgres, wraps the test in an outer transaction that rolls back at teardown (per-test isolation without explicit cleanup).
• unique_login: Callable[[], str] — factory that returns a unique login per call (avoids unique-constraint clashes across test files running in parallel).
• db_session: AsyncSession — if you need raw session access.

Patterns

1. Seed via async with database: (write transaction), assert via async with database.manager.session: (read-only) — same request scope:

async def test_select_many_filters_by_login_substring(
    database: DBGateway, unique_login: Callable[[], str]
) -> None:
    needle = unique_login()

    async with database:
        role = (await database.role.select(name="User")).result()
        await database.user.create(login=needle, password="hashed", role_uuid=role.uuid)
        await database.user.create(login=unique_login(), password="hashed", role_uuid=role.uuid)

    async with database.manager.session:
        total, users = (await database.user.select_many(login=needle, limit=None)).result()

    assert total == 1
    assert users[0].login == needle

2. Test limit=None explicitly — internal callers rely on it. The contract layer never lets None through HTTP, but use cases/tasks do.

async def test_select_many_unlimited_returns_all_rows(
    database: DBGateway, unique_login: Callable[[], str]
) -> None:
    async with database:
        role = (await database.role.select(name="User")).result()
        for _ in range(25):
            await database.user.create(login=unique_login(), password="hashed", role_uuid=role.uuid)

    async with database.manager.session:
        total, users = (await database.user.select_many(limit=None)).result()

    assert total >= 25
    assert len(users) == total

3. Test @on_integrity raises the right domain exception on constraint violation:

async def test_create_duplicate_login_raises_conflict(
    database: DBGateway, unique_login: Callable[[], str]
) -> None:
    login = unique_login()
    async with database:
        role = (await database.role.select(name="User")).result()
        await database.user.create(login=login, password="hashed", role_uuid=role.uuid)

    with pytest.raises(ConflictError):
        async with database:
            role = (await database.role.select(name="User")).result()
            await database.user.create(login=login, password="hashed", role_uuid=role.uuid)

Mandatory scenarios per repository method

• create — success + @on_integrity raises domain exception on constraint violation.
• select — by-uuid success, by-uuid not found raises NotFoundError, each alternative filter key.
• select_many — default (no filters), each filter key individually, limit=None returns all rows, finite limit caps result, offset paginates correctly, order_by="asc" vs "desc".
• update — each updatable column individually (not "update all at once" — that would mask silent field drops). For @on_integrity methods, include the conflict case.
• delete — success + not found.

E2E tests — endpoints

For every endpoint function in presentation/http/v1/endpoints//.py there is at least one happy-path e2e test. For mutating endpoints (POST/PATCH/PUT/DELETE), there must also be a mutation-persistence test that follows the mutation with a GET and verifies the change landed in the DB.

Infra notes (load-bearing — don't break these)

• httpx.AsyncClient + httpx.ASGITransport, not TestClient. TestClient bridges sync→async through anyio's portal, which spawns its own loop; asyncpg connections made in the test's session-scoped loop crash with "attached to a different loop" inside the portal. ASGITransport runs the app in the same loop as the test.
• Alembic migrations run in a sync session-scoped fixture (_migrate_database), not inside the async db_engine fixture — migrations/env.py uses asyncio.run internally and it explodes inside an already-running event loop.
• pyproject.toml sets asyncio_default_test_loop_scope = "session" to keep the asyncpg connection loop consistent across tests and fixtures.
• cipher_settings fixture generates a valid base64 HS256 key and sets CIPHER_SECRET_KEY == CIPHER_PUBLIC_KEY — JWT.create signs with secret_key, JWT.verify_token decodes with public_key, symmetric HS256 needs them identical.
• The real Hasher (Argon2) is pulled from the e2e container via the hasher fixture — fake hashers would not be recognised by the login use case which uses the real Argon2 verifier.

Auth fixtures

The e2e conftest ships three chained fixtures that you reuse — do not re-implement auth per test:

• hasher: Hasher — resolves the real Argon2 hasher from the e2e container.
• admin_credentials: tuple[str, str] — seeds an Admin-role user directly via DBGateway (bypassing the 2-step register/confirm flow that needs an out-of-band verification code) and returns (login, password).
• admin_token: str — logs in as the seeded admin via POST /v1/auth/login and returns the bearer access token.

For user-role tests (non-admin), seed via database + hasher directly in the test body using the same pattern.

Patterns

1. Happy path + envelope shape:

async def test_select_users_returns_paginated_envelope(
    client: httpx.AsyncClient, admin_token: str
) -> None:
    response = await client.get("/v1/admin/users", headers=_auth(admin_token))

    assert response.status_code == 200
    body = response.json()
    assert set(body.keys()) >= {"data", "offset", "limit", "total"}
    assert body["offset"] == 0
    assert body["limit"] == 10  # contract default

2. Mutation-persistence — the rule that would have caught the UpdateUserRequest field-mismatch bug:

async def test_update_user_active_flag_persists(
    client: httpx.AsyncClient,
    admin_token: str,
    database: DBGateway,
    unique_login: Callable[[], str],
    hasher: Hasher,
) -> None:
    login = unique_login()
    async with database:
        role = (await database.role.select(name="User")).result()
        user = (await database.user.create(
            login=login, password=hasher.hash_password("x"), role_uuid=role.uuid,
        )).result()

    patch = await client.patch(
        f"/v1/admin/users/{user.uuid}",
        headers=_auth(admin_token),
        json={"active": False},
    )
    assert patch.status_code == 200
    assert patch.json()["active"] is False

    get = await client.get(
        "/v1/admin/users",
        headers=_auth(admin_token),
        params={"login": login},
    )
    assert get.status_code == 200
    assert get.json()["data"][0]["active"] is False

Never write a PATCH/POST/DELETE e2e test that only asserts response.status_code == 200. Always follow up with a GET (or the same endpoint) that proves the mutation landed. 200 without persistence is what bit us on AdminUpdateUser.active — the endpoint returned 200 and silently dropped the field.

3. Test loads= opt-in / opt-out both ways:

async def test_select_users_omits_role_by_default(
    client: httpx.AsyncClient, admin_token: str
) -> None:
    response = await client.get("/v1/admin/users", headers=_auth(admin_token))
    assert response.status_code == 200
    users = response.json()["data"]
    assert all("role" not in u for u in users)  # exclude_none drops unloaded

async def test_select_users_loads_role_when_requested(
    client: httpx.AsyncClient, admin_token: str
) -> None:
    response = await client.get(
        "/v1/admin/users", headers=_auth(admin_token), params={"loads": ["role"]},
    )
    assert response.status_code == 200
    users = response.json()["data"]
    assert all("role" in u for u in users)

The "without loads" test is the one that catches model_validate(orm) → MissingGreenlet bugs. Every endpoint that has optional loads must have both sides tested.

4. Contract validation boundaries — for every strict field, test both sides of the constraint:

async def test_select_users_rejects_limit_below_min(
    client: httpx.AsyncClient, admin_token: str
) -> None:
    response = await client.get(
        "/v1/admin/users", headers=_auth(admin_token), params={"limit": 5},
    )
    assert response.status_code == 400  # project converts RequestValidationError to 400
    body = response.json()
    assert body["message"] == "Validation error"
    assert body["detail"][0]["loc"] == ["query", "limit"]
    assert "greater than or equal to 10" in body["detail"][0]["msg"]

Not 422. The project converts RequestValidationError to 400 in presentation/http/common/exception_handlers.py. If you assert 422, the test will be wrong even when the logic is correct.

5. Authorization matrix — for every non-public endpoint:

async def test_select_users_unauthorized_without_token(client: httpx.AsyncClient) -> None:
    response = await client.get("/v1/admin/users")
    assert response.status_code == 401

async def test_admin_endpoint_rejects_user_role_token(
    client: httpx.AsyncClient,
    database: DBGateway,
    hasher: Hasher,
    unique_login: Callable[[], str],
) -> None:
    login = unique_login()
    async with database:
        role = (await database.role.select(name="User")).result()
        await database.user.create(
            login=login, password=hasher.hash_password("pw"), role_uuid=role.uuid,
        )
    login_resp = await client.post(
        "/v1/auth/login",
        json={"login": login, "password": "pw", "fingerprint": "test"},
    )
    user_token = login_resp.json()["token"]

    response = await client.get("/v1/admin/users",

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [ocbunknown](https://github.com/ocbunknown)
- **Source:** [ocbunknown/fastapi-claude-template](https://github.com/ocbunknown/fastapi-claude-template)
- **License:** MIT

Install and usage instructions live in the source repository linked above.