Контракт репозитория в pytest: один набор тестов для памяти и SQLite

Обработчик обращений проверяют с репозиторием на словаре. В CI все сценарии зелёные. На рабочем сервисе приходит повтор заявки, SQLite отклоняет вставку по первичному ключу, и клиент видит ошибку. Тестовый репозиторий при повторе просто заменял значение в словаре. Он принимал те же аргументы, но выполнял другую операцию.

Обе реализации можно описать одним Protocol. Проверка типов покажет, что у них есть методы add и get с ожидаемыми сигнатурами. Решение о поведении при повторе нужно выразить отдельными тестами. Иначе код, который ловит конфликт или показывает причину отказа, проверяется на хранилище с другими правилами.

Сформулируйте контракт через видимое состояние

Возьмём три требования. Новую заявку можно прочитать по ключу. Неизвестный ключ возвращает None. Повторная вставка возбуждает DuplicateTicket и сохраняет исходный текст. Последняя проверка особенно важна: реализация может сначала перезаписать заявку, а потом сообщить о конфликте.

Параметризованная фикстура pytest по очереди создаёт оба хранилища. Каждый тест получает свежий экземпляр. Для SQLite используем файл в tmp_path, а соединение закрываем при завершении теста. Сохраните пример в test_ticket_repository.py, установите pytest командой python -m pip install pytest и запустите python -m pytest -q.

import sqlite3
from dataclasses import dataclass
from typing import Protocol

import pytest


@dataclass(frozen=True)
class Ticket:
    key: str
    text: str


class DuplicateTicket(Exception):
    pass


class Tickets(Protocol):
    def add(self, ticket: Ticket) -> None: ...
    def get(self, key: str) -> Ticket | None: ...


class MemoryTickets:
    def __init__(self):
        self.items: dict[str, Ticket] = {}

    def add(self, ticket: Ticket) -> None:
        if ticket.key in self.items:
            raise DuplicateTicket(ticket.key)
        self.items[ticket.key] = ticket

    def get(self, key: str) -> Ticket | None:
        return self.items.get(key)


class SqliteTickets:
    def __init__(self, db: sqlite3.Connection):
        self.db = db
        self.db.execute("CREATE TABLE tickets (key TEXT PRIMARY KEY, text TEXT NOT NULL)")

    def add(self, ticket: Ticket) -> None:
        with self.db:
            inserted = self.db.execute(
                "INSERT INTO tickets (key, text) VALUES (?, ?) ON CONFLICT(key) DO NOTHING",
                (ticket.key, ticket.text),
            )
            if inserted.rowcount == 0:
                raise DuplicateTicket(ticket.key)

    def get(self, key: str) -> Ticket | None:
        row = self.db.execute("SELECT key, text FROM tickets WHERE key = ?", (key,)).fetchone()
        return None if row is None else Ticket(*row)


@pytest.fixture(params=["memory", "sqlite"])
def repository(request, tmp_path):
    if request.param == "memory":
        yield MemoryTickets()
    else:
        db = sqlite3.connect(tmp_path / "tickets.db")
        try:
            yield SqliteTickets(db)
        finally:
            db.close()


def test_reads_saved_ticket(repository: Tickets):
    ticket = Ticket("A-19", "Cannot open the report")
    repository.add(ticket)
    assert repository.get(ticket.key) == ticket


def test_absent_key(repository: Tickets):
    assert repository.get("unknown") is None


def test_conflict_keeps_original(repository: Tickets):
    original = Ticket("A-19", "Cannot open the report")
    repository.add(original)
    with pytest.raises(DuplicateTicket):
        repository.add(Ticket(original.key, "Different text"))
    assert repository.get(original.key) == original

Pytest выполняет шесть проверок. При сбое имя теста включает параметр, например test_conflict_keeps_original[memory]. Чтобы увидеть пользу, уберите проверку ключа перед присваиванием в MemoryTickets.add(). Сценарий конфликта для памяти упадёт, а сценарий для SQLite продолжит проходить. Так можно обнаружить, что тестовая подмена обещает вызывающему коду другое поведение.

Проверяйте особенности конкретной реализации отдельно

Для контракта подходят правила, которые нужны клиенту репозитория независимо от способа хранения. Для SQLite полезен отдельный тест, который закрывает соединение и читает запись через новое. Этим вы проверите фиксацию транзакции, а не видимость записи в рамках того же соединения. Для памяти такой сценарий лишён смысла. При добавлении PostgreSQL включите адаптер в общий набор и добавьте проверки, связанные с его изоляцией и конкурентной записью.

Тесты бизнес-логики с быстрой подменой можно оставить. Общий набор стоит прогонять при изменении любого адаптера. Начните с операций, на результат которых завязаны ответы приложения. Если обработчик различает «нет записи» и «запись уже существует», закрепите эти различия на границе хранилища до того, как появится третий способ записи.

На бесплатном курсе Latorn «Python для продвинутых» вы поработаете с Protocol, фикстурами и подменами в pytest, а в итоговом трекере задач отделите файловое хранение от логики через репозиторий. Открыть программу курса по продвинутому Python.