Практика: Production-ready проект — Pytest: Профессиональные инструменты

Вы изучили xdist, src layout, advanced fixtures, plugins, coverage thresholds. Пора применить всё вместе на реальном проекте!

Цель: Создать production-ready проект со всеми изученными инструментами.

Что будем строить

Task Management API с:

src layout структура
pytest-xdist для параллельного запуска
Advanced fixtures (session/module scope)
Кастомный плагин для команды
Coverage thresholds (80%)
CI/CD ready

Структура проекта

task-api/
├── src/
│   └── taskapi/
│       ├── __init__.py
│       ├── models/
│       │   ├── __init__.py
│       │   └── task.py
│       ├── services/
│       │   ├── __init__.py
│       │   └── task_service.py
│       └── api/
│           ├── __init__.py
│           └── routes.py
├── tests/
│   ├── conftest.py
│   ├── unit/
│   │   ├── test_task_model.py
│   │   └── test_task_service.py
│   └── integration/
│       └── test_api_routes.py
├── pytest_team_plugin/
│   ├── __init__.py
│   └── plugin.py
├── pyproject.toml
├── pytest.ini
├── .coveragerc
└── README.md

Шаг 1: Создание структуры

pyproject.toml

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
 
[project]
name = "taskapi"
version = "1.0.0"
description = "Task Management API"
requires-python = ">=3.9"
dependencies = [
    "fastapi>=0.104.0",
    "sqlalchemy>=2.0.0",
    "pydantic>=2.0.0",
]
 
[project.optional-dependencies]
dev = [
    "pytest>=7.4.0",
    "pytest-cov>=4.1.0",
    "pytest-xdist>=3.5.0",
    "httpx>=0.25.0",  # Для тестирования FastAPI
]
 
[tool.setuptools.packages.find]
where = ["src"]
 
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = "test_*.py"
python_classes = "Test*"
python_functions = "test_*"

pytest.ini

[pytest]
# Директории с тестами
testpaths = tests
 
# Опции по умолчанию
addopts =
    -v
    --tb=short
    --strict-markers
    --cov=src/taskapi
    --cov-report=term-missing
    --cov-report=html
    --cov-fail-under=80
    -n auto  # Параллельный запуск
    --dist loadscope  # Оптимизация для фикстур
 
# Маркеры
markers =
    slow: Медленные тесты (> 1 секунды)
    integration: Integration тесты
    unit: Unit тесты
    api: API endpoint тесты
    smoke: Критичные smoke тесты

.coveragerc

[run]
source = src/taskapi
omit =
    */tests/*
    */__pycache__/*
 
[report]
precision = 2
show_missing = True
skip_covered = False
 
[html]
directory = htmlcov

Шаг 2: Код приложения

Models

# src/taskapi/models/task.py
 
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
 
@dataclass
class Task:
    id: int
    title: str
    description: str
    completed: bool = False
    created_at: Optional[datetime] = None
 
    def complete(self):
        """Отметить как выполненное"""
        self.completed = True
 
    def update(self, title=None, description=None):
        """Обновить задачу"""
        if title:
            self.title = title
        if description:
            self.description = description

Services

# src/taskapi/services/task_service.py
 
from typing import List, Optional
from datetime import datetime
from ..models.task import Task
 
class TaskService:
    def __init__(self):
        self._tasks = {}
        self._next_id = 1
 
    def create(self, title: str, description: str) -> Task:
        """Создать задачу"""
 
        task = Task(
            id=self._next_id,
            title=title,
            description=description,
            created_at=datetime.now()
        )
        self._tasks[task.id] = task
        self._next_id += 1
        return task
 
    def get(self, task_id: int) -> Optional[Task]:
        """Получить задачу"""
 
        return self._tasks.get(task_id)
 
    def list(self) -> List[Task]:
        """Список всех задач"""
 
        return list(self._tasks.values())
 
    def complete(self, task_id: int) -> bool:
        """Отметить задачу как выполненную"""
 
        task = self._tasks.get(task_id)
        if task:
            task.complete()
            return True
        return False

Шаг 3: Advanced fixtures

tests/conftest.py

import pytest
from src.taskapi.services.task_service import TaskService
from src.taskapi.models.task import Task
 
# ===== Session-scoped fixtures =====
 
@pytest.fixture(scope="session")
def session_config():
    """Конфигурация для всей сессии"""
 
    return {
        "env": "test",
        "debug": True,
    }
 
# ===== Module-scoped fixtures =====
 
@pytest.fixture(scope="module")
def task_service():
    """TaskService для модуля (переиспользуется)"""
 
    return TaskService()
 
# ===== Function-scoped fixtures =====
 
@pytest.fixture
def clean_task_service():
    """Чистый TaskService для каждого теста"""
 
    return TaskService()
 
@pytest.fixture
def sample_task(clean_task_service):
    """Пример задачи"""
 
    return clean_task_service.create(
        title="Sample Task",
        description="This is a sample task"
    )
 
# ===== Factories =====
 
@pytest.fixture
def make_task(clean_task_service):
    """Factory для создания задач"""
 
    tasks = []
 
    def _make(title="Task", description="Description"):
        task = clean_task_service.create(title, description)
        tasks.append(task)
        return task
 
    yield _make
 
    # Cleanup автоматический
    for task in tasks:
        clean_task_service._tasks.pop(task.id, None)
 
# ===== Autouse fixtures =====
 
@pytest.fixture(autouse=True)
def reset_state():
    """Сбрасываем глобальное состояние перед каждым тестом"""
 
    # Setup
    yield
    # Teardown
    pass  # Cleanup если нужен

Шаг 4: Тесты

Unit тесты

# tests/unit/test_task_model.py
 
import pytest
from src.taskapi.models.task import Task
 
@pytest.mark.unit
def test_task_creation():
    """Тест создания задачи"""
 
    task = Task(id=1, title="Test", description="Test task")
 
    assert task.id == 1
    assert task.title == "Test"
    assert task.completed == False
 
@pytest.mark.unit
def test_task_complete():
    """Тест отметки как выполненной"""
 
    task = Task(id=1, title="Test", description="Test")
 
    task.complete()
 
    assert task.completed == True
 
@pytest.mark.unit
def test_task_update():
    """Тест обновления задачи"""
 
    task = Task(id=1, title="Old", description="Old desc")
 
    task.update(title="New", description="New desc")
 
    assert task.title == "New"
    assert task.description == "New desc"

# tests/unit/test_task_service.py
 
import pytest
from src.taskapi.services.task_service import TaskService
 
@pytest.mark.unit
def test_create_task(clean_task_service):
    """Тест создания задачи через сервис"""
 
    task = clean_task_service.create("Task 1", "Description 1")
 
    assert task.id == 1
    assert task.title == "Task 1"
 
@pytest.mark.unit
def test_get_task(sample_task, clean_task_service):
    """Тест получения задачи"""
 
    task = clean_task_service.get(sample_task.id)
 
    assert task is not None
    assert task.title == "Sample Task"
 
@pytest.mark.unit
def test_list_tasks(make_task):
    """Тест списка задач"""
 
    task1 = make_task("Task 1")
    task2 = make_task("Task 2")
    task3 = make_task("Task 3")
 
    service = TaskService()
    tasks = service.list()
 
    # Фабрика создала 3 задачи, но они в другом сервисе!
    # Это демонстрация изоляции

Integration тесты

# tests/integration/test_task_workflow.py
 
import pytest
 
@pytest.mark.integration
def test_complete_task_workflow(clean_task_service):
    """Полный цикл работы с задачей"""
 
    # Create
    task = clean_task_service.create("Buy milk", "From the store")
    assert task.id == 1
 
    # Get
    retrieved = clean_task_service.get(1)
    assert retrieved.title == "Buy milk"
 
    # Complete
    success = clean_task_service.complete(1)
    assert success == True
    assert retrieved.completed == True
 
    # List
    tasks = clean_task_service.list()
    assert len(tasks) == 1
    assert tasks[0].completed == True
 
@pytest.mark.integration
@pytest.mark.slow
def test_multiple_tasks_workflow(make_task):
    """Работа с множеством задач"""
 
    # Создаём 100 задач
    tasks = [make_task(f"Task {i}") for i in range(100)]
 
    assert len(tasks) == 100
    assert all(not task.completed for task in tasks)

Шаг 5: Кастомный плагин

pytest_team_plugin/plugin.py

import pytest
import time
 
def pytest_configure(config):
    """Регистрируем маркеры"""
    config.addinivalue_line(
        "markers",
        "team: Tests for team features"
    )
 
def pytest_collection_modifyitems(items):
    """Сортируем smoke тесты первыми"""
 
    smoke_tests = []
    other_tests = []
 
    for item in items:
        if "smoke" in item.keywords:
            smoke_tests.append(item)
        else:
            other_tests.append(item)
 
    items[:] = smoke_tests + other_tests
 
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item):
    """Логируем медленные тесты"""
 
    start = time.time()
    yield
    duration = time.time() - start
 
    if duration > 1.0 and "slow" not in item.keywords:
        print(f"\n⚠️ Unmarked slow test: {item.name} ({duration:.2f}s)")

Активация в conftest.py:

# tests/conftest.py
 
pytest_plugins = ["pytest_team_plugin.plugin"]

Шаг 6: Запуск

Установка

# Установка пакета в editable режиме
pip install -e ".[dev]"

Запуск тестов

# Все тесты (параллельно с coverage)
pytest
 
# Только unit тесты (быстро)
pytest -m unit
 
# Только integration (медленно)
pytest -m integration
 
# Smoke тесты
pytest -m smoke
 
# Без параллельности
pytest -n 0
 
# С детальным coverage
pytest --cov-report=html
open htmlcov/index.html

Поздравляю! 🎉

Вы завершили курс pytest-professional-tools!

Что вы создали

Production-ready структура — src layout
8x ускорение — pytest-xdist
Оптимизированные фикстуры — session/module scope
Кастомный плагин — автоматизация для команды
Quality gates — coverage thresholds
CI-ready — готов к GitLab/GitHub

Что дальше?

Вы прошли Level 3 (Professional). Следующие шаги:

Level 4: DB Testing — Alembic, factory_boy, Testcontainers
Level 5: Async Race Conditions — ТОЛЬКО для asyncio
Level 6: TDD/CI/CD — Test-Driven Development

Применяйте изученное:

Мигрируйте свои проекты на src layout
Включите xdist для ускорения
Настройте coverage thresholds в CI
Создайте плагин для команды

Устранение неисправностей

Используйте pytest-cov (оно агрегирует репорты).

pip install pytest-cov
pytest -n auto --cov=src

Включите распределение по scope.

# pytest.ini
[pytest]
addopts = -n auto --dist loadscope

Активируйте его в conftest.py.

pytest_plugins = ["pytest_team_plugin.plugin"]

pytest-cov сам объединяет отчёты воркеров, просто используйте его.

🎉 Congratulations on completing the course!