Testy bazy danych w pytest: izolacja testów i rollback w SQLAlchemy

Kod z bazą działa albo wygląda, że działa, dopóki go nie przetestujesz. W tym artykule piszesz testy dla warstwy danych z poprzednich części: modeli i sesji w SQLAlchemy 2 oraz migracji w Alembicie. Ten temat omawiamy na kursie Python online, w module o bazach danych i sesjach. Wszystkie testy uruchomiłem lokalnie (Python 3.13, pytest 9.1, pytest-asyncio 1.4, SQLAlchemy 2.1, FastAPI 0.159).

Czy bazę w testach trzeba mockować?

Jeśli testujesz zapytania, relacje i ograniczenia, mock Ci tego nie sprawdzi. Atrapa zwróci to, co jej każesz, nawet jeśli prawdziwe zapytanie jest błędne. Atrapy świetnie sprawdzają się przy zewnętrznych usługach, o czym piszę w artykule o testowaniu kodu z API LLM. Bazę SQLite możesz uruchomić w milisekundach, więc nie ma powodu jej udawać.

Jak skonfigurować pytest dla testów async?

pip install pytest pytest-asyncio httpx

Domyślnie pytest-asyncio działa w trybie strict i test async def bez znacznika kończy się błędem async def functions are not natively supported. Najprościej włączyć tryb automatyczny w pliku pytest.ini:

[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function

Po tej zmianie każdy async def test_... działa bez dodatkowych dekoratorów.

Jak napisać fixture z bazą w pamięci?

W pliku conftest.py utwórz silnik dla bazy w pamięci, włącz klucze obce, załóż tabele i oddaj sesję testowi. Po teście wszystko znika razem z silnikiem.

import pytest
from sqlalchemy import event
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from db import Base
import models  # noqa: F401


@pytest.fixture
async def engine():
    engine = create_async_engine("sqlite+aiosqlite://")

    @event.listens_for(engine.sync_engine, "connect")
    def enable_foreign_keys(dbapi_connection, connection_record):
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


@pytest.fixture
async def session(engine):
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
    async with SessionLocal() as session:
        yield session

Adres sqlite+aiosqlite:// (bez nazwy pliku) oznacza bazę w pamięci. SQLAlchemy sam dobiera wtedy StaticPool, czyli jedno połączenie współdzielone przez silnik, więc tabele założone w jednym kroku są widoczne w następnym. Sprawdziłem to: ręczne ustawianie puli nie jest potrzebne.

Jakie testy napisać dla warstwy danych?

import pytest
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from models import Conversation, Message, User


async def test_zapis_i_odczyt_rozmowy(session):
    user = User(email="anna@example.com")
    session.add(Conversation(title="Pierwsza", user=user))
    await session.commit()

    conv = await session.scalar(select(Conversation))
    assert conv.title == "Pierwsza"
    assert conv.user_id == user.id


async def test_baza_jest_pusta_na_starcie(session):
    assert await session.scalar(select(User)) is None


async def test_klucz_obcy_jest_pilnowany(session):
    session.add(Conversation(user_id=999, title="sierota"))
    with pytest.raises(IntegrityError):
        await session.commit()


async def test_usuniecie_rozmowy_usuwa_wiadomosci(session):
    conv = Conversation(title="x", user=User(email="a@b.pl"))
    conv.messages.append(Message(role="user", content="hej"))
    session.add(conv)
    await session.commit()

    await session.delete(conv)
    await session.commit()
    assert await session.scalar(select(Message)) is None

Każdy test sprawdza jedno ograniczenie: zapis, pustą bazę, klucz obcy i kaskadę usuwania. W moim projekcie wszystkie przeszły, a cały zestaw z testami endpointów (6 testów) zajął poniżej 0,1 sekundy.

Dlaczego testy muszą być od siebie odizolowane?

Test, który zostawia dane w bazie, psuje następne. Pokazuję to na przykładzie, którego nie powinieneś naśladować: dwa testy używają tego samego pliku bazy i nikt go nie czyści.

async def test_a_dodaje():
    async with (await _maker())() as s:
        s.add(User(email="a@b.pl"))
        await s.commit()


async def test_b_oczekuje_pustej_bazy():
    async with (await _maker())() as s:
        assert await s.scalar(select(func.count()).select_from(User)) == 0

Wynik: drugi test pada z komunikatem assert 1 == 0, bo widzi użytkownika z pierwszego. Przy następnym uruchomieniu pada już pierwszy test: UNIQUE constraint failed: users.email, bo plik z poprzedniego przebiegu nadal istnieje. Najgorsze w takich błędach jest to, że zależą od kolejności testów, więc raz wychodzą, a raz nie.

Dwie zasady, które to rozwiązują:

• każdy test zaczyna od znanego stanu bazy (pusta albo zasilona danymi, które sam ustawia),
• nic nie przecieka między testami: ani plik, ani sesja, ani niezatwierdzona transakcja.

Baza w pamięci tworzona w fixture od nowa dla każdego testu spełnia obie zasady bez dodatkowej pracy.

Co się stanie, jeśli silnik testowy nie ma włączonych kluczy obcych?

Test przejdzie z błędnego powodu. Odtworzyłem to: silnik bez PRAGMA foreign_keys=ON i test, który oczekuje IntegrityError. Wynik: Failed: DID NOT RAISE IntegrityError. SQLite przyjął wiersz z nieistniejącym user_id, więc test, który miał chronić integralność danych, wykrył brak ochrony w samym teście.

To powód, dla którego włączenie kluczy obcych jest w fixture engine, a nie w osobnym miejscu. Jeśli produkcyjny silnik ma je włączone (jak w artykule o sesjach w SQLAlchemy), silnik testowy musi mieć dokładnie to samo.

Jak przetestować endpointy FastAPI z prawdziwą bazą?

FastAPI pozwala podmienić zależność na czas testu przez app.dependency_overrides. Zamiast produkcyjnej get_session endpoint dostaje sesję z bazy testowej. Klient httpx wywołuje aplikację bezpośrednio, bez uruchamiania serwera.

from httpx import ASGITransport, AsyncClient

from db import get_session
from main import app


@pytest.fixture
async def client(engine):
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

    async def override_get_session():
        async with SessionLocal() as session:
            yield session

    app.dependency_overrides[get_session] = override_get_session
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()


async def test_endpoint_dodaje_wiadomosc(client):
    r = await client.post("/conversations", params={"title": "T"})
    assert r.status_code == 201
    r = await client.post(
        "/conversations/1/messages", json={"role": "user", "content": "Cześć"}
    )
    assert r.status_code == 201
    r = await client.get("/conversations/1")
    assert r.json()["messages"][0]["content"] == "Cześć"


async def test_brak_rozmowy_daje_404(client):
    r = await client.get("/conversations/99")
    assert r.status_code == 404

Linia app.dependency_overrides.clear() po teście jest ważna. Bez niej podmiana zostaje w aplikacji także dla kolejnych testów, a to znowu prowadzi do wzajemnego zależenia testów od siebie.

Jak wycofywać zmiany po każdym teście zamiast tworzyć bazę od nowa?

Gdy schemat jest duży albo zasilasz bazę przed testami, tworzenie jej na każdy test bywa drogie. Wtedy zakładasz bazę raz na całą sesję testową, a każdy test działa w transakcji, którą po teście wycofujesz. Wywołania commit() w testowanym kodzie zatwierdzają wtedy tylko zagnieżdżony savepoint, więc zewnętrzny rollback i tak wszystko cofa.

W SQLAlchemy służy do tego join_transaction_mode="create_savepoint":

from sqlalchemy.ext.asyncio import AsyncSession


@pytest.fixture
async def session(engine):
    async with engine.connect() as conn:
        outer = await conn.begin()
        async with AsyncSession(
            conn, join_transaction_mode="create_savepoint", expire_on_commit=False
        ) as session:
            yield session
        await outer.rollback()

Na pierwszy rzut oka działa. Ja jednak uruchomiłem to na jednej bazie w pliku i dostałem dane przeciekające między testami: assert 200 == 404 (rozmowa z poprzedniego testu nadal istniała) oraz assert 2 == 1 (kolejna rozmowa dostała id 2). Przyczyna leży w sterowniku sqlite3: domyślnie nie otwiera transakcji tak, jak oczekuje SQLAlchemy, więc zewnętrzny rollback nic nie cofa.

Naprawa to znane obejście, które dodajesz do silnika:

@event.listens_for(engine.sync_engine, "connect")
def do_connect(dbapi_connection, connection_record):
    dbapi_connection.isolation_level = None
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()


@event.listens_for(engine.sync_engine, "begin")
def do_begin(conn):
    conn.exec_driver_sql("BEGIN")

Pierwsza funkcja wyłącza automatyczne zarządzanie transakcją po stronie sterownika, a druga sama wysyła BEGIN na początku każdej transakcji SQLAlchemy. Po tej zmianie ten sam zestaw dał 3 zaliczone testy endpointów (id == 1 w każdym) i 6 razem z testami kluczy obcych, kaskady i pustej bazy.

Dla fixture engine na poziomie całej sesji testowej pamiętaj o zasięgu pętli zdarzeń. Fixture z scope="session" bez ustawienia pętli kończy się błędem ScopeMismatch: You tried to access the function scoped fixture _function_scoped_runner. Dodaj do pytest.ini:

[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = session
asyncio_default_test_loop_scope = session

Sprawdziłem to tylko na SQLite. Obejście dotyczy sterownika sqlite3, więc przy innej bazie najpierw sprawdź, czy w ogóle jest potrzebne. Ustawienia pętli zdarzeń są niezależne od bazy.

Którego sposobu izolacji użyć?

Zasada praktyczna: dopóki testy trwają ułamek sekundy, wybierz prostszy sposób. Skomplikowana izolacja, która po cichu przecieka, jest gorsza niż brak izolacji, bo daje fałszywe poczucie bezpieczeństwa.

Jakie błędy mentor poprawia najczęściej?

Najgroźniejszy test to taki, który przechodzi z niewłaściwego powodu. Silnik testowy bez kluczy obcych potwierdzi, że „wszystko działa”, a produkcja odrzuci ten sam zapis. Zanim zaufasz zielonym testom, upewnij się, że baza testowa ma te same ustawienia co produkcyjna, i raz sprawdź, że test potrafi się wywalić.

Co dalej?

Warstwa danych ma modele, migracje i testy. W ostatnim artykule cyklu dodamy to, co zmienia ją w prawdziwą aplikację: izolację użytkowników. Każdy zobaczy tylko własne rozmowy, a Ty napiszesz test, który to udowodni: użytkownik B próbuje odczytać rozmowę użytkownika A i dostaje odmowę.

Wszystkie te zagadnienia omówimy na przyszłym kursie AI developer, który niedługo ogłosimy. Śledź listę naszych ścieżek, żeby nie przegapić startu.