Юнит-тесты, моки, интеграционное тестирование, end-to-end проверки.
Нетестированный код — сломанный код. В этой теме научимся писать юнит-тесты, моки, интеграционное и end-to-end тестирование для MCP серверов.
╱▔▔▔▔▔▔▔▔▔╲
╱ E2E ╲
╱ тесты ╲
╱───────────────╲
╱ Интеграция ╲
╱───────────────────╲
╱ Юнит-тесты ╲
╱───────────────────────╲
Юнит-тесты (70%) — отдельные функции, инструменты без внешних зависимостей
Интеграционные (20%) — взаимодействие с БД, внешними API
E2E (10%) — полный цикл через MCP клиент
# tests/test_tools.py
import pytest
from src.server import mcp
class TestCalculatorTools:
"""Тесты инструментов калькулятора."""
def test_add_positive_numbers(self):
"""Сложение положительных чисел."""
result = mcp.get_tool("add").func(2, 3)
assert result == 5
def test_add_negative_numbers(self):
"""Сложение отрицательных чисел."""
result = mcp.get_tool("add").func(-2, -3)
assert result == -5
def test_subtract(self):
"""Вычитание."""
result = mcp.get_tool("subtract").func(10, 4)
assert result == 6
def test_multiply(self):
"""Умножение."""
result = mcp.get_tool("multiply").func(3, 7)
assert result == 21
def test_divide(self):
"""Деление."""
result = mcp.get_tool("divide").func(10, 2)
assert result == 5.0
def test_divide_by_zero(self):
"""Деление на ноль выбрасывает ошибку."""
with pytest.raises(ValueError, match="Деление на ноль"):
mcp.get_tool("divide").func(10, 0)# tests/test_database_tools.py
import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from src.server import mcp
class TestDatabaseTools:
"""Тесты инструментов базы данных."""
@pytest.fixture
def mock_db_pool(self):
"""Фикстура с моком пула БД."""
with patch('src.server.db_pool') as mock_pool:
yield mock_pool
@pytest.mark.asyncio
async def test_get_user_exists(self, mock_db_pool):
"""Получение существующего пользователя."""
# Настраиваем мок
mock_conn = AsyncMock()
mock_conn.fetchrow.return_value = {
"id": 1,
"username": "alice",
"email": "alice@example.com"
}
mock_pool = AsyncMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_db_pool.return_value = mock_pool
# Вызов инструмента
result = await mcp.get_tool("get_user").func(1)
# Проверки
assert result["username"] == "alice"
mock_conn.fetchrow.assert_called_once()
@pytest.mark.asyncio
async def test_get_user_not_found(self, mock_db_pool):
"""Получение несуществующего пользователя."""
mock_conn = AsyncMock()
mock_conn.fetchrow.return_value = None
mock_pool = AsyncMock()
mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
mock_db_pool.return_value = mock_pool
with pytest.raises(ValueError, match="User not found"):
await mcp.get_tool("get_user").func(999)# tests/test_resources.py
import pytest
from src.server import mcp
class TestResources:
"""Тесты ресурсов."""
def test_config_resource(self):
"""Чтение ресурса конфигурации."""
resource = mcp.get_resource("config://app/settings")
content = resource.func()
assert "APP_NAME" in content
assert "VERSION" in content
def test_system_time_resource(self):
"""Ресурс системного времени."""
resource = mcp.get_resource("system://clock")
content = resource.func()
# Проверяем формат ISO 8601
from datetime import datetime
try:
datetime.fromisoformat(content)
except ValueError:
pytest.fail("Invalid ISO format")# tests/test_validation.py
import pytest
from src.server import mcp
class TestValidation:
"""Тесты валидации входных данных."""
def test_search_empty_query(self):
"""Поиск с пустым запросом."""
with pytest.raises(ValueError, match="минимум 2 символа"):
mcp.get_tool("search").func("")
def test_search_short_query(self):
"""Поиск с коротким запросом."""
with pytest.raises(ValueError, match="минимум 2 символа"):
mcp.get_tool("search").func("a")
def test_search_valid_query(self):
"""Поиск с валидным запросом."""
# Мок базы данных
with patch('src.server.database.search') as mock_search:
mock_search.return_value = [{"id": 1, "title": "Result"}]
result = mcp.get_tool("search").func("python")
assert len(result) == 1
mock_search.assert_called_once()
def test_create_user_invalid_email(self):
"""Создание пользователя с невалидным email."""
with pytest.raises(ValueError, match="Невалидный email"):
mcp.get_tool("create_user").func(
name="test",
email="invalid-email"
)# tests/integration/test_database.py
import pytest
import asyncpg
from src.server import db_pool
@pytest.fixture(scope="module")
async def test_db():
"""Фикстура с тестовой базой данных."""
# Создаём тестовую БД
conn = await asyncpg.connect("postgresql://localhost/postgres")
await conn.execute("DROP DATABASE IF EXISTS test_mcp")
await conn.execute("CREATE DATABASE test_mcp")
await conn.close()
# Инициализируем пул
db_pool._dsn = "postgresql://localhost/test_mcp"
await db_pool.create_pool()
# Создаём таблицы
async with db_pool.acquire() as conn:
await conn.execute("""
CREATE TABLE users (
id SERIAL PRIMARY KEY,
username TEXT UNIQUE,
email TEXT
)
""")
yield db_pool
# Очистка
await db_pool.close_pool()
conn = await asyncpg.connect("postgresql://localhost/postgres")
await conn.execute("DROP DATABASE test_mcp")
await conn.close()
@pytest.mark.asyncio
async def test_create_and_get_user(test_db):
"""Создание и получение пользователя."""
async with test_db.acquire() as conn:
# Создаём пользователя
await conn.execute(
"INSERT INTO users (username, email) VALUES ($1, $2)",
"testuser", "test@example.com"
)
# Получаем пользователя
row = await conn.fetchrow(
"SELECT * FROM users WHERE username = $1",
"testuser"
)
assert row["email"] == "test@example.com"
@pytest.mark.asyncio
async def test_user_not_found(test_db):
"""Получение несуществующего пользователя."""
async with test_db.acquire() as conn:
row = await conn.fetchrow(
"SELECT * FROM users WHERE id = $1",
999
)
assert row is None# tests/integration/test_external_api.py
import pytest
from unittest.mock import patch
import aiohttp
from src.server import http_pool
@pytest.fixture
async def mock_http_server(aiohttp_server):
"""Фикстура с тестовым HTTP сервером."""
from aiohttp import web
async def handler(request):
return web.json_response({
"status": "success",
"data": {"id": 1, "name": "Test"}
})
app = web.Application()
app.router.add_get('/api/test', handler)
return await aiohttp_server(app)
@pytest.mark.asyncio
async def test_fetch_url(mock_http_server):
"""Получение URL с тестового сервера."""
url = f"http://{mock_http_server.host}:{mock_http_server.port}/api/test"
async with http_pool.get_session() as session:
async with session.get(url) as response:
data = await response.json()
assert data["status"] == "success"
assert data["data"]["id"] == 1# tests/e2e/test_e2e.py
import pytest
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
@pytest.fixture
async def mcp_client():
"""Фикстура с MCP клиентом."""
server_params = StdioServerParameters(
command="poetry",
args=["run", "python", "src/server.py"],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
yield session
@pytest.mark.asyncio
async def test_list_tools(mcp_client):
"""Получение списка инструментов."""
tools = await mcp_client.list_tools()
assert len(tools.tools) > 0
tool_names = [t.name for t in tools.tools]
assert "add" in tool_names
@pytest.mark.asyncio
async def test_call_tool(mcp_client):
"""Вызов инструмента через клиент."""
result = await mcp_client.call_tool("add", {"a": 2, "b": 3})
assert result.isError is False
assert result.content[0].text == "5"
@pytest.mark.asyncio
async def test_read_resource(mcp_client):
"""Чтение ресурса через клиент."""
result = await mcp_client.read_resource("config://app/settings")
assert result.contents[0].text is not None
assert "APP_NAME" in result.contents[0].text
@pytest.mark.asyncio
async def test_tool_error(mcp_client):
"""Обработка ошибки инструмента."""
result = await mcp_client.call_tool("divide", {"a": 10, "b": 0})
assert result.isError is True
assert "ноль" in result.content[0].text# tests/conftest.py
import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock
@pytest.fixture
def mock_database():
"""Мок базы данных."""
db = MagicMock()
db.get_user = AsyncMock(return_value={"id": 1, "name": "Test"})
db.create_user = AsyncMock(return_value={"id": 1})
db.delete_user = AsyncMock()
return db
@pytest.fixture
def mock_external_api():
"""Мок внешнего API."""
api = MagicMock()
api.fetch = AsyncMock(return_value={"status": "success"})
api.post = AsyncMock(return_value={"id": 1})
return api
@pytest.fixture
def sample_user():
"""Пример данных пользователя."""
return {
"id": 1,
"username": "testuser",
"email": "test@example.com",
"role": "user"
}
@pytest.fixture
def sample_auth_context():
"""Пример контекста аутентификации."""
return {
"user_id": "1",
"role": "admin",
"token": "test-token"
}# tests/utils.py
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def timeout_context(seconds: float):
"""Контекст для тестирования таймаутов."""
try:
async with asyncio.timeout(seconds):
yield
except asyncio.TimeoutError:
pytest.fail(f"Operation timed out after {seconds}s")
def assert_raises_async(exception_type, coro, match=None):
"""Проверка, что корутина выбрасывает исключение."""
import re
try:
asyncio.run(coro)
pytest.fail(f"Expected {exception_type.__name__} to be raised")
except exception_type as e:
if match and not re.search(match, str(e)):
pytest.fail(f"Exception message '{e}' doesn't match '{match}'")# Установка
poetry add --group dev pytest-cov
# Запуск тестов с покрытием
poetry run pytest tests/ --cov=src --cov-report=html
# Открыть отчёт
open htmlcov/index.html# .coveragerc
[run]
source = src
omit =
*/tests/*
*/__init__.py
*/conftest.py
[report]
exclude_lines =
pragma: no cover
def __repr__
raise AssertionError
raise NotImplementedError
if __name__ == .__main__.:
[html]
directory = htmlcov# .gitlab-ci.yml
test:
stage: test
image: python:3.11
script:
- pip install poetry
- poetry install
- poetry run pytest tests/ --cov=src --cov-report=xml
- poetry run coverage report --fail-under=80
artifacts:
reports:
coverage_report:
coverage_format: cobertura
path: coverage.xml# ❌ ПЛОХО: Тесты зависят друг от друга
def test_create_user():
user = create_user("test")
assert user.id == 1
def test_get_user():
# Зависит от test_create_user
user = get_user(1)
assert user.username == "test"
# ✅ ХОРОШО: Каждый тест независим
def test_create_user():
user = create_user("test")
assert user.id == 1
def test_get_user():
# Создаём свои данные
user = create_user("another")
fetched = get_user(user.id)
assert fetched.username == "another"def test_edge_cases():
"""Тестирование граничных значений."""
# Пустые значения
assert validate("") == False
# Максимальные значения
assert validate("a" * 1000) == True
# Специальные символы
assert validate("<script>alert(1)</script>") == False
# Unicode
assert validate("Привет") == True@pytest.mark.parametrize("a,b,expected", [
(1, 2, 3),
(0, 0, 0),
(-1, 1, 0),
(100, 200, 300),
])
def test_add(a, b, expected):
assert add(a, b) == expected| Тип тестов | Что тестировать | Инструменты |
|---|---|---|
| Юнит | Отдельные функции, инструменты | pytest, unittest.mock |
| Интеграция | Взаимодействие с БД, API | testcontainers, aiohttp_server |
| E2E | Полный цикл через MCP клиент | mcp.ClientSession |
Следующая тема: Деплой и мониторинг — Docker, Kubernetes, логирование, метрики.
Далее: Деплой и мониторинг