846 lines
27 KiB
Markdown
846 lines
27 KiB
Markdown
# 🧪 Тестирование системы аутентификации
|
||
|
||
## 🎯 Обзор
|
||
|
||
Комплексная стратегия тестирования системы аутентификации с unit, integration и E2E тестами.
|
||
|
||
## 🏗️ Структура тестов
|
||
|
||
```
|
||
tests/auth/
|
||
├── unit/
|
||
│ ├── test_session_manager.py
|
||
│ ├── test_oauth_manager.py
|
||
│ ├── test_batch_operations.py
|
||
│ ├── test_monitoring.py
|
||
│ └── test_utils.py
|
||
├── integration/
|
||
│ ├── test_redis_integration.py
|
||
│ ├── test_oauth_flow.py
|
||
│ ├── test_middleware.py
|
||
│ └── test_decorators.py
|
||
├── e2e/
|
||
│ ├── test_login_flow.py
|
||
│ ├── test_oauth_flow.py
|
||
│ └── test_session_management.py
|
||
└── fixtures/
|
||
├── auth_fixtures.py
|
||
├── redis_fixtures.py
|
||
└── oauth_fixtures.py
|
||
```
|
||
|
||
## 🔧 Unit Tests
|
||
|
||
### SessionTokenManager Tests
|
||
|
||
```python
|
||
import pytest
|
||
from unittest.mock import AsyncMock, patch
|
||
from auth.tokens.sessions import SessionTokenManager
|
||
|
||
class TestSessionTokenManager:
|
||
|
||
@pytest.fixture
|
||
def session_manager(self):
|
||
return SessionTokenManager()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_session(self, session_manager):
|
||
"""Тест создания сессии"""
|
||
with patch('auth.tokens.sessions.redis') as mock_redis:
|
||
mock_redis.hset = AsyncMock()
|
||
mock_redis.sadd = AsyncMock()
|
||
mock_redis.expire = AsyncMock()
|
||
|
||
token = await session_manager.create_session(
|
||
user_id="123",
|
||
username="testuser"
|
||
)
|
||
|
||
assert token is not None
|
||
assert len(token) > 20
|
||
mock_redis.hset.assert_called()
|
||
mock_redis.sadd.assert_called()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_verify_session_valid(self, session_manager):
|
||
"""Тест проверки валидной сессии"""
|
||
with patch('auth.jwtcodec.decode_jwt') as mock_decode:
|
||
mock_decode.return_value = {
|
||
"user_id": "123",
|
||
"username": "testuser",
|
||
"exp": int(time.time()) + 3600
|
||
}
|
||
|
||
with patch('auth.tokens.sessions.redis') as mock_redis:
|
||
mock_redis.exists.return_value = True
|
||
|
||
payload = await session_manager.verify_session("valid_token")
|
||
|
||
assert payload is not None
|
||
assert payload["user_id"] == "123"
|
||
assert payload["username"] == "testuser"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_verify_session_invalid(self, session_manager):
|
||
"""Тест проверки невалидной сессии"""
|
||
with patch('auth.jwtcodec.decode_jwt') as mock_decode:
|
||
mock_decode.return_value = None
|
||
|
||
payload = await session_manager.verify_session("invalid_token")
|
||
|
||
assert payload is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_revoke_session_token(self, session_manager):
|
||
"""Тест отзыва токена сессии"""
|
||
with patch('auth.tokens.sessions.redis') as mock_redis:
|
||
mock_redis.delete = AsyncMock(return_value=1)
|
||
mock_redis.srem = AsyncMock()
|
||
|
||
result = await session_manager.revoke_session_token("test_token")
|
||
|
||
assert result is True
|
||
mock_redis.delete.assert_called()
|
||
mock_redis.srem.assert_called()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_user_sessions(self, session_manager):
|
||
"""Тест получения сессий пользователя"""
|
||
with patch('auth.tokens.sessions.redis') as mock_redis:
|
||
mock_redis.smembers.return_value = {b"token1", b"token2"}
|
||
mock_redis.hgetall.return_value = {
|
||
b"user_id": b"123",
|
||
b"username": b"testuser",
|
||
b"last_activity": b"1640995200"
|
||
}
|
||
|
||
sessions = await session_manager.get_user_sessions("123")
|
||
|
||
assert len(sessions) == 2
|
||
assert sessions[0]["token"] == "token1"
|
||
assert sessions[0]["user_id"] == "123"
|
||
```
|
||
|
||
### OAuthTokenManager Tests
|
||
|
||
```python
|
||
import pytest
|
||
from unittest.mock import AsyncMock, patch
|
||
from auth.tokens.oauth import OAuthTokenManager
|
||
|
||
class TestOAuthTokenManager:
|
||
|
||
@pytest.fixture
|
||
def oauth_manager(self):
|
||
return OAuthTokenManager()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_store_oauth_tokens(self, oauth_manager):
|
||
"""Тест сохранения OAuth токенов"""
|
||
with patch('auth.tokens.oauth.redis') as mock_redis:
|
||
mock_redis.setex = AsyncMock()
|
||
|
||
await oauth_manager.store_oauth_tokens(
|
||
user_id="123",
|
||
provider="google",
|
||
access_token="access_token_123",
|
||
refresh_token="refresh_token_123",
|
||
expires_in=3600
|
||
)
|
||
|
||
# Проверяем, что токены сохранены
|
||
assert mock_redis.setex.call_count == 2 # access + refresh
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_token(self, oauth_manager):
|
||
"""Тест получения OAuth токена"""
|
||
with patch('auth.tokens.oauth.redis') as mock_redis:
|
||
mock_redis.get.return_value = b'{"token": "access_token_123", "expires_in": 3600}'
|
||
|
||
token_data = await oauth_manager.get_token("123", "google", "oauth_access")
|
||
|
||
assert token_data is not None
|
||
assert token_data["token"] == "access_token_123"
|
||
assert token_data["expires_in"] == 3600
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_revoke_oauth_tokens(self, oauth_manager):
|
||
"""Тест отзыва OAuth токенов"""
|
||
with patch('auth.tokens.oauth.redis') as mock_redis:
|
||
mock_redis.delete = AsyncMock(return_value=2)
|
||
|
||
result = await oauth_manager.revoke_oauth_tokens("123", "google")
|
||
|
||
assert result is True
|
||
mock_redis.delete.assert_called()
|
||
```
|
||
|
||
### BatchTokenOperations Tests
|
||
|
||
```python
|
||
import pytest
|
||
from unittest.mock import AsyncMock, patch
|
||
from auth.tokens.batch import BatchTokenOperations
|
||
|
||
class TestBatchTokenOperations:
|
||
|
||
@pytest.fixture
|
||
def batch_operations(self):
|
||
return BatchTokenOperations()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_batch_validate_tokens(self, batch_operations):
|
||
"""Тест массовой валидации токенов"""
|
||
tokens = ["token1", "token2", "token3"]
|
||
|
||
with patch.object(batch_operations, '_validate_token_batch') as mock_validate:
|
||
mock_validate.return_value = {
|
||
"token1": True,
|
||
"token2": False,
|
||
"token3": True
|
||
}
|
||
|
||
results = await batch_operations.batch_validate_tokens(tokens)
|
||
|
||
assert results["token1"] is True
|
||
assert results["token2"] is False
|
||
assert results["token3"] is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_batch_revoke_tokens(self, batch_operations):
|
||
"""Тест массового отзыва токенов"""
|
||
tokens = ["token1", "token2", "token3"]
|
||
|
||
with patch.object(batch_operations, '_revoke_token_batch') as mock_revoke:
|
||
mock_revoke.return_value = 2 # 2 токена отозваны
|
||
|
||
revoked_count = await batch_operations.batch_revoke_tokens(tokens)
|
||
|
||
assert revoked_count == 2
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cleanup_expired_tokens(self, batch_operations):
|
||
"""Тест очистки истекших токенов"""
|
||
with patch('auth.tokens.batch.redis') as mock_redis:
|
||
# Мокаем поиск истекших токенов
|
||
mock_redis.scan_iter.return_value = [
|
||
"session:123:expired_token1",
|
||
"session:456:expired_token2"
|
||
]
|
||
mock_redis.ttl.return_value = -1 # Истекший токен
|
||
mock_redis.delete = AsyncMock(return_value=1)
|
||
|
||
cleaned_count = await batch_operations.cleanup_expired_tokens()
|
||
|
||
assert cleaned_count >= 0
|
||
```
|
||
|
||
## 🔗 Integration Tests
|
||
|
||
### Redis Integration Tests
|
||
|
||
```python
|
||
import pytest
|
||
import asyncio
|
||
from storage.redis import redis
|
||
from auth.tokens.sessions import SessionTokenManager
|
||
|
||
class TestRedisIntegration:
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_redis_connection(self):
|
||
"""Тест подключения к Redis"""
|
||
result = await redis.ping()
|
||
assert result is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_session_lifecycle(self):
|
||
"""Тест полного жизненного цикла сессии"""
|
||
sessions = SessionTokenManager()
|
||
|
||
# Создаем сессию
|
||
token = await sessions.create_session(
|
||
user_id="test_user",
|
||
username="testuser"
|
||
)
|
||
|
||
assert token is not None
|
||
|
||
# Проверяем сессию
|
||
payload = await sessions.verify_session(token)
|
||
assert payload is not None
|
||
assert payload["user_id"] == "test_user"
|
||
|
||
# Получаем сессии пользователя
|
||
user_sessions = await sessions.get_user_sessions("test_user")
|
||
assert len(user_sessions) >= 1
|
||
|
||
# Отзываем сессию
|
||
revoked = await sessions.revoke_session_token(token)
|
||
assert revoked is True
|
||
|
||
# Проверяем, что сессия отозвана
|
||
payload = await sessions.verify_session(token)
|
||
assert payload is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_concurrent_sessions(self):
|
||
"""Тест множественных сессий"""
|
||
sessions = SessionTokenManager()
|
||
|
||
# Создаем несколько сессий одновременно
|
||
tasks = []
|
||
for i in range(5):
|
||
task = sessions.create_session(
|
||
user_id="concurrent_user",
|
||
username=f"user_{i}"
|
||
)
|
||
tasks.append(task)
|
||
|
||
tokens = await asyncio.gather(*tasks)
|
||
|
||
# Проверяем, что все токены созданы
|
||
assert len(tokens) == 5
|
||
assert all(token is not None for token in tokens)
|
||
|
||
# Проверяем, что все сессии валидны
|
||
for token in tokens:
|
||
payload = await sessions.verify_session(token)
|
||
assert payload is not None
|
||
|
||
# Очищаем тестовые данные
|
||
for token in tokens:
|
||
await sessions.revoke_session_token(token)
|
||
```
|
||
|
||
### OAuth Flow Integration Tests
|
||
|
||
```python
|
||
import pytest
|
||
from unittest.mock import AsyncMock, patch
|
||
from auth.oauth import oauth_login_http, oauth_callback_http
|
||
|
||
class TestOAuthIntegration:
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_state_flow(self):
|
||
"""Тест OAuth state flow"""
|
||
from auth.oauth import store_oauth_state, get_oauth_state
|
||
|
||
# Сохраняем state
|
||
state = "test_state_123"
|
||
redirect_uri = "http://localhost:3000"
|
||
|
||
await store_oauth_state(state, redirect_uri)
|
||
|
||
# Получаем state
|
||
stored_data = await get_oauth_state(state)
|
||
|
||
assert stored_data is not None
|
||
assert stored_data["redirect_uri"] == redirect_uri
|
||
|
||
# Проверяем, что state удален после использования
|
||
stored_data_again = await get_oauth_state(state)
|
||
assert stored_data_again is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_login_redirect(self):
|
||
"""Тест OAuth login redirect"""
|
||
mock_request = AsyncMock()
|
||
mock_request.query_params = {
|
||
"provider": "google",
|
||
"state": "test_state",
|
||
"redirect_uri": "http://localhost:3000"
|
||
}
|
||
|
||
with patch('auth.oauth.store_oauth_state') as mock_store:
|
||
with patch('auth.oauth.generate_provider_url') as mock_generate:
|
||
mock_generate.return_value = "https://accounts.google.com/oauth/authorize?..."
|
||
|
||
response = await oauth_login_http(mock_request)
|
||
|
||
assert response.status_code == 307 # Redirect
|
||
mock_store.assert_called_once()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_callback_success(self):
|
||
"""Тест успешного OAuth callback"""
|
||
mock_request = AsyncMock()
|
||
mock_request.query_params = {
|
||
"code": "auth_code_123",
|
||
"state": "test_state"
|
||
}
|
||
|
||
with patch('auth.oauth.get_oauth_state') as mock_get_state:
|
||
mock_get_state.return_value = {
|
||
"redirect_uri": "http://localhost:3000"
|
||
}
|
||
|
||
with patch('auth.oauth.exchange_code_for_user_data') as mock_exchange:
|
||
mock_exchange.return_value = {
|
||
"id": "123",
|
||
"email": "test@example.com",
|
||
"name": "Test User"
|
||
}
|
||
|
||
with patch('auth.oauth._create_or_update_user') as mock_create_user:
|
||
mock_create_user.return_value = AsyncMock(id=123)
|
||
|
||
response = await oauth_callback_http(mock_request)
|
||
|
||
assert response.status_code == 307 # Redirect
|
||
assert "access_token=" in response.headers["location"]
|
||
```
|
||
|
||
## 🌐 E2E Tests
|
||
|
||
### Login Flow E2E Tests
|
||
|
||
```python
|
||
import pytest
|
||
from httpx import AsyncClient
|
||
from main import app
|
||
|
||
class TestLoginFlowE2E:
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_complete_login_flow(self):
|
||
"""Тест полного flow входа в систему"""
|
||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||
|
||
# 1. Регистрация пользователя
|
||
register_response = await client.post("/auth/register", json={
|
||
"email": "test@example.com",
|
||
"password": "TestPassword123!",
|
||
"name": "Test User"
|
||
})
|
||
|
||
assert register_response.status_code == 200
|
||
|
||
# 2. Вход в систему
|
||
login_response = await client.post("/auth/login", json={
|
||
"email": "test@example.com",
|
||
"password": "TestPassword123!"
|
||
})
|
||
|
||
assert login_response.status_code == 200
|
||
data = login_response.json()
|
||
assert data["success"] is True
|
||
assert "token" in data
|
||
|
||
# Проверяем установку cookie
|
||
cookies = login_response.cookies
|
||
assert "session_token" in cookies
|
||
|
||
# 3. Проверка защищенного endpoint с cookie
|
||
session_response = await client.get("/auth/session", cookies={
|
||
"session_token": cookies["session_token"]
|
||
})
|
||
|
||
assert session_response.status_code == 200
|
||
session_data = session_response.json()
|
||
assert session_data["user"]["email"] == "test@example.com"
|
||
|
||
# 4. Выход из системы
|
||
logout_response = await client.post("/auth/logout", cookies={
|
||
"session_token": cookies["session_token"]
|
||
})
|
||
|
||
assert logout_response.status_code == 200
|
||
|
||
# 5. Проверка, что сессия недоступна после выхода
|
||
invalid_session_response = await client.get("/auth/session", cookies={
|
||
"session_token": cookies["session_token"]
|
||
})
|
||
|
||
assert invalid_session_response.status_code == 401
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bearer_token_auth(self):
|
||
"""Тест аутентификации через Bearer token"""
|
||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||
|
||
# Вход в систему
|
||
login_response = await client.post("/auth/login", json={
|
||
"email": "test@example.com",
|
||
"password": "TestPassword123!"
|
||
})
|
||
|
||
token = login_response.json()["token"]
|
||
|
||
# Использование Bearer token
|
||
protected_response = await client.get("/auth/session", headers={
|
||
"Authorization": f"Bearer {token}"
|
||
})
|
||
|
||
assert protected_response.status_code == 200
|
||
data = protected_response.json()
|
||
assert data["user"]["email"] == "test@example.com"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_invalid_credentials(self):
|
||
"""Тест входа с неверными данными"""
|
||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||
|
||
response = await client.post("/auth/login", json={
|
||
"email": "test@example.com",
|
||
"password": "WrongPassword"
|
||
})
|
||
|
||
assert response.status_code == 401
|
||
data = response.json()
|
||
assert data["success"] is False
|
||
assert "error" in data
|
||
```
|
||
|
||
### OAuth E2E Tests
|
||
|
||
```python
|
||
import pytest
|
||
from unittest.mock import patch
|
||
from httpx import AsyncClient
|
||
from main import app
|
||
|
||
class TestOAuthFlowE2E:
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_oauth_google_flow(self):
|
||
"""Тест OAuth flow с Google"""
|
||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||
|
||
# 1. Инициация OAuth
|
||
oauth_response = await client.get(
|
||
"/auth/oauth/google",
|
||
params={
|
||
"state": "test_state_123",
|
||
"redirect_uri": "http://localhost:3000"
|
||
},
|
||
follow_redirects=False
|
||
)
|
||
|
||
assert oauth_response.status_code == 307
|
||
assert "accounts.google.com" in oauth_response.headers["location"]
|
||
|
||
# 2. Мокаем OAuth callback
|
||
with patch('auth.oauth.exchange_code_for_user_data') as mock_exchange:
|
||
mock_exchange.return_value = {
|
||
"id": "google_user_123",
|
||
"email": "user@gmail.com",
|
||
"name": "Google User"
|
||
}
|
||
|
||
callback_response = await client.get(
|
||
"/auth/oauth/google/callback",
|
||
params={
|
||
"code": "auth_code_123",
|
||
"state": "test_state_123"
|
||
},
|
||
follow_redirects=False
|
||
)
|
||
|
||
assert callback_response.status_code == 307
|
||
location = callback_response.headers["location"]
|
||
assert "access_token=" in location
|
||
|
||
# Извлекаем токен из redirect URL
|
||
import urllib.parse
|
||
parsed = urllib.parse.urlparse(location)
|
||
query_params = urllib.parse.parse_qs(parsed.query)
|
||
access_token = query_params["access_token"][0]
|
||
|
||
# 3. Проверяем, что токен работает
|
||
session_response = await client.get("/auth/session", headers={
|
||
"Authorization": f"Bearer {access_token}"
|
||
})
|
||
|
||
assert session_response.status_code == 200
|
||
data = session_response.json()
|
||
assert data["user"]["email"] == "user@gmail.com"
|
||
```
|
||
|
||
## 🧰 Test Fixtures
|
||
|
||
### Auth Fixtures
|
||
|
||
```python
|
||
import pytest
|
||
import asyncio
|
||
from auth.tokens.sessions import SessionTokenManager
|
||
from auth.tokens.oauth import OAuthTokenManager
|
||
|
||
@pytest.fixture
|
||
async def session_manager():
|
||
"""Фикстура SessionTokenManager"""
|
||
return SessionTokenManager()
|
||
|
||
@pytest.fixture
|
||
async def oauth_manager():
|
||
"""Фикстура OAuthTokenManager"""
|
||
return OAuthTokenManager()
|
||
|
||
@pytest.fixture
|
||
async def test_user_token(session_manager):
|
||
"""Фикстура для создания тестового токена"""
|
||
token = await session_manager.create_session(
|
||
user_id="test_user_123",
|
||
username="testuser"
|
||
)
|
||
|
||
yield token
|
||
|
||
# Cleanup
|
||
await session_manager.revoke_session_token(token)
|
||
|
||
@pytest.fixture
|
||
async def authenticated_client():
|
||
"""Фикстура для аутентифицированного клиента"""
|
||
from httpx import AsyncClient
|
||
from main import app
|
||
|
||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||
# Создаем пользователя и получаем токен
|
||
login_response = await client.post("/auth/login", json={
|
||
"email": "test@example.com",
|
||
"password": "TestPassword123!"
|
||
})
|
||
|
||
token = login_response.json()["token"]
|
||
|
||
# Настраиваем клиент с токеном
|
||
client.headers.update({"Authorization": f"Bearer {token}"})
|
||
|
||
yield client
|
||
|
||
@pytest.fixture
|
||
async def oauth_tokens(oauth_manager):
|
||
"""Фикстура для OAuth токенов"""
|
||
await oauth_manager.store_oauth_tokens(
|
||
user_id="test_user_123",
|
||
provider="google",
|
||
access_token="test_access_token",
|
||
refresh_token="test_refresh_token",
|
||
expires_in=3600
|
||
)
|
||
|
||
yield {
|
||
"user_id": "test_user_123",
|
||
"provider": "google",
|
||
"access_token": "test_access_token",
|
||
"refresh_token": "test_refresh_token"
|
||
}
|
||
|
||
# Cleanup
|
||
await oauth_manager.revoke_oauth_tokens("test_user_123", "google")
|
||
```
|
||
|
||
### Redis Fixtures
|
||
|
||
```python
|
||
import pytest
|
||
from storage.redis import redis
|
||
|
||
@pytest.fixture(scope="session")
|
||
async def redis_client():
|
||
"""Фикстура Redis клиента"""
|
||
yield redis
|
||
|
||
# Cleanup после всех тестов
|
||
await redis.flushdb()
|
||
|
||
@pytest.fixture
|
||
async def clean_redis():
|
||
"""Фикстура для очистки Redis перед тестом"""
|
||
# Очищаем тестовые ключи
|
||
test_keys = await redis.keys("test:*")
|
||
if test_keys:
|
||
await redis.delete(*test_keys)
|
||
|
||
yield
|
||
|
||
# Очищаем после теста
|
||
test_keys = await redis.keys("test:*")
|
||
if test_keys:
|
||
await redis.delete(*test_keys)
|
||
```
|
||
|
||
## 📊 Test Configuration
|
||
|
||
### pytest.ini
|
||
|
||
```ini
|
||
[tool:pytest]
|
||
asyncio_mode = auto
|
||
testpaths = tests
|
||
python_files = test_*.py
|
||
python_classes = Test*
|
||
python_functions = test_*
|
||
addopts =
|
||
-v
|
||
--tb=short
|
||
--strict-markers
|
||
--disable-warnings
|
||
--cov=auth
|
||
--cov-report=html
|
||
--cov-report=term-missing
|
||
--cov-fail-under=80
|
||
|
||
markers =
|
||
unit: Unit tests
|
||
integration: Integration tests
|
||
e2e: End-to-end tests
|
||
slow: Slow tests
|
||
redis: Tests requiring Redis
|
||
oauth: OAuth related tests
|
||
```
|
||
|
||
### conftest.py
|
||
|
||
```python
|
||
import pytest
|
||
import asyncio
|
||
from unittest.mock import AsyncMock
|
||
from httpx import AsyncClient
|
||
from main import app
|
||
|
||
# Настройка asyncio для тестов
|
||
@pytest.fixture(scope="session")
|
||
def event_loop():
|
||
"""Создает event loop для всей сессии тестов"""
|
||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||
yield loop
|
||
loop.close()
|
||
|
||
# Мок Redis для unit тестов
|
||
@pytest.fixture
|
||
def mock_redis():
|
||
"""Мок Redis клиента"""
|
||
mock = AsyncMock()
|
||
mock.ping.return_value = True
|
||
mock.get.return_value = None
|
||
mock.set.return_value = True
|
||
mock.delete.return_value = 1
|
||
mock.exists.return_value = False
|
||
mock.ttl.return_value = -1
|
||
mock.hset.return_value = 1
|
||
mock.hgetall.return_value = {}
|
||
mock.sadd.return_value = 1
|
||
mock.smembers.return_value = set()
|
||
mock.srem.return_value = 1
|
||
mock.expire.return_value = True
|
||
mock.setex.return_value = True
|
||
return mock
|
||
|
||
# Test client
|
||
@pytest.fixture
|
||
async def test_client():
|
||
"""Тестовый HTTP клиент"""
|
||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||
yield client
|
||
```
|
||
|
||
## 🚀 Running Tests
|
||
|
||
### Команды запуска
|
||
|
||
```bash
|
||
# Все тесты
|
||
pytest
|
||
|
||
# Unit тесты
|
||
pytest tests/auth/unit/ -m unit
|
||
|
||
# Integration тесты
|
||
pytest tests/auth/integration/ -m integration
|
||
|
||
# E2E тесты
|
||
pytest tests/auth/e2e/ -m e2e
|
||
|
||
# Тесты с покрытием
|
||
pytest --cov=auth --cov-report=html
|
||
|
||
# Параллельный запуск
|
||
pytest -n auto
|
||
|
||
# Только быстрые тесты
|
||
pytest -m "not slow"
|
||
|
||
# Конкретный тест
|
||
pytest tests/auth/unit/test_session_manager.py::TestSessionTokenManager::test_create_session
|
||
```
|
||
|
||
### CI/CD Integration
|
||
|
||
```yaml
|
||
# .github/workflows/tests.yml
|
||
name: Tests
|
||
|
||
on: [push, pull_request]
|
||
|
||
jobs:
|
||
test:
|
||
runs-on: ubuntu-latest
|
||
|
||
services:
|
||
redis:
|
||
image: redis:6.2
|
||
ports:
|
||
- 6379:6379
|
||
options: >-
|
||
--health-cmd "redis-cli ping"
|
||
--health-interval 10s
|
||
--health-timeout 5s
|
||
--health-retries 5
|
||
|
||
steps:
|
||
- uses: actions/checkout@v3
|
||
|
||
- name: Set up Python
|
||
uses: actions/setup-python@v4
|
||
with:
|
||
python-version: '3.12'
|
||
|
||
- name: Install dependencies
|
||
run: |
|
||
pip install -r requirements.dev.txt
|
||
|
||
- name: Run unit tests
|
||
run: |
|
||
pytest tests/auth/unit/ -m unit --cov=auth
|
||
|
||
- name: Run integration tests
|
||
run: |
|
||
pytest tests/auth/integration/ -m integration
|
||
env:
|
||
REDIS_URL: redis://localhost:6379/0
|
||
|
||
- name: Run E2E tests
|
||
run: |
|
||
pytest tests/auth/e2e/ -m e2e
|
||
env:
|
||
REDIS_URL: redis://localhost:6379/0
|
||
JWT_SECRET_KEY: test_secret_key_for_ci
|
||
|
||
- name: Upload coverage
|
||
uses: codecov/codecov-action@v3
|
||
```
|
||
|
||
## 📈 Test Metrics
|
||
|
||
### Coverage Goals
|
||
- **Unit Tests**: ≥ 90% coverage
|
||
- **Integration Tests**: ≥ 80% coverage
|
||
- **E2E Tests**: Critical paths covered
|
||
- **Overall**: ≥ 85% coverage
|
||
|
||
### Performance Benchmarks
|
||
- **Unit Tests**: < 100ms per test
|
||
- **Integration Tests**: < 1s per test
|
||
- **E2E Tests**: < 10s per test
|
||
- **Total Test Suite**: < 5 minutes
|
||
|
||
### Quality Metrics
|
||
- **Test Reliability**: ≥ 99% pass rate
|
||
- **Flaky Tests**: < 1% of total tests
|
||
- **Test Maintenance**: Regular updates with code changes
|