Мини-проект: Todo с кешем и API

src/
├── models.py          # Todo модель
├── repository.py      # Работа с "БД" (in-memory)
├── cache.py           # Redis кеш
├── quotes_api.py      # HTTP API для цитат
└── todo_service.py    # Бизнес-логика
 
tests/
├── conftest.py        # Общие фикстуры
├── unit/
│   ├── test_repository.py
│   ├── test_cache.py
│   └── test_quotes_api.py
└── integration/
    └── test_todo_service.py

# Устанавливаем зависимости
pip install pytest pytest-cov freezegun requests-mock

# src/models.py
 
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
 
@dataclass
class Todo:
    id: int
    title: str
    completed: bool = False
    created_at: Optional[datetime] = None
    motivation_quote: Optional[str] = None
 
    def mark_completed(self):
        """Отметить как выполненное"""
 
        self.completed = True
 
    def add_motivation(self, quote: str):
        """Добавить мотивационную цитату"""
 
        self.motivation_quote = quote

# src/repository.py
 
from typing import Dict, List, Optional
from src.models import Todo
 
class TodoRepository:
    """In-memory хранилище Todo"""
 
    def __init__(self):
        self._storage: Dict[int, Todo] = {}
        self._next_id = 1
 
    def create(self, title: str) -> Todo:
        """Создаёт новый Todo"""
 
        from datetime import datetime
 
        todo = Todo(
            id=self._next_id,
            title=title,
            created_at=datetime.now()
        )
        self._storage[todo.id] = todo
        self._next_id += 1
        return todo
 
    def get(self, todo_id: int) -> Optional[Todo]:
        """Получает Todo по ID"""
 
        return self._storage.get(todo_id)
 
    def get_all(self) -> List[Todo]:
        """Получает все Todo"""
 
        return list(self._storage.values())
 
    def delete(self, todo_id: int) -> bool:
        """Удаляет Todo"""
 
        if todo_id in self._storage:
            del self._storage[todo_id]
            return True
        return False

# src/cache.py
 
from typing import Optional
import json
 
class RedisCache:
    """Простой кеш (можно заменить на реальный Redis)"""
 
    def __init__(self):
        self._cache: dict = {}
 
    def get(self, key: str) -> Optional[str]:
        """Получает значение из кеша"""
 
        return self._cache.get(key)
 
    def set(self, key: str, value: str, ttl: int = 300):
        """Сохраняет значение в кеш"""
 
        self._cache[key] = value
 
    def delete(self, key: str):
        """Удаляет значение из кеша"""
 
        if key in self._cache:
            del self._cache[key]
 
    def clear(self):
        """Очищает весь кеш"""
 
        self._cache.clear()

# src/quotes_api.py
 
import requests
 
class QuotesAPIClient:
    """Клиент для получения мотивационных цитат (сервис заведомо фейковый)"""
 
    def __init__(self, base_url: str = "https://quotes.invalid"):
        self.base_url = base_url
 
    def get_random_quote(self) -> str:
        """Получает случайную цитату"""
 
        try:
            response = requests.get(f"{self.base_url}/random", timeout=5)
            response.raise_for_status()
            data = response.json()
            return f"{data['content']} — {data['author']}"
        except requests.RequestException:
            # Сервис не существует, поэтому чаще всего попадём сюда без моков
            return "Stay motivated!"  # Fallback

# src/todo_service.py
 
from typing import List, Optional
from src.models import Todo
from src.repository import TodoRepository
from src.cache import RedisCache
from src.quotes_api import QuotesAPIClient
 
class TodoService:
    """Сервис для работы с Todo"""
 
    def __init__(
        self,
        repository: TodoRepository,
        cache: RedisCache,
        quotes_api: QuotesAPIClient
    ):
        self.repository = repository
        self.cache = cache
        self.quotes_api = quotes_api
 
    def create_todo(self, title: str, add_motivation: bool = False) -> Todo:
        """Создаёт новый Todo с опциональной мотивацией"""
 
        todo = self.repository.create(title)
 
        if add_motivation:
            quote = self.quotes_api.get_random_quote()
            todo.add_motivation(quote)
 
        # Инвалидируем кеш списка
        self.cache.delete("todos:all")
 
        return todo
 
    def get_todo(self, todo_id: int) -> Optional[Todo]:
        """Получает Todo (с кешированием)"""
 
        # Проверяем кеш
        cache_key = f"todo:{todo_id}"
        cached = self.cache.get(cache_key)
 
        if cached:
            import json
            data = json.loads(cached)
            return Todo(**data)
 
        # Если нет в кеше — берём из БД
        todo = self.repository.get(todo_id)
 
        if todo:
            # Сохраняем в кеш
            import json
            from dataclasses import asdict
            self.cache.set(cache_key, json.dumps(asdict(todo)))
 
        return todo
 
    def get_all_todos(self) -> List[Todo]:
        """Получает все Todo (с кешированием)"""
 
        # Проверяем кеш
        cached = self.cache.get("todos:all")
 
        if cached:
            import json
            data = json.loads(cached)
            return [Todo(**item) for item in data]
 
        # Если нет в кеше — берём из БД
        todos = self.repository.get_all()
 
        # Сохраняем в кеш
        import json
        from dataclasses import asdict
        self.cache.set("todos:all", json.dumps([asdict(t) for t in todos]))
 
        return todos
 
    def complete_todo(self, todo_id: int) -> bool:
        """Отмечает Todo как выполненное"""
 
        todo = self.repository.get(todo_id)
 
        if not todo:
            return False
 
        todo.mark_completed()
 
        # Инвалидируем кеш
        self.cache.delete(f"todo:{todo_id}")
        self.cache.delete("todos:all")
 
        return True

# tests/conftest.py
 
import pytest
from src.repository import TodoRepository
from src.cache import RedisCache
from src.quotes_api import QuotesAPIClient
from src.todo_service import TodoService
 
@pytest.fixture
def repository():
    """Чистый repository для каждого теста"""
 
    return TodoRepository()
 
@pytest.fixture
def cache():
    """Чистый кеш для каждого теста"""
 
    cache = RedisCache()
    yield cache
    cache.clear()
 
@pytest.fixture
def quotes_api():
    """Реальный API-клиент"""
 
    return QuotesAPIClient()
 
@pytest.fixture
def mock_quotes_api():
    """Мок API-клиента для unit-тестов"""
 
    from unittest.mock import Mock
    mock = Mock(spec=QuotesAPIClient)
    mock.get_random_quote.return_value = "Test quote — Test Author"
    return mock
 
@pytest.fixture
def todo_service(repository, cache, mock_quotes_api):
    """TodoService с моками"""
 
    return TodoService(
        repository=repository,
        cache=cache,
        quotes_api=mock_quotes_api
    )

# tests/unit/test_repository.py
 
import pytest
from freezegun import freeze_time
from datetime import datetime
 
@freeze_time("2025-01-15 10:00:00")
def test_create_todo(repository):
    """Тест создания Todo"""
 
    todo = repository.create("Buy milk")
 
    assert todo.id == 1
    assert todo.title == "Buy milk"
    assert todo.completed == False
    assert todo.created_at == datetime(2025, 1, 15, 10, 0, 0)
 
def test_get_todo(repository):
    """Тест получения Todo"""
 
    created = repository.create("Buy milk")
 
    todo = repository.get(created.id)
 
    assert todo.title == "Buy milk"
 
def test_get_nonexistent_todo(repository):
    """Тест получения несуществующего Todo"""
 
    todo = repository.get(999)
 
    assert todo is None
 
def test_get_all_todos(repository):
    """Тест получения всех Todo"""
 
    repository.create("Task 1")
    repository.create("Task 2")
    repository.create("Task 3")
 
    todos = repository.get_all()
 
    assert len(todos) == 3
    assert todos[0].title == "Task 1"
    assert todos[2].title == "Task 3"
 
def test_delete_todo(repository):
    """Тест удаления Todo"""
 
    todo = repository.create("Delete me")
 
    result = repository.delete(todo.id)
 
    assert result == True
    assert repository.get(todo.id) is None
 
def test_delete_nonexistent_todo(repository):
    """Тест удаления несуществующего Todo"""
 
    result = repository.delete(999)
 
    assert result == False

# tests/unit/test_cache.py
 
def test_cache_get_set(cache):
    """Тест сохранения и получения"""
 
    cache.set("key1", "value1")
 
    result = cache.get("key1")
 
    assert result == "value1"
 
def test_cache_get_nonexistent(cache):
    """Тест получения несуществующего ключа"""
 
    result = cache.get("nonexistent")
 
    assert result is None
 
def test_cache_delete(cache):
    """Тест удаления из кеша"""
 
    cache.set("key1", "value1")
    cache.delete("key1")
 
    result = cache.get("key1")
 
    assert result is None
 
def test_cache_clear(cache):
    """Тест очистки кеша"""
 
    cache.set("key1", "value1")
    cache.set("key2", "value2")
 
    cache.clear()
 
    assert cache.get("key1") is None
    assert cache.get("key2") is None

# tests/unit/test_quotes_api.py
 
import pytest
import requests_mock
 
def test_get_random_quote_success():
    """Тест успешного получения цитаты"""
 
    with requests_mock.Mocker() as m:
        m.get("https://quotes.invalid/random", json={
            "content": "Be yourself",
            "author": "Oscar Wilde"
        })
 
        client = QuotesAPIClient()
        quote = client.get_random_quote()
 
        assert quote == "Be yourself — Oscar Wilde"
 
def test_get_random_quote_network_error():
    """Тест обработки сетевой ошибки"""
 
    with requests_mock.Mocker() as m:
        m.get("https://quotes.invalid/random", exc=requests.ConnectionError)
 
        client = QuotesAPIClient()
        quote = client.get_random_quote()
 
        assert quote == "Stay motivated!"  # Fallback
 
def test_get_random_quote_timeout():
    """Тест обработки timeout"""
 
    with requests_mock.Mocker() as m:
        m.get("https://quotes.invalid/random", exc=requests.Timeout)
 
        client = QuotesAPIClient()
        quote = client.get_random_quote()
 
        assert quote == "Stay motivated!"

# tests/integration/test_todo_service.py
 
import pytest
 
def test_create_todo_without_motivation(todo_service):
    """Тест создания Todo без мотивации"""
 
    todo = todo_service.create_todo("Buy milk")
 
    assert todo.id == 1
    assert todo.title == "Buy milk"
    assert todo.motivation_quote is None
 
def test_create_todo_with_motivation(todo_service):
    """Тест создания Todo с мотивацией"""
 
    todo = todo_service.create_todo("Learn pytest", add_motivation=True)
 
    assert todo.motivation_quote == "Test quote — Test Author"
    # Проверяем что quotes_api был вызван
    todo_service.quotes_api.get_random_quote.assert_called_once()
 
def test_get_todo_from_cache(todo_service):
    """Тест получения Todo из кеша"""
 
    # Создаём Todo
    created = todo_service.create_todo("Task 1")
 
    # Первый вызов — из БД
    todo1 = todo_service.get_todo(created.id)
    assert todo1.title == "Task 1"
 
    # Второй вызов — из кеша (репозиторий не вызывается)
    todo2 = todo_service.get_todo(created.id)
    assert todo2.title == "Task 1"
 
def test_get_all_todos_cache_invalidation(todo_service):
    """Тест инвалидации кеша при создании"""
 
    # Получаем пустой список (кешируется)
    todos1 = todo_service.get_all_todos()
    assert len(todos1) == 0
 
    # Создаём Todo (кеш инвалидируется)
    todo_service.create_todo("New task")
 
    # Получаем обновлённый список
    todos2 = todo_service.get_all_todos()
    assert len(todos2) == 1
 
def test_complete_todo(todo_service):
    """Тест отметки Todo как выполненного"""
 
    todo = todo_service.create_todo("Task")
 
    result = todo_service.complete_todo(todo.id)
 
    assert result == True
    # Проверяем что Todo обновлён
    updated = todo_service.repository.get(todo.id)
    assert updated.completed == True
 
def test_complete_nonexistent_todo(todo_service):
    """Тест отметки несуществующего Todo"""
 
    result = todo_service.complete_todo(999)
 
    assert result == False

# pytest.ini
 
[pytest]
testpaths = tests
addopts =
    -v
    --tb=short
    --cov=src
    --cov-report=term-missing
    --cov-report=html
    --strict-markers
 
markers =
    unit: Unit-тесты
    integration: Integration-тесты

pytest

========================= test session starts ==========================
collected 18 items
 
tests/unit/test_repository.py::test_create_todo PASSED          [  5%]
tests/unit/test_repository.py::test_get_todo PASSED             [ 11%]
tests/unit/test_repository.py::test_get_nonexistent_todo PASSED [ 16%]
tests/unit/test_repository.py::test_get_all_todos PASSED        [ 22%]
tests/unit/test_repository.py::test_delete_todo PASSED          [ 27%]
tests/unit/test_repository.py::test_delete_nonexistent_todo ... [ 33%]
tests/unit/test_cache.py::test_cache_get_set PASSED             [ 38%]
tests/unit/test_cache.py::test_cache_get_nonexistent PASSED     [ 44%]
tests/unit/test_cache.py::test_cache_delete PASSED              [ 50%]
tests/unit/test_cache.py::test_cache_clear PASSED               [ 55%]
tests/unit/test_quotes_api.py::test_get_random_quote_success ... [ 61%]
tests/unit/test_quotes_api.py::test_get_random_quote_network ... [ 66%]
tests/unit/test_quotes_api.py::test_get_random_quote_timeout ... [ 72%]
tests/integration/test_todo_service.py::test_create_todo_wit... [ 77%]
tests/integration/test_todo_service.py::test_create_todo_wit... [ 83%]
tests/integration/test_todo_service.py::test_get_todo_from_c... [ 88%]
tests/integration/test_todo_service.py::test_get_all_todos_c... [ 94%]
tests/integration/test_todo_service.py::test_complete_todo P... [100%]
 
---------- coverage: platform darwin, python 3.11 -----------
Name                    Stmts   Miss  Cover   Missing
-----------------------------------------------------
src/models.py              10      0   100%
src/repository.py          20      0   100%
src/cache.py               12      0   100%
src/quotes_api.py          10      1    90%   18
src/todo_service.py        35      2    94%   45, 67
-----------------------------------------------------
TOTAL                      87      3    97%
 
==================== 18 passed in 1.23s ============================

open htmlcov/index.html