pytest-aiohttp, TestClient, моки, интеграционные тесты с testcontainers, покрытие кода.
Научитесь писать unit- и интеграционные тесты для aiohttp-приложений с использованием TestClient, моков и фикстур
pip install pytest pytest-aiohttp pytest-cov
pip install httpx # Альтернатива TestClientpytest-aiohttp предоставляет фикстуры для тестирования aiohttp-приложений.
# tests/test_handlers.py
import pytest
from aiohttp import web
async def hello_handler(request):
return web.Response(text='Hello, World!')
async def test_hello(aiohttp_client):
"""Простой тест обработчика"""
app = web.Application()
app.router.add_get('/', hello_handler)
client = await aiohttp_client(app)
resp = await client.get('/')
assert resp.status == 200
text = await resp.text()
assert text == 'Hello, World!'# tests/conftest.py
import pytest
import asyncpg
from aiohttp import web
from app.main import create_app
from app.config import settings
@pytest.fixture
async def app():
"""Создание тестового приложения"""
app = create_app()
yield app
@pytest.fixture
async def client(aiohttp_client, app):
"""TestClient для запросов к приложению"""
return await aiohttp_client(app)# tests/test_tasks.py
import pytest
async def test_list_tasks(client):
"""Тест списка задач"""
resp = await client.get('/api/v1/tasks')
assert resp.status == 200
data = await resp.json()
assert 'tasks' in data
assert 'total' in data
assert isinstance(data['tasks'], list)# tests/test_tasks.py
from unittest.mock import AsyncMock, patch
async def test_get_task(client):
"""Тест получения задачи с моком репозитория"""
mock_task = {
'id': 1,
'title': 'Test Task',
'description': 'Description',
'completed': False,
'user_id': 1
}
# Мок метода get
with patch.object(
client.app['task_repo'],
'get',
new=AsyncMock(return_value=mock_task)
):
resp = await client.get('/api/v1/tasks/1')
assert resp.status == 200
data = await resp.json()
assert data['id'] == 1
assert data['title'] == 'Test Task'# tests/conftest.py
import pytest
import asyncpg
from app.repositories.tasks import TaskRepository
TEST_DB_URL = 'postgresql://postgres:postgres@localhost:5432/task_manager_test'
@pytest.fixture
async def db_pool():
"""Пул соединений с тестовой БД"""
pool = await asyncpg.create_pool(TEST_DB_URL)
yield pool
await pool.close()
@pytest.fixture
async def task_repo(db_pool):
"""Репозиторий с тестовой БД"""
return TaskRepository(db_pool)
@pytest.fixture
async def app_with_db(db_pool):
"""Приложение с подключением к тестовой БД"""
app = create_app()
app['db'] = db_pool
app['task_repo'] = TaskRepository(db_pool)
yield app# tests/conftest.py
import pytest
from app.utils.jwt import create_access_token
from datetime import timedelta
@pytest.fixture
def auth_token():
"""JWT-токен для аутентифицированных запросов"""
return create_access_token(
data={'sub': '1', 'email': 'test@example.com'},
expires_delta=timedelta(hours=1)
)
@pytest.fixture
def auth_headers(auth_token):
"""Заголовки с токеном"""
return {'Authorization': f'Bearer {auth_token}'}# tests/test_auth.py
async def test_list_tasks_requires_auth(client):
"""Тест требования аутентификации"""
resp = await client.get('/api/v1/tasks')
assert resp.status == 401
async def test_list_tasks_with_auth(client, auth_headers):
"""Тест списка задач с токеном"""
resp = await client.get('/api/v1/tasks', headers=auth_headers)
assert resp.status == 200
data = await resp.json()
assert 'tasks' in data# tests/test_tasks.py
import pytest
from unittest.mock import AsyncMock, patch
async def test_create_task(client, auth_headers):
"""Тест создания задачи"""
task_data = {
'title': 'New Task',
'description': 'Test description'
}
mock_created = {
'id': 1,
**task_data,
'completed': False,
'user_id': 1
}
with patch.object(
client.app['task_repo'],
'create',
new=AsyncMock(return_value=mock_created)
):
resp = await client.post(
'/api/v1/tasks',
json=task_data,
headers=auth_headers
)
assert resp.status == 201
data = await resp.json()
assert data['id'] == 1
assert data['title'] == 'New Task'
assert data['completed'] == Falseasync def test_create_task_validation(client, auth_headers):
"""Тест валидации данных"""
# Пустой title
resp = await client.post(
'/api/v1/tasks',
json={'title': ''},
headers=auth_headers
)
assert resp.status == 400
# Отсутствие title
resp = await client.post(
'/api/v1/tasks',
json={'description': 'No title'},
headers=auth_headers
)
assert resp.status == 400
# Слишком длинный title
resp = await client.post(
'/api/v1/tasks',
json={'title': 'x' * 300},
headers=auth_headers
)
assert resp.status == 400# tests/test_integration.py
import pytest
import asyncpg
from testcontainers.postgres import PostgresContainer
from app.main import create_app
from app.repositories.tasks import TaskRepository
@pytest.fixture(scope='session')
def postgres_container():
"""Запуск PostgreSQL в Docker для тестов"""
with PostgresContainer('postgres:15') as postgres:
yield postgres
@pytest.fixture
async def integration_app(postgres_container):
"""Приложение с реальной БД в контейнере"""
db_url = postgres_container.get_connection_url()
pool = await asyncpg.create_pool(db_url)
app = create_app()
app['db'] = pool
app['task_repo'] = TaskRepository(pool)
# Создание таблиц
await pool.execute('''
CREATE TABLE tasks (
id SERIAL PRIMARY KEY,
title VARCHAR(200) NOT NULL,
description TEXT,
completed BOOLEAN DEFAULT FALSE,
user_id INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
yield app
await pool.close()
async def test_full_crud(integration_app, aiohttp_client, auth_headers):
"""Полный тест CRUD операций"""
client = await aiohttp_client(integration_app)
# Create
resp = await client.post(
'/api/v1/tasks',
json={'title': 'Integration Test', 'description': 'Test'},
headers=auth_headers
)
assert resp.status == 201
created = await resp.json()
task_id = created['id']
# Read
resp = await client.get(f'/api/v1/tasks/{task_id}', headers=auth_headers)
assert resp.status == 200
# Update
resp = await client.put(
f'/api/v1/tasks/{task_id}',
json={'completed': True},
headers=auth_headers
)
assert resp.status == 200
updated = await resp.json()
assert updated['completed'] == True
# Delete
resp = await client.delete(
f'/api/v1/tasks/{task_id}',
headers=auth_headers
)
assert resp.status == 204
# Verify deleted
resp = await client.get(f'/api/v1/tasks/{task_id}', headers=auth_headers)
assert resp.status == 404# tests/test_validation.py
import pytest
@pytest.mark.parametrize('title,expected_status', [
('', 400),
('x' * 201, 400),
(None, 400),
('Valid Title', 201),
('A', 201),
('x' * 200, 201),
])
async def test_create_task_title_validation(
client, auth_headers, title, expected_status
):
"""Тест валидации title с разными значениями"""
data = {'title': title} if title is not None else {}
resp = await client.post(
'/api/v1/tasks',
json=data,
headers=auth_headers
)
assert resp.status == expected_status# Запуск тестов с покрытием
pytest --cov=app --cov-report=html
# Запуск с отчётом в консоль
pytest --cov=app --cov-report=term-missing
# Минимальное покрытие 80%
pytest --cov=app --cov-fail-under=80# pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
python_functions = test_*
asyncio_mode = auto
addopts =
-v
--cov=app
--cov-report=term-missing
--cov-fail-under=80# tests/test_websocket.py
import pytest
from aiohttp import web, WSMsgType
async def test_websocket_connection(aiohttp_client):
"""Тест WebSocket подключения"""
app = web.Application()
app.router.add_get('/ws', websocket_handler)
client = await aiohttp_client(app)
# Подключение к WebSocket
ws = await client.ws_connect('/ws')
# Получение приветствия
msg = await ws.receive()
assert msg.type == WSMsgType.TEXT
assert msg.data == 'Connected!'
# Отправка сообщения
await ws.send_str('Hello')
# Получение эха
msg = await ws.receive()
assert msg.type == WSMsgType.TEXT
assert msg.data == 'Echo: Hello'
# Закрытие
await ws.close()# tests/conftest.py
import pytest
@pytest.fixture
async def created_task(client, auth_headers):
"""Созданная задача для тестов"""
resp = await client.post(
'/api/v1/tasks',
json={'title': 'Test Task', 'description': 'Test'},
headers=auth_headers
)
return await resp.json()
async def test_update_task(client, auth_headers, created_task):
"""Тест обновления с фикстурой созданной задачи"""
task_id = created_task['id']
resp = await client.put(
f'/api/v1/tasks/{task_id}',
json={'completed': True},
headers=auth_headers
)
assert resp.status == 200
data = await resp.json()
assert data['completed'] == TrueУбедитесь, что вы понимаете:
Переходите к вопросам для закрепления.
Далее: Деплой и продакшен-настройки