Какие были трудности при тестировании приложения?

Трудности при тестировании приложения

Тестирование — один из самых сложных аспектов разработки. Рассмотрю реальные проблемы, с которыми сталкиваюсь на проектах.

1. Асинхронный код и Race Conditions

Проблема: Асинхронный код сложно тестировать, т.к. порядок выполнения недетерминирован.

# Плохой тест — может падать нерегулярно
@pytest.mark.asyncio
async def test_concurrent_updates():
    user = await create_user()
    
    await asyncio.gather(
        update_balance(user.id, 100),
        update_balance(user.id, 50),
    )
    # Race condition: какой порядок выполнения?
    assert await get_balance(user.id) == 150  # Может быть 50 или 100

Решение:

# Использовать SELECT FOR UPDATE для синхронизации
@pytest.mark.asyncio
async def test_concurrent_updates():
    user = await create_user()
    
    # В коде используем транзакции
    async with db.transaction():
        result = await db.execute(
            "UPDATE users SET balance = balance + $1 "
            "WHERE id = $2 FOR UPDATE",
            100, user.id
        )
    
    balance = await get_balance(user.id)
    assert balance == 100

2. Мокирование сложных зависимостей

Проблема: Много зависимостей — BDD, cache, queue, files.

# Сложный код с много зависимостями
async def process_payment(user_id: int, amount: float):
    # 1. Проверить баланс (БД)
    user = await db.get_user(user_id)
    if user.balance < amount:
        raise InsufficientFunds()
    
    # 2. Вызвать внешний API
    payment_id = await stripe_api.charge(amount)
    
    # 3. Обновить БД
    await db.update_payment(payment_id, user_id)
    
    # 4. Отправить письмо
    await email_service.send_receipt(user.email)
    
    # 5. Кэшировать результат
    await redis.set(f"payment:{payment_id}", payment_id)

Решение: Использовать dependency injection

from unittest.mock import AsyncMock
import pytest
from fastapi import Depends
from typing import Callable

class FakePaymentService:
    async def charge(self, amount: float):
        return "test_payment_id_123"

class FakeEmailService:
    async def send_receipt(self, email: str):
        pass

@pytest.fixture
def mock_stripe():
    return AsyncMock(return_value="payment_123")

@pytest.fixture
def mock_email():
    return AsyncMock()

@pytest.mark.asyncio
async def test_process_payment(mock_stripe, mock_email):
    result = await process_payment(
        user_id=1,
        amount=100,
        stripe=mock_stripe,
        email=mock_email
    )
    
    mock_stripe.charge.assert_called_once_with(100)
    mock_email.send_receipt.assert_called_once()
    assert result == "payment_123"

3. Тестирование БД

Проблема: Тесты медленные, грязнят БД, оставляют данные.

# Плохо: медленно и грязно
@pytest.mark.asyncio
async def test_user_creation():
    user = await db.create_user(name="John")
    assert user.name == "John"
    # Данные остаются в БД!

# Хорошо: используем транзакции
@pytest.mark.asyncio
async def test_user_creation(db_transaction):
    async with db_transaction:  # Откатится после теста
        user = await db.create_user(name="John")
        assert user.name == "John"
        # Откат транзакции после теста

Использовать pytest-asyncio с --reuse-db:

pytest --reuse-db  # Переиспользовать БД между тестами

4. Тестирование сетевых запросов

Проблема: HTTP запросы медленные, ненадёжные, дорогие.

# Плохо: реальный запрос
async def test_get_user_from_external_api():
    user = await fetch_user(id=123)  # Настоящий HTTP запрос!
    assert user.name == "John"

Решение: Использовать VCR для записи

import pytest
from pytest_httpserver import HTTPServer
from vcr import VCR

vcr = VCR(cassette_library_dir="tests/cassettes")

@pytest.mark.asyncio
@vcr.use_cassette("test_external_api.yaml")
async def test_get_user_from_external_api():
    # Первый запуск: записывает реальный ответ в cassette
    # Последующие: использует кэшированный ответ
    user = await fetch_user(id=123)
    assert user.name == "John"

5. Тестирование машинного состояния (FSM)

Проблема: Сложные переходы между состояниями.

# Плохо: неполное покрытие
def test_order_workflow():
    order = Order(status="pending")
    order.confirm()
    assert order.status == "confirmed"

# Хорошо: проверить все переходы
import pytest

@pytest.mark.parametrize("from_status,action,to_status,should_succeed", [
    ("pending", "confirm", "confirmed", True),
    ("confirmed", "confirm", "confirmed", False),  # Duplicate confirm
    ("pending", "ship", None, False),  # Invalid transition
    ("confirmed", "ship", "shipped", True),
    ("shipped", "cancel", None, False),  # Can't cancel shipped
])
def test_order_transitions(from_status, action, to_status, should_succeed):
    order = Order(status=from_status)
    
    if should_succeed:
        getattr(order, action)()
        assert order.status == to_status
    else:
        with pytest.raises(InvalidStateTransition):
            getattr(order, action)()

6. Тестирование больших объёмов данных

Проблема: Тесты медленные на больших датасетах.

# Плохо: создаёт 10000 записей
async def test_bulk_import():
    users = [await create_user(f"user_{i}") for i in range(10000)]
    result = await import_users(users)
    assert result.count == 10000

# Хорошо: используем фабрики
from factory import Factory
import factory

class UserFactory(Factory):
    class Meta:
        model = User
    
    name = factory.Sequence(lambda n: f"user_{n}")
    email = factory.Sequence(lambda n: f"user_{n}@example.com")

async def test_bulk_import():
    # Создаёт только нужное количество
    users = UserFactory.create_batch(100)
    result = await import_users(users)
    assert result.count == 100

7. Flaky тесты

Проблема: Тесты падают нерегулярно (~5% от запусков).

# Flaky: зависит от времени
def test_cache_expiry():
    cache.set("key", "value", ttl=1)
    # Если тест медленный, может истечь время
    assert cache.get("key") == "value"

# Фиксим: явная управление временем
from freezegun import freeze_time

@freeze_time("2024-01-01 12:00:00")
def test_cache_expiry():
    cache.set("key", "value", ttl=1)
    with freeze_time("2024-01-01 12:00:05"):  # Плюс 5 секунд
        assert cache.get("key") is None

8. Покрытие кода (Coverage)

Проблема: 100% coverage не значит всё протестировано.

# 100% coverage, но не всё протестировано
def process(data):
    if data.get("type") == "error":  # Coverage: checked
        raise ValueError()
    if data.get("count") > 100:  # Coverage: checked
        return "many"
    return "few"

# Тест покрывает только happy path
def test_process():
    assert process({"count": 10}) == "few"  # 100% покрытие?

Решение: Тестировать все branches

import pytest

@pytest.mark.parametrize("data,expected", [
    ({"type": "error"}, pytest.raises(ValueError)),
    ({"count": 150}, "many"),
    ({"count": 50}, "few"),
])
def test_process(data, expected):
    if isinstance(expected, type):
        with expected:
            process(data)
    else:
        assert process(data) == expected

Практические рекомендации

• Используйте fixtures для переиспользуемой логики
• Пишите unit тесты быстро, integration тесты медленнее
• Держите 90%+ покрытие с акцентом на critical paths
• Используйте VCR для HTTP, freezegun для времени
• Тестируйте edge cases и error paths
• Избегайте flaky тестов через freezegun и явное управление состоянием
• Profilerе медленные тесты и оптимизируйте