Pytest: Борьба с Race Conditions в Async-коде

8 / 989%

Устойчивость к сбоям: падение Redis и PostgreSQL

40 минут

Проектный якорь

В проде Redis временно недоступен, Postgres выбивается в предел по коннектам, тесты на idempotency падают раз в несколько прогонов.
Наша цель — закрепить инфраструктуру фикстур и научиться детектить/чинить эти сбои.

Воспроизводим случайный фейл

PYTHONHASHSEED=1 pytest tests/integration -q --random-order --reruns 2 --reruns-delay 0.5 --maxfail=1

--random-order вскрывает зависимость от кеша между тестами.
--reruns используем только для диагностики, не для маскировки.

Интеграционный тест: падение кеша

@pytest.mark.integration
def test_graceful_degradation_when_cache_down(cache_down, api_client):
    resp = api_client.get("/tasks/u1")
    assert resp.status_code == 200
    assert resp.json()["id"] == "u1"
    assert "cache.unavailable" in api_client.captured_metrics

Фикстура cache_down убивает Redis-подключение, проверяем обратное давление и метрики.

Гонки idempotency

@pytest.mark.asyncio
async def test_idempotency_on_retry(task_service, make_task_payload):
    payload = make_task_payload(idempotency_key="k1")
    r1, r2 = await asyncio.gather(
        task_service.create_task(payload),
        task_service.create_task(payload),
    )
    assert {r1.status, r2.status} == {"accepted"}
    assert await task_service.ledger.count("k1") == 1

Фикс

Лок вокруг кеша + транзакция на вставку.
Проверка idempotency ключа в БД, не только в Redis.

Исчерпание пула соединений

@pytest.mark.integration
def test_database_connection_pool_exhaustion(db_pool):
    conns = [db_pool.acquire() for _ in range(db_pool.max_size)]
    with pytest.raises(TimeoutError):
        db_pool.acquire(timeout=0.1)
    for conn in conns:
        conn.release()

Фикс

Лимит параллельных клиентов в тестах (pytest-xdist -n 4 с max_size ≥ воркеров).
Автосоздание/освобождение коннектов в фикстурах, чтобы не оставлять висящие handle.

Карантин flaky-тестов

import os
import pytest
 
def pytest_runtest_makereport(item, call):
    if call.when == "call" and call.excinfo is not None:
        item.config.cache.set(f"flaky/{item.nodeid}", {
            "seed": os.getenv("PYTHONHASHSEED"),
            "markers": list(item.iter_markers()),
        })

Сохраняйте seed/маркеры в артефакты CI, чтобы воспроизводить локально.
Заводите «карантинный» прогон -m flaky --runslow отдельно от основного.

Чеклист урока

Интеграционные тесты как контракты между БД, кешом и API.
Воспроизводимость flaky: random order, фиксированный seed, сбор диагностики.
Фиксы: блокировки, транзакции, явная деградация при падении кеша.
Карантин + тикеты: не оставлять flaky без хозяина.

Back to course curriculum

Устойчивость к сбоям: падение Redis и PostgreSQL — Pytest: Борьба с Race Conditions в Async-коде — Potapov.me