core/docs/auth/testing.md

🧪 Тестирование системы аутентификации

🎯 Обзор

Комплексная стратегия тестирования системы аутентификации с 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

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

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

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

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

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

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

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

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

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

[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

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

Команды запуска

# Все тесты
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

# .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