Кратко и по делу: системный флоу для поиска/фикса и набор практик; затем разбор примера с конкретными альтернативами.
1) Системный подход (шаги)
- Сбор и статистика: прогонять подозрительный тест $N$ раз (например $N=$ $100$) в CI-подобной среде, чтобы оценить вероятность флейка. Формула шанса хотя бы одного падения за $N$ прогонов при истинной вероятности падения $p$: $$P_{\text{встреча}}=1-(1-p{)}^N.$$ - Репродукция в среде CI: запускать в контейнере/машине с теми же переменными окружения, ресурсами и параллелизмом.
- Изоляция: отключать соседние тесты, использовать tmpdirs/уникальные порты/файловые блокировки, запускать тест в одиночном процессе.
- Диагностика: добавлять подробный лог, сохранять дампы/трейсы, стресс-прогоны (много итераций), запуски с увеличенным уровнем параллелизма чтобы выявить гонки.
- Исправление и предотвращение: убрать ненадёжные конструкции (fixed sleeps, глоб. состояния, непредсказуемый рандом), ввести синхронизацию, deterministic-seed и mock/заглушки для внешних зависимостей.
- Практический шаг: пометить тест как flaky/skip + баг-репорт, если не удаётся быстро устранить.
2) Частые причины и таргетированные меры
- Зависимость от времени: заменить жесткие sleeps на ожидание события/poll с таймаутом; использовать фиксацию времени (freezegun) для логики, зависящей от clock.
- Рандом/стороны: фиксировать seed: random.seed($42$), numpy.random.seed($42$), PYTHONHASHSEED=$42$.
- Параллелизм/гонки: использовать локальные ресурсы, mutex/lock, fixtures с scope='function'; запускать под инструментами для гонок (если C/C++ расширения — ThreadSanitizer).
- Нестабильные внешние сервисы: мокать HTTP/DB (responses, vcrpy, fakeredis, testcontainers), убрать сетевой флап.
3) Инструменты и приёмы
- pytest-rerunfailures / pytest-flaky: временное решение --перезапуск теста (--reruns $3$), но не лечит причину.
- pytest-xdist: репродукция проблем параллелизма (pytest -n $4$).
- freezegun: фиксация времени.
- responses / vcrpy / fakeredis / testcontainers: стабилизация внешних зависимостей.
- strace / faulthandler / логирование / сохранение stdout/stderr для CI.
- Контейнеризация (Docker) + воспроизводимые образа CI.
- Запуск «стресс» циклов: `for i in $(seq 1 $100$); do pytest -k test_name ...; done` для нахождения редких провалов.
- Для гонок: запуск с максимальным параллелизмом, многократные прогоны и санитайзеры.
4) Разбор примера
Исходный (хрупкий):
def test_job():
start_job()
time.sleep(0.1)
assert job_finished()
Проблемы: жесткое ожидание $0.1$ — ненадёжно (машины CI медленнее), нет проверки таймаута/условия.
Устойчивые альтернативы (пояснения внутри):
- Polling с таймаутом (обычно лучше, чем фикс sleep):
def test_job():
start_job()
deadline = time.time() + $1.0$ while time.time() < deadline:
if job_finished():
break
time.sleep($0.01$)
assert job_finished()
Плюсы: адаптируется к реальной задержке, есть явный таймаут.
- Синхронизация через Event/Condition (если код может сигнализировать):
from threading import Event
finished = Event()
# в коде jobs: finished.set() когда готово
def test_job():
start_job()
assert finished.wait(timeout=$1.0$) # вернёт True если событие произошло
Плюсы: детерминировано, нет угадывания таймаута, быстро.
- Использование future/await (async):
async def test_job(async_client):
fut = start_job_async()
await asyncio.wait_for(fut, timeout=$1.0$)
assert fut.done() and fut.result().success
- Мокирование/инжекция зависимости: если job — внешний сервис, заменить его заглушкой, вернуть готовый результат и проверить обработку.
- Проверка прогресса вместо полного ожидания: проверять конкретные побочные эффекты (запись файла, запись в БД), а не только общую job_finished().
5) Стратегия стабилизации процесса разработки
- Запуск тестов локально в CI-like контейнере перед PR.
- Обязательные прогоны «flake hunting»: подозрительный тест гонять $N$ раз в CI-отдельном job.
- CI: собирать статистику флейков по тестам, автоматически помечать повторяющие падения.
- Код-ревью: избегать time.sleep, глобальных стейт-флагов, неблокирующего поведения без синхронизации.
- Документировать и фиксировать seed/окружение для тестов.
6) Быстрая чек-лист для каждого флейка
- Можно ли воспроизвести локально в докер/CI-образе? Если нет — собрать логи.
- Есть ли рандом? Зафиксируйте seed.
- Есть ли sleep? Замените на wait/poll/event.
- Влияет ли параллельность? Запустите xdist/однопоточно.
- Мокируете ли внешние сервисы? Если нет — замокируйте.
- Добавьте больше логов и повторных запусков $N=$ $100$ для статистики.
Если нужно, могу предложить конкретный паттерн ожидания/мока для вашего кода (покажите start_job/job_finished), или подготовить CI-пайплайн для автоматического поиска флейков.