TestClient, фикстуры, мокирование зависимостей, интеграционное тестирование, тестирование WebSocket
TestClient, фикстуры, мокирование зависимостей, интеграционное тестирование
Starlite предоставляет TestClient для тестирования приложений без запуска реального сервера.
from starlite import Starlite, get
from starlite.testing import TestClient
@get("/hello")
def hello() -> str:
return "Hello, World!"
app = Starlite(route_handlers=[hello])
def test_hello():
with TestClient(app=app) as client:
response = client.get("/hello")
assert response.status_code == 200
assert response.json() == "Hello, World!"from starlite.testing import TestClient
def test_all_methods():
with TestClient(app) as client:
# GET
response = client.get("/items")
assert response.status_code == 200
# POST
response = client.post(
"/items",
json={"name": "Item 1", "price": 9.99}
)
assert response.status_code == 201
# PUT
response = client.put(
"/items/1",
json={"name": "Updated", "price": 19.99}
)
assert response.status_code == 200
# DELETE
response = client.delete("/items/1")
assert response.status_code == 204
# PATCH
response = client.patch(
"/items/1",
json={"price": 29.99}
)
assert response.status_code == 200def test_query_params():
with TestClient(app) as client:
response = client.get(
"/items",
params={"page": 1, "page_size": 20}
)
assert response.status_code == 200
# С несколькими значениями
response = client.get(
"/search",
params={"tags": ["python", "web"]}
)
assert response.status_code == 200def test_headers():
with TestClient(app) as client:
response = client.get(
"/protected",
headers={"Authorization": "Bearer token123"}
)
assert response.status_code == 200
# Custom заголовки
response = client.get(
"/api/data",
headers={"X-API-Key": "secret-key"}
)
assert response.status_code == 200def test_cookies():
with TestClient(app) as client:
# Отправка cookies
response = client.get(
"/dashboard",
cookies={"session_id": "abc123"}
)
assert response.status_code == 200
# Получение cookies из ответа
response = client.post("/login", json={...})
assert "session" in response.cookies# conftest.py
import pytest
from starlite import Starlite
from starlite.testing import TestClient
@pytest.fixture
def app() -> Starlite:
return Starlite(route_handlers=[...])
@pytest.fixture
def client(app: Starlite) -> TestClient:
return TestClient(app=app)# test_api.py
def test_hello(client: TestClient):
response = client.get("/hello")
assert response.status_code == 200# conftest.py
import pytest
from starlite import Starlite, Provide
from starlite.testing import TestClient
@pytest.fixture
def database():
db = TestDatabase()
db.create_tables()
yield db
db.drop_tables()
@pytest.fixture
def app(database) -> Starlite:
return Starlite(
route_handlers=[...],
dependencies={
"db": Provide(lambda: database),
},
)
@pytest.fixture
def client(app: Starlite) -> TestClient:
return TestClient(app=app)from starlite.testing import TestClient
def test_with_mock_dependency(client: TestClient):
# Мок зависимости
mock_repo = MockUserRepository()
mock_repo.get_by_id.return_value = User(id=1, name="Test")
# Переопределение
client.app.dependency_overrides["repo"] = Provide(
lambda: mock_repo
)
response = client.get("/users/1")
assert response.status_code == 200
assert response.json()["name"] == "Test"
# Очистка
client.app.dependency_overrides.clear()from unittest.mock import patch, MagicMock
def test_external_service():
with patch("myapp.services.ExternalAPI") as mock_api:
mock_api.return_value.get_data.return_value = {"key": "value"}
response = client.get("/external-data")
assert response.status_code == 200
assert response.json() == {"key": "value"}import pytest
from unittest.mock import AsyncMock, MagicMock
@pytest.fixture
def mock_db_session():
session = MagicMock()
session.query = MagicMock()
session.add = MagicMock()
session.commit = MagicMock()
return session
@pytest.fixture
def mock_db(mock_db_session):
with patch("myapp.dependencies.get_db") as mock:
mock.return_value = mock_db_session
yield mock_db_session
def test_create_user(client: TestClient, mock_db):
mock_db.query().filter().first.return_value = None
mock_db.add.return_value = None
response = client.post(
"/users",
json={"name": "Test", "email": "test@example.com"}
)
assert response.status_code == 201
mock_db.commit.assert_called_once()import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
@pytest.fixture
async def db_engine():
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
echo=True,
)
yield engine
await engine.dispose()
@pytest.fixture
async def db_session(db_engine):
async_session = sessionmaker(
db_engine,
class_=AsyncSession,
expire_on_commit=False,
)
async with async_session() as session:
yield session
@pytest.fixture
def app(db_session):
return Starlite(
route_handlers=[...],
dependencies={
"session": Provide(lambda: db_session),
},
)
@pytest.fixture
def client(app):
return TestClient(app=app)
@pytest.mark.asyncio
async def test_create_user_db(client: TestClient, db_session):
response = client.post(
"/users",
json={"name": "Test", "email": "test@example.com"}
)
assert response.status_code == 201
# Проверка в БД
from sqlalchemy import select
from myapp.models import User
result = await db_session.execute(
select(User).where(User.email == "test@example.com")
)
user = result.scalar_one()
assert user is not None
assert user.name == "Test"import pytest
import redis.asyncio as redis
@pytest.fixture
async def redis_client():
client = redis.Redis(host="localhost", port=6379)
yield client
await client.flushdb()
await client.close()
@pytest.fixture
def app(redis_client):
return Starlite(
route_handlers=[...],
dependencies={
"redis": Provide(lambda: redis_client),
},
)
@pytest.mark.asyncio
async def test_cache(client: TestClient, redis_client):
# Первый запрос — кэш пуст
response = client.get("/data/1")
assert response.status_code == 200
# Проверка кэша
cached = await redis_client.get("data:1")
assert cached is not None
# Второй запрос — из кэша
response = client.get("/data/1")
assert response.status_code == 200from starlite.testing import TestClient
def test_websocket_echo(client: TestClient):
with client.websocket_connect("/ws") as websocket:
websocket.send_text("hello")
data = websocket.receive_text()
assert data == "Echo: hello"def test_websocket_json(client: TestClient):
with client.websocket_connect("/ws") as websocket:
websocket.send_json({"action": "subscribe", "channel": "news"})
data = websocket.receive_json()
assert data["status"] == "subscribed"def test_websocket_broadcast(client: TestClient):
# Подключение первого клиента
ws1 = client.websocket_connect("/ws")
ws1.__enter__()
# Подключение второго клиента
ws2 = client.websocket_connect("/ws")
ws2.__enter__()
try:
# Отправка сообщения через первого клиента
ws1.send_text("hello from ws1")
# Оба клиента должны получить сообщение
data1 = ws1.receive_text()
data2 = ws2.receive_text()
assert "hello from ws1" in data1
assert "hello from ws1" in data2
finally:
ws1.__exit__(None, None, None)
ws2.__exit__(None, None, None)from starlite.exceptions import NotFoundException
def test_not_found(client: TestClient):
response = client.get("/users/999")
assert response.status_code == 404
assert response.json()["detail"] == "User not found"
def test_validation_error(client: TestClient):
response = client.post(
"/users",
json={"name": ""} # Пустое имя
)
assert response.status_code == 400
assert "validation" in response.json()["detail"].lower()def test_custom_exception_handler(client: TestClient):
response = client.get("/protected")
assert response.status_code == 403
assert response.json()["error"] == "access_denied"import pytest
@pytest.mark.parametrize(
"email,expected_status",
[
("valid@example.com", 201),
("invalid-email", 400),
("@example.com", 400),
("user@", 400),
],
)
def test_create_user_validation(
client: TestClient,
email: str,
expected_status: int,
):
response = client.post(
"/users",
json={"name": "Test", "email": email}
)
assert response.status_code == expected_statusdef test_logging_middleware(client: TestClient, caplog):
with caplog.at_level("INFO"):
response = client.get("/hello")
assert response.status_code == 200
assert "GET /hello" in caplog.text
assert response.status_code in caplog.text# ✅ Хорошо — фикстуры в conftest.py
@pytest.fixture
def client():
return TestClient(app=app)
def test_hello(client):
...
def test_goodbye(client):
...
# ❌ Плохо — дублирование
def test_hello():
client = TestClient(app=app)
...
def test_goodbye():
client = TestClient(app=app)
...# ✅ Хорошо — каждый тест независим
def test_create_user(client):
response = client.post("/users", json={...})
assert response.status_code == 201
def test_get_user(client):
# Создаёт свои данные
client.post("/users", json={...})
response = client.get("/users/1")
assert response.status_code == 200
# ❌ Плохо — тесты зависят друг от друга
def test_create_user(client):
response = client.post("/users", json={...})
# Создаёт user с id=1
def test_get_user(client):
# Ожидает, что test_create_user уже создал user с id=1
response = client.get("/users/1")# ✅ Хорошо — очистка в фикстуре
@pytest.fixture
def db_session():
session = create_session()
yield session
session.rollback()
session.close()
# ❌ Плохо — состояние остаётся
@pytest.fixture
def db_session():
return create_session()# ✅ Хорошо — транзакция откатывается
@pytest.fixture
def db_session():
session = create_session()
transaction = session.begin()
yield session
transaction.rollback()
# ❌ Плохо — данные остаются в БД
@pytest.fixture
def db_session():
session = create_session()
yield session
session.close()# ✅ Хорошо — тестирование граничных случаев
@pytest.mark.parametrize(
"page,expected",
[
(0, 400), # Граница: 0
(1, 200), # Минимальное валидное
(1000, 200), # Максимальное валидное
(1001, 400), # Граница: > 1000
],
)
def test_pagination(client, page, expected):
response = client.get("/items", params={"page": page})
assert response.status_code == expectedТестирование в Starlite включает:
В следующей теме мы изучим плагины и расширения.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.