Как воспроизводить плавающие ошибки: ревью и рефакторинг

Тест, который падает один раз из двадцати, обычно скрывает не «случайность», а потерянное условие: порядок выполнения потоков, общий изменяемый объект, задержку внешнего сервиса или слишком короткий таймаут. Ниже — практический алгоритм для Python 3.12 и pytest 8.x: как собрать контекст сбоя, превратить плавающий тест в повторяемый сценарий и провести ревью так, чтобы рефакторинг не замаскировал проблему.

Сначала зафиксируйте, что именно плавает

Фраза «тест иногда падает» недостаточна для ревью. До изменения кода запишите четыре параметра:

• точную команду запуска и имя теста;
• версию Python, pytest, операционной системы и базы данных;
• входные данные, порядок тестов и параллельность;
• полный текст ошибки, логи и длительность успешного и неуспешного запуска.

Запустите проблемный тест отдельно с подробным выводом: python -m pytest path/to/test_file.py::test_name -vv -x --durations=10. Опция -x останавливает прогон на первом сбое, а --durations=10 показывает самые медленные тесты. Это не исправляет проблему, но убирает шум от последующих падений и помогает увидеть связь с таймингом.

После этого повторите тест серией отдельных запусков. Количество повторов и долю сбоев зафиксируйте в отчёте, но не принимайте «прошёл 100 раз» за доказательство исправления. Ошибка, зависящая от редкого порядка событий, может не проявиться в коротком прогоне.

Цель воспроизведения — не заставить баг упасть любой ценой, а найти условие, при котором его результат перестаёт быть случайным.

Что искать на ревью

Фиксированную задержку вместо события

Самый узнаваемый шаблон — запуск фоновой работы, time.sleep(0.1), а затем проверка результата. Такой тест зависит от скорости машины: на ноутбуке он проходит, в CI под нагрузкой просыпается раньше worker-потока.

import time
from threading import Thread


def test_background_job():
    result = []

    def worker():
        time.sleep(0.2)
        result.append("done")

    thread = Thread(target=worker)
    thread.start()
    time.sleep(0.1)

    assert result == ["done"]

В ревью замечание должно быть конкретным: «Тест проверяет состояние после фиксированной задержки, но не ждёт завершения работы. При задержке worker больше 100 мс будет ложный сбой». Предложение «увеличить sleep до секунды» только снижает частоту ошибки и добавляет время всему прогону.

Ожидаемый результат должен приходить через канал

Для фонового worker-а передайте результат через очередь. Она одновременно сообщает о завершении и переносит исключение из потока в тест. Пример ниже самодостаточен для Python 3.12.

from queue import Queue
from threading import Thread
from typing import Callable


def run_in_background(
    job: Callable[[], str],
    results: Queue[tuple[bool, object]],
) -> None:
    try:
        results.put((True, job()))
    except Exception as error:
        results.put((False, error))


def test_background_job():
    results: Queue[tuple[bool, object]] = Queue()

    worker = Thread(
        target=run_in_background,
        args=(lambda: "done", results),
    )
    worker.start()

    succeeded, value = results.get(timeout=2)
    worker.join(timeout=1)

    assert not worker.is_alive()
    assert succeeded is True
    assert value == "done"

Здесь таймаут остаётся, но его роль другая. Он защищает тест от вечного ожидания, а не изображает завершение работы. Если задача зависла, тест завершится понятным сбоем по таймауту. При этом проверка join не позволяет оставить фоновый поток после теста.

Случайность должна иметь управляемый источник

Вызов глобального генератора случайных чисел в бизнес-логике и тесте одновременно делает причину сбоя неясной. Ревьюйте такие места по двум вопросам: кто создаёт генератор и можно ли передать его в код явно.

import random


def choose_server(servers: list[str]) -> str:
    return random.choice(servers)

Для алгоритма, поведение которого нужно проверять по сценариям, передайте экземпляр генератора. Это не делает результат «правильным» само по себе, зато позволяет повторить конкретную последовательность.

import random


def choose_server(
    servers: list[str],
    generator: random.Random,
) -> str:
    return generator.choice(servers)


def test_choose_server_is_repeatable():
    first_run = random.Random(42)
    second_run = random.Random(42)

    first_result = [choose_server(["a", "b", "c"], first_run) for _ in range(5)]
    second_result = [choose_server(["a", "b", "c"], second_run) for _ in range(5)]

    assert first_result == second_result

Не подменяйте этим тестом проверку распределения. Фиксированный seed нужен для сценария и диагностики. Статистические свойства выбора проверяются отдельными тестами с заранее определёнными критериями и объёмом выборки.

Сравните подходы до рефакторинга

Сноска. Детерминированность — способность получить тот же результат при тех же входных данных и условиях. Она не означает, что система всегда должна быть последовательной: конкурентный код может сохранять параллельность, но его границы и результаты должны быть наблюдаемыми.

Как проводить ревью плавающего теста

1. Найдите источник времени. Отметьте каждый sleep, таймаут, повторный запрос и ожидание очереди. Для каждого вызова спросите, какое событие он должен заменить.
2. Проверьте общий доступ. Ищите глобальные списки, словари, кеши, переменные окружения, временные файлы и таблицы базы данных. Если тесты меняют один ресурс, проверьте их запуск в другом порядке.
3. Разделите причины. Ошибка внешнего сервиса, гонка потоков и недетерминированный тест требуют разных исправлений. Не лечите сетевой сбой фиксацией random seed.
4. Сделайте сбой диагностичным. Добавьте в сообщение идентификатор сценария, фактическое состояние и границу таймаута. Логи не должны зависеть от случайного порядка строк.
5. Проверьте изоляцию. После теста должны быть закрыты соединения, завершены потоки, очищены временные файлы и возвращены изменённые настройки. Иначе следующий тест может получить чужое состояние.

Хороший комментарий в ревью описывает механизм: «Здесь читается общий словарь после запуска потока без синхронизации; результат зависит от того, успел ли worker выполнить запись». Плохой комментарий — «потенциальная проблема с потоками» без сценария, наблюдаемого симптома и предложения по проверке.

Что ломается после рефакторинга

Таймаут превращается в скрытый бесконечный цикл

После отказа от sleep разработчик иногда заменяет его циклом с повторной проверкой состояния. Если у цикла нет общего срока, зависший worker оставит CI ждать бесконечно. Таймаут должен быть на границе ожидания, а его значение — частью контракта теста, а не случайным числом внутри функции.

Ошибка потока снова теряется

Сам факт завершения потока не доказывает успешность задачи. Если исключение произошло внутри worker-а, тест может увидеть только пустой результат. Передавайте наружу либо значение, либо исключение, как в примере с очередью, и проверяйте оба варианта отдельными тестами.

Фиксированный seed создаёт ложную уверенность

Один seed проверяет один маршрут выполнения. После исправления оставьте регрессионный тест с ним, но отдельно запускайте несколько заранее выбранных seed-значений или используйте генерацию сценариев, если это предусмотрено вашим тестовым инструментом. Номер seed записывайте в отчёт, иначе найденный случай будет трудно восстановить.

Логи меняют порядок

В многопоточном приложении порядок строк в логе не обязан совпадать с порядком событий. Для анализа добавьте время, идентификатор операции и имя потока, но не стройте тест только на порядке логов. Проверяйте состояние или сообщение, которое является частью контракта.

Когда считать проблему закрытой

Рефакторинг можно отправлять в ревью, если у вас есть исходный симптом, минимальный сценарий и объяснение причины. После изменения выполните следующие проверки:

• проблемный тест запускается отдельно и в полном наборе;
• проверены разные порядки тестов и чистое окружение;
• таймаут приводит к понятному сбою, а не к зависанию;
• ошибка фоновой работы попадает в основной поток проверки;
• внешние ресурсы закрываются даже при исключении;
• в отчёте указаны Python 3.12, pytest 8.x и условия запуска.

Если сбой связан с сетью или базой данных, измерьте отдельно время подключения, запроса и обработки ответа. Не смешивайте эти значения в один большой таймаут: иначе после инцидента будет непонятно, что именно замедлилось.

Что делать дальше

Сохраните минимальный воспроизводимый тест как регрессионный и добавьте его в обычный прогон. В CI наблюдайте длительность теста, число таймаутов и повторяемость падений по окружениям. Следующим шагом проверьте соседние тесты, которые используют тот же кеш, временный каталог, worker или генератор случайных чисел: плавающий сбой часто появляется не в одном месте, а на границе между двумя сценариями.