Запускаем FastAPI сервис в Docker через Testcontainers. Фикстура для управления жизненным циклом контейнера. Тестируем по реальной HTTP-сети. Таймауты, connection errors, retry-логика
Unit тесты с моками проверяют что ваш код вызывает API. Integration тесты с Testcontainers проверяют что ваш код работает с реальным API по сети.
Цель: Научиться запускать сервис в Docker, тестировать по реальной HTTP-сети и ловить проблемы которые моки пропускают.
Вы должны уметь:
# Docker Compose (из Module 0)
docker-compose up -d
# Pytest фикстуры
@pytest.fixture(scope="session")
def service():
# Setup
yield
# TeardownЕсли не уверены — вернитесь к Module 0: Docker PostgreSQL.
# src/todo_client.py
import requests
class TodoClient:
def __init__(self, base_url):
self.base_url = base_url
def create_todo(self, title, description):
"""Создать todo через API"""
response = requests.post(
f"{self.base_url}/todos",
json={"title": title, "description": description}
)
return response.json()Unit тест с моком:
# tests/test_todo_client.py
from unittest.mock import patch, Mock
@patch('requests.post')
def test_create_todo_unit(mock_post):
"""Unit тест — мокируем requests"""
# Мок возвращает что мы хотим
mock_response = Mock()
mock_response.json.return_value = {"id": 1, "title": "Test"}
mock_post.return_value = mock_response
client = TodoClient("http://api.example.com")
result = client.create_todo("Test", "Description")
assert result["id"] == 1✅ Тест проходит!
Но в production:
# ❌ Реальный API возвращает 500 Internal Server Error
# ❌ Реальный API требует авторизацию
# ❌ Реальный API имеет таймауты 30 секунд
# ❌ Мок ничего из этого не проверяет!pip install testcontainers requestsСоздайте простой API для тестирования:
# api/main.py
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
todos = []
class TodoCreate(BaseModel):
title: str
description: str
class TodoResponse(BaseModel):
id: int
title: str
description: str
@app.post("/todos", response_model=TodoResponse)
def create_todo(todo: TodoCreate):
"""Создать todo"""
new_todo = {
"id": len(todos) + 1,
"title": todo.title,
"description": todo.description
}
todos.append(new_todo)
return new_todo
@app.get("/todos", response_model=list[TodoResponse])
def list_todos():
"""Список todos"""
return todos# api/Dockerfile
FROM python:3.11-slim
WORKDIR /app
RUN pip install fastapi uvicorn
COPY main.py .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]# tests/conftest.py
import pytest
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_strategies import wait_for_logs
import requests
import time
@pytest.fixture(scope="session")
def api_container():
"""Запустить FastAPI в Docker"""
# Создать контейнер
container = DockerContainer("api:latest")
container.with_exposed_ports(8000)
container.with_bind_ports(8000, 8000)
# Запустить
container.start()
# Подождать пока API готов
wait_for_logs(container, "Uvicorn running", timeout=30)
time.sleep(2) # Дополнительная задержка
yield container
# Остановить после тестов
container.stop()
@pytest.fixture(scope="session")
def api_base_url(api_container):
"""URL для API"""
host = api_container.get_container_host_ip()
port = api_container.get_exposed_port(8000)
return f"http://{host}:{port}"# tests/test_todo_integration.py
import requests
def test_create_todo_integration(api_base_url):
"""Integration тест — реальный HTTP"""
# Создайте todo через реальный API
response = requests.post(
f"{api_base_url}/todos",
json={"title": "Test Todo", "description": "Test Description"}
)
# Проверьте статус
assert response.status_code == 200
# Проверьте ответ
data = response.json()
assert data["id"] == 1
assert data["title"] == "Test Todo"
assert data["description"] == "Test Description"
def test_list_todos_integration(api_base_url):
"""Тест списка todos"""
# Создайте todo
requests.post(
f"{api_base_url}/todos",
json={"title": "Todo 1", "description": "Desc 1"}
)
# Получите список
response = requests.get(f"{api_base_url}/todos")
assert response.status_code == 200
todos = response.json()
assert len(todos) >= 1
assert todos[0]["title"] == "Todo 1"# Соберите Docker image для API
docker build -t api:latest api/
# Запустите тесты
pytest tests/test_todo_integration.py -v
# Вывод:
# tests/test_todo_integration.py::test_create_todo_integration PASSED
# tests/test_todo_integration.py::test_list_todos_integration PASSED✅ Тесты работают с реальным HTTP!
# api/main.py
import time
@app.post("/todos-slow", response_model=TodoResponse)
def create_todo_slow(todo: TodoCreate):
"""Медленный endpoint"""
time.sleep(5) # Задержка 5 секунд
return create_todo(todo)def test_timeout_handling(api_base_url):
"""Тест таймаута"""
try:
response = requests.post(
f"{api_base_url}/todos-slow",
json={"title": "Test", "description": "Desc"},
timeout=2 # Таймаут 2 секунды
)
assert False, "Должен был быть таймаут!"
except requests.Timeout:
# ✅ Таймаут произошёл как ожидалось
pass# src/todo_client.py
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class TodoClient:
def __init__(self, base_url):
self.base_url = base_url
# Настройка retry
session = requests.Session()
retries = Retry(
total=3,
backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retries)
session.mount("http://", adapter)
session.mount("https://", adapter)
self.session = session
def create_todo(self, title, description):
response = self.session.post(
f"{self.base_url}/todos",
json={"title": title, "description": description}
)
return response.json()# api/main.py
retry_count = 0
@app.post("/todos-flaky", response_model=TodoResponse)
def create_todo_flaky(todo: TodoCreate):
"""Endpoint который падает первые 2 раза"""
global retry_count
retry_count += 1
if retry_count < 3:
raise HTTPException(status_code=500, detail="Temporary error")
return create_todo(todo)# tests/test_retry.py
def test_retry_logic(api_base_url):
"""Тест что клиент делает retry"""
client = TodoClient(api_base_url)
# Первые 2 запроса упадут, 3-й успешный
result = client.create_todo("Test", "Description")
assert result["title"] == "Test"
# ✅ Retry сработал!def test_connection_error():
"""Тест когда API недоступен"""
client = TodoClient("http://localhost:9999") # Несуществующий порт
try:
client.create_todo("Test", "Description")
assert False, "Должна была быть ошибка подключения!"
except requests.ConnectionError:
# ✅ Ошибка обработана
passВопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.
Далее: API + БД: сквозные production-сценарии