Собираем все знания курса: тестируем полноценное приложение с кешем, удаленным API и обработкой ошибок через фикстуры, моки и интеграционные тесты
⚠️ Урок устарел. Свежая версия в курсе pytest-junior: Мини-проект: Todo с кешем и API.
🎯 За 30 минут вы:
✅ Протестируете полноценное приложение с внешними зависимостями
✅ Увидите, как вместе работают фикстуры, моки и параметризация
✅ Покроете не только happy path, но и ошибки/деградацию
Полный пример кода доступен в репозитории курса: https://github.com/potapov-me/pytest-from-zero-to-confidence — там же лежит исправленная версия на ветке fixed.
todo-app/
src/
app.py # 🎯 Ядро приложения
cache.py # 💾 Кеш на файловой системе
remote.py # 🌐 Клиент для удаленного API
tests/
conftest.py # 🔧 Все фикстуры проекта
unit/
test_app.py
test_cache.py
test_remote.py
integration/
test_todo_flow.py
⚠️ В примере ниже специально оставлены пару ошибок/шероховатостей, чтобы вы получили реальный опыт их поиска и отладки перед запуском тестов.
# 📁 src/app.py
import uuid
from typing import Dict, Any
class TodoApp:
"""Приложение для задач с кешем и удаленной синхронизацией"""
def __init__(self, cache, remote):
self.cache = cache
self.remote = remote
self.tasks: Dict[str, Dict] = {}
def add_task(self, text: str) -> str:
task_id = str(uuid.uuid4())
task = {"id": task_id, "text": text.strip(), "done": False}
if not task["text"]:
raise ValueError("Task text is empty")
self.tasks[task_id] = task
self.cache.set(f"task_{task_id}", task)
self.remote.sync_task(task)
return task_id
def complete_task(self, task_id: str) -> Dict[str, Any]:
# 🛡️ Кеш может быть недоступен
try:
self.cache.invalidate(f"task_{task_id}")
except Exception:
self.remote.log("cache_unavailable", task_id=task_id)
task = self.tasks.get(task_id)
if not task:
raise ValueError(f"Task {task_id} not found")
task["done"] = True
self.remote.sync_task(task)
return task
def list_tasks(self) -> list:
return list(self.tasks.values())# 📁 src/cache.py
import json
class CacheError(Exception):
pass
class FileCache:
"""Простой кеш на файловой системе"""
def __init__(self, storage_path):
self.storage_path = storage_path
def set(self, key, value):
try:
file_path = self.storage_path / f"{key}.json"
file_path.write_text(json.dumps(value))
except Exception as e:
raise CacheError(f"Failed to set cache: {e}")
def invalidate(self, key):
try:
file_path = self.storage_path / f"{key}.json"
if file_path.exists():
file_path.unlink()
except Exception as e:
raise CacheError(f"Failed to invalidate cache: {e}")# 📁 src/remote.py
import requests
class RemoteService:
"""Клиент для удаленного API задач"""
def __init__(self, base_url):
self.base_url = base_url
def sync_task(self, task):
response = requests.post(f"{self.base_url}/tasks/sync", json=task, timeout=10)
response.raise_for_status()
def log(self, event, **data):
requests.post(
f"{self.base_url}/logs", json={"event": event, "data": data}, timeout=5
)RemoteService — внешняя зависимость TodoApp: он принимает задачи/логи и синхронизирует их с удалённым API. В тестах мы подменяем его моками (fake_remote, broken_remote, spy_remote), чтобы не делать реальные HTTP-запросы и проверять логику приложения в изоляции.
# 📁 tests/conftest.py
import pytest
from unittest.mock import Mock
from src.app import TodoApp
from src.cache import FileCache
from src.remote import RemoteService
@pytest.fixture
def temp_storage(tmp_path):
return tmp_path / "cache"
@pytest.fixture
def working_cache(temp_storage):
return FileCache(temp_storage)
@pytest.fixture
def broken_cache():
cache = Mock()
cache.set.side_effect = Exception("Cache storage failed")
cache.invalidate.side_effect = Exception("Cache invalidation failed")
return cache
@pytest.fixture
def fake_remote():
remote = Mock()
remote.sync_task.return_value = None
remote.log.return_value = None
return remote
@pytest.fixture
def spy_remote():
remote = Mock()
remote.calls = []
def record_sync(task):
remote.calls.append(("sync_task", task))
def record_log(event, **data):
remote.calls.append(("log", event, data))
remote.sync_task.side_effect = record_sync
remote.log.side_effect = record_log
remote.assert_sync_called = lambda: remote.sync_task.called
remote.assert_log_contains = lambda text: any(text in str(call) for call in remote.calls)
return remote
@pytest.fixture
def broken_remote():
remote = Mock()
remote.sync_task.side_effect = Exception("Network error")
remote.log.side_effect = Exception("Log service unavailable")
return remote
@pytest.fixture
def app(working_cache, fake_remote):
return TodoApp(working_cache, fake_remote)
@pytest.fixture
def app_with_sample_tasks(app):
task1_id = app.add_task("Learn pytest")
task2_id = app.add_task("Write tests")
return app, [task1_id, task2_id]# 📁 tests/integration/test_todo_flow.py
import os
import pytest
from src.app import TodoApp
from src.cache import FileCache
class TestTodoAppHappyPath:
"""Когда всё работает"""
def test_add_task(self, app):
task_id = app.add_task("Learn integration testing")
tasks = app.list_tasks()
assert len(tasks) == 1
assert tasks[0]["text"] == "Learn integration testing"
assert tasks[0]["done"] is False
def test_complete_task(self, app_with_sample_tasks):
app, task_ids = app_with_sample_tasks
task = app.complete_task(task_ids[0])
assert task["done"] is True
tasks = app.list_tasks()
assert tasks[0]["done"] is True
class TestTodoAppErrorHandling:
"""Обработка ошибок"""
def test_cache_degradation(self, temp_storage, spy_remote):
cache = FileCache(temp_storage)
app = TodoApp(cache, spy_remote)
task_id = app.add_task("Important task")
os.chmod(temp_storage, 0o000) # Ломаем доступ к кешу
task = app.complete_task(task_id)
assert task["done"] is True
assert spy_remote.assert_log_contains("cache_unavailable")
def test_remote_unavailable(self, working_cache, broken_remote):
app = TodoApp(working_cache, broken_remote)
task_id = app.add_task("Offline task")
assert task_id is not None
tasks = app.list_tasks()
assert len(tasks) == 1
def test_complete_missing_task(self, app):
with pytest.raises(ValueError, match="Task unknown not found"):
app.complete_task("unknown")
class TestTodoAppEdgeCases:
"""Граничные случаи"""
@pytest.mark.parametrize("text", [
" task with spaces ",
"very-long-task-" * 10,
"task with 🎉 emoji",
"", # Пустая задача
])
def test_add_various_tasks(self, app, text):
if not text.strip():
with pytest.raises(ValueError):
app.add_task(text)
else:
task_id = app.add_task(text)
assert task_id is not None
tasks = app.list_tasks()
assert tasks[0]["text"] == text.strip()Задача 1: полный цикл
def test_full_todo_workflow():
"""Добавление → завершение → проверка состояния"""
# 1. Добавьте несколько задач через app
# 2. Завершите одну
# 3. Проверьте: задача выполнена, остальные не изменились, кеш инвалидирован
passЗадача 2: восстановление после сбоя кеша
def test_recovery_after_cache_failure():
"""Приложение восстанавливается после падения кеша"""
# 1. Добавьте задачу при работающем кеше
# 2. Сымитируйте падение кеша (удалите файлы/права)
# 3. Завершите задачу — должно логировать сбой, но завершить
# 4. Восстановите кеш и добавьте новую задачу
passПоздравляю! Вы прошли путь от первого теста до профессионального тестирования реального приложения. 🎉
Теперь вы готовы тестировать любые проекты с уверенностью!
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.
Далее: Зачем это всё: тесты как суперсила джуна