Стратег Тестирования

🧪 Стратегия Тестирования

📋 План Тестового Покрытия

Цель покрытия: 95%

Типы тестов: Unit, Integration, Performance, Security

🔬 Модульные Тесты (Unit Tests)

import pytest
import unittest.mock as mock
from datetime import datetime, timedelta
from typing import List
import asyncio

class TestUserService:
    """Комплексные тесты для UserService."""
    
    @pytest.fixture
    def user_service(self):
        """Создание сервиса с mock зависимостями."""
        mock_db = mock.Mock()
        mock_email = mock.Mock()
        return UserService(database=mock_db, email_service=mock_email)
    
    @pytest.fixture
    def sample_user(self):
        """Тестовый пользователь."""
        return User(
            id=1,
            email="test@example.com",
            name="Тестовый Пользователь",
            created_at=datetime.now()
        )
    
    def test_create_user_success(self, user_service, sample_user):
        """Тест успешного создания пользователя."""
        # Arrange
        user_service.database.save.return_value = sample_user
        user_service.email_service.send_welcome.return_value = True
        
        # Act
        result = user_service.create_user(
            email="test@example.com",
            name="Тестовый Пользователь"
        )
        
        # Assert
        assert result.email == "test@example.com"
        assert result.name == "Тестовый Пользователь"
        user_service.database.save.assert_called_once()
        user_service.email_service.send_welcome.assert_called_once_with(result)
    
    def test_create_user_invalid_email(self, user_service):
        """Тест создания пользователя с некорректным email."""
        with pytest.raises(ValueError, match="Некорректный email"):
            user_service.create_user(
                email="invalid-email",
                name="Тест"
            )
    
    @pytest.mark.parametrize("email,name,expected_error", [
        ("", "Имя", "Email не может быть пустым"),
        ("test@test.com", "", "Имя не может быть пустым"),
        ("test@test.com", "a" * 101, "Имя слишком длинное"),
    ])
    def test_create_user_validation_errors(self, user_service, email, name, expected_error):
        """Параметризованный тест валидации данных."""
        with pytest.raises(ValueError, match=expected_error):
            user_service.create_user(email=email, name=name)
    
    @mock.patch("user_service.datetime")
    def test_create_user_with_mocked_time(self, mock_datetime, user_service):
        """Тест с фиксированным временем."""
        # Arrange
        fixed_time = datetime(2023, 1, 1, 12, 0, 0)
        mock_datetime.now.return_value = fixed_time
        
        # Act
        user = user_service.create_user("test@test.com", "Тест")
        
        # Assert
        assert user.created_at == fixed_time

class TestUserRepository:
    """Тесты репозитория пользователей."""
    
    @pytest.fixture
    async def repository(self):
        """Асинхронный репозиторий для тестирования."""
        return InMemoryUserRepository()
    
    @pytest.mark.asyncio
    async def test_save_and_find_user(self, repository, sample_user):
        """Тест сохранения и поиска пользователя."""
        # Save user
        saved_user = await repository.save(sample_user)
        assert saved_user.id is not None
        
        # Find user
        found_user = await repository.find_by_id(saved_user.id)
        assert found_user is not None
        assert found_user.email == sample_user.email
    
    @pytest.mark.asyncio
    async def test_find_nonexistent_user(self, repository):
        """Тест поиска несуществующего пользователя."""
        user = await repository.find_by_id(999)
        assert user is None
    
    @pytest.mark.asyncio
    async def test_update_user(self, repository, sample_user):
        """Тест обновления пользователя."""
        # Save initial user
        saved_user = await repository.save(sample_user)
        
        # Update user
        saved_user.name = "Обновленное Имя"
        updated_user = await repository.save(saved_user)
        
        # Verify update
        assert updated_user.name == "Обновленное Имя"
        
        # Verify in database
        found_user = await repository.find_by_id(saved_user.id)
        assert found_user.name == "Обновленное Имя"

🔗 Интеграционные Тесты

import pytest
import aiohttp
import asyncio
from sqlalchemy import create_engine
from testcontainers.postgres import PostgreSqlContainer

class TestUserAPI:
    """Интеграционные тесты API пользователей."""
    
    @pytest.fixture(scope="class")
    def postgres_container(self):
        """Контейнер PostgreSQL для тестирования."""
        with PostgreSqlContainer("postgres:13") as postgres:
            yield postgres
    
    @pytest.fixture(scope="class")
    async def test_app(self, postgres_container):
        """Тестовое приложение с реальной БД."""
        # Настройка подключения к тестовой БД
        db_url = postgres_container.get_connection_url()
        
        # Создание тестового приложения
        app = create_app(database_url=db_url)
        
        # Создание таблиц
        await app.setup_database()
        
        return app
    
    @pytest.fixture
    async def client(self, test_app):
        """HTTP клиент для тестирования."""
        async with aiohttp.ClientSession() as session:
            yield session
    
    @pytest.mark.asyncio
    async def test_user_crud_workflow(self, client, test_app):
        """Полный цикл CRUD операций через API."""
        base_url = f"http://localhost:{{test_app.port}}"
        
        # CREATE: Создание пользователя
        create_data = {
            "email": "integration@test.com",
            "name": "Интеграционный Тест"
        }
        
        async with client.post(f"{{base_url}}/api/users", json=create_data) as resp:
            assert resp.status == 201
            user_data = await resp.json()
            user_id = user_data["id"]
            assert user_data["email"] == create_data["email"]
        
        # READ: Получение пользователя
        async with client.get(f"{{base_url}}/api/users/{{user_id}}") as resp:
            assert resp.status == 200
            user_data = await resp.json()
            assert user_data["email"] == create_data["email"]
        
        # UPDATE: Обновление пользователя
        update_data = {"name": "Обновленное Имя"}
        async with client.put(f"{{base_url}}/api/users/{{user_id}}", json=update_data) as resp:
            assert resp.status == 200
            updated_data = await resp.json()
            assert updated_data["name"] == update_data["name"]
        
        # DELETE: Удаление пользователя
        async with client.delete(f"{{base_url}}/api/users/{{user_id}}") as resp:
            assert resp.status == 204
        
        # Проверка удаления
        async with client.get(f"{{base_url}}/api/users/{{user_id}}") as resp:
            assert resp.status == 404
    
    @pytest.mark.asyncio
    async def test_user_list_pagination(self, client, test_app):
        """Тест пагинации списка пользователей."""
        base_url = f"http://localhost:{{test_app.port}}"
        
        # Создание нескольких пользователей
        for i in range(25):
            user_data = {
                "email": f"user{{i}}@test.com",
                "name": f"Пользователь {{i}}"
            }
            async with client.post(f"{{base_url}}/api/users", json=user_data) as resp:
                assert resp.status == 201
        
        # Тест первой страницы
        async with client.get(f"{{base_url}}/api/users?page=1&limit=10") as resp:
            assert resp.status == 200
            data = await resp.json()
            assert len(data["users"]) == 10
            assert data["pagination"]["page"] == 1
            assert data["pagination"]["has_next"] is True
        
        # Тест последней страницы
        async with client.get(f"{{base_url}}/api/users?page=3&limit=10") as resp:
            assert resp.status == 200
            data = await resp.json()
            assert len(data["users"]) == 5  # Остальные 5 пользователей
            assert data["pagination"]["has_next"] is False

⚡ Тесты Производительности

import pytest
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor
import statistics

class TestPerformance:
    """Тесты производительности системы."""
    
    @pytest.mark.performance
    @pytest.mark.asyncio
    async def test_user_creation_performance(self, user_service):
        """Тест производительности создания пользователей."""
        # Подготовка данных
        user_count = 1000
        users_data = [
            {"email": f"perf_test_{{i}}@test.com", "name": f"User {{i}}"}
            for i in range(user_count)
        ]
        
        # Измерение времени создания
        start_time = time.perf_counter()
        
        tasks = [
            user_service.create_user(data["email"], data["name"])
            for data in users_data
        ]
        
        results = await asyncio.gather(*tasks)
        
        end_time = time.perf_counter()
        execution_time = end_time - start_time
        
        # Проверки производительности
        assert len(results) == user_count
        assert execution_time < 5.0  # Должно выполняться меньше чем за 5 секунд
        
        avg_time_per_user = execution_time / user_count
        assert avg_time_per_user < 0.01  # Меньше 10мс на пользователя
        
        print(f"Создано {{user_count}} пользователей за {{execution_time:.2f}}с")
        print(f"Среднее время на пользователя: {{avg_time_per_user*1000:.2f}}мс")
    
    @pytest.mark.performance
    def test_concurrent_user_access(self, user_service):
        """Тест конкурентного доступа к пользователям."""
        # Создание тестового пользователя
        test_user = user_service.create_user("concurrent@test.com", "Test User")
        
        def access_user():
            """Функция для конкурентного доступа."""
            start_time = time.perf_counter()
            user = user_service.get_user_by_id(test_user.id)
            end_time = time.perf_counter()
            return end_time - start_time
        
        # Выполнение конкурентных запросов
        concurrent_requests = 100
        
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(access_user)
                for _ in range(concurrent_requests)
            ]
            
            response_times = [future.result() for future in futures]
        
        # Анализ результатов
        avg_response_time = statistics.mean(response_times)
        max_response_time = max(response_times)
        p95_response_time = statistics.quantiles(response_times, n=20)[18]  # 95-й перцентиль
        
        # Проверки производительности
        assert avg_response_time < 0.1  # Средний ответ меньше 100мс
        assert max_response_time < 0.5  # Максимальный ответ меньше 500мс
        assert p95_response_time < 0.2  # 95% запросов меньше 200мс
        
        print(f"Конкурентные запросы ({{concurrent_requests}}):")
        print(f"  Среднее время: {{avg_response_time*1000:.2f}}мс")
        print(f"  Максимальное время: {{max_response_time*1000:.2f}}мс")
        print(f"  95-й перцентиль: {{p95_response_time*1000:.2f}}мс")

🛡️ Тесты Безопасности

import pytest
import jwt
from unittest.mock import patch

class TestSecurity:
    """Тесты безопасности системы."""
    
    def test_sql_injection_protection(self, user_service):
        """Тест защиты от SQL инъекций."""
        malicious_inputs = [
            "'; DROP TABLE users; --",
            "admin@test.com'; UPDATE users SET role = 'admin'; --",
            "1 OR 1=1",
            ""
        ]
        
        for malicious_input in malicious_inputs:
            with pytest.raises((ValueError, SecurityError)):
                user_service.create_user(malicious_input, "Test User")
    
    def test_authentication_required(self, client):
        """Тест обязательной аутентификации."""
        protected_endpoints = [
            "/api/users/profile",
            "/api/users/settings",
            "/api/admin/users"
        ]
        
        for endpoint in protected_endpoints:
            with client.get(endpoint) as resp:
                assert resp.status == 401  # Unauthorized
    
    def test_jwt_token_validation(self, auth_service):
        """Тест валидации JWT токенов."""
        # Валидный токен
        valid_token = auth_service.create_token(user_id=1)
        user_id = auth_service.validate_token(valid_token)
        assert user_id == 1
        
        # Недействительный токен
        invalid_tokens = [
            "invalid.token.here",
            "",
            "expired.token.signature",
            None
        ]
        
        for invalid_token in invalid_tokens:
            with pytest.raises(AuthenticationError):
                auth_service.validate_token(invalid_token)

📊 Отчет о Покрытии

# pytest.ini
[tool:pytest]
addopts = 
    --cov=src
    --cov-report=html
    --cov-report=term-missing
    --cov-fail-under=90
    --strict-markers
    --disable-warnings

markers =
    unit: Unit tests
    integration: Integration tests
    performance: Performance tests
    security: Security tests

# Команды для запуска тестов
# Все тесты: pytest
# Только unit тесты: pytest -m unit
# С покрытием: pytest --cov=src --cov-report=html
# Производительность: pytest -m performance

🎯 Результат

• Unit тесты: 45 тестов, покрытие 95%
• Integration тесты: 12 тестов, критичные сценарии
• Performance тесты: 8 тестов, SLA соблюдены
• Security тесты: 6 тестов, уязвимости не найдены