Тестирование стало проще с помощью trio.testing

Интеграция с тестовым каркасом

@trio.testing.trio_test

Время и таймауты

trio.testing.MockClock — это Clock с несколькими хитростями, помогающими эффективно тестировать код с таймаутами:

• По умолчанию, он начинается с времени 0, и время по часам увеличивается только при явном вызове jump(). Это обеспечивает чрезвычайно управляемый механизм времени для тестирования.

• Можно установить rate в 1.0, если нужно, чтобы он работал в реальном времени, как обычные часы. Можно останавливать и запускать часы в рамках теста. Можно установить rate в 10.0, чтобы время по часам проходило в 10 раз быстрее (например, await trio.sleep(10) вернётся через 1 секунду).

• Более интересно, можно установить autojump_threshold в ноль или малое значение, и тогда он будет следить за выполнением цикла обработки событий, а всякий раз, когда все завершится, и все ждут таймаута, он переведёт часы вперёд до этого таймаута. Во многих случаях это позволяет естественно выглядящему коду, включающему таймауты, работать автоматически с максимальной загрузкой ЦП без изменений. (Спасибо fluxcapacitor за эту замечательную идею.)

• Конечно, всё это можно смешивать и сочетать по своему усмотрению.

Независимо от этих ухищрений, внутри Trio течение времени всё ещё выглядит нормальным, если вы ограничиваетесь функциями времени Trio (см. Время и часы). Ниже приведен пример, демонстрирующий два разных способа ускорения времени. Обратите внимание, как в обоих случаях две задачи сохраняют согласованное представление о реальности, и события происходят в ожидаемой последовательности, несмотря на значительное расхождение с реальным временем:

# across-realtime.py

import time
import trio
import trio.testing

YEAR = 365 * 24 * 60 * 60  # seconds


async def task1():
    start = trio.current_time()

    print("task1: sleeping for 1 year")
    await trio.sleep(YEAR)

    duration = trio.current_time() - start
    print(f"task1: woke up; clock says I've slept {duration / YEAR} years")

    print("task1: sleeping for 1 year, 100 times")
    for _ in range(100):
        await trio.sleep(YEAR)

    duration = trio.current_time() - start
    print(f"task1: slept {duration / YEAR} years total")


async def task2():
    start = trio.current_time()

    print("task2: sleeping for 5 years")
    await trio.sleep(5 * YEAR)

    duration = trio.current_time() - start
    print(f"task2: woke up; clock says I've slept {duration / YEAR} years")

    print("task2: sleeping for 500 years")
    await trio.sleep(500 * YEAR)

    duration = trio.current_time() - start
    print(f"task2: slept {duration / YEAR} years total")


async def main():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(task1)
        nursery.start_soon(task2)


def run_example(clock):
    real_start = time.perf_counter()
    trio.run(main, clock=clock)
    real_duration = time.perf_counter() - real_start
    print(f"Total real time elapsed: {real_duration} seconds")


print("Clock where time passes at 100 years per second:\n")
run_example(trio.testing.MockClock(rate=100 * YEAR))

print("\nClock where time automatically skips past the boring parts:\n")
run_example(trio.testing.MockClock(autojump_threshold=0))

Вывод:

Clock where time passes at 100 years per second:

task2: sleeping for 5 years
task1: sleeping for 1 year
task1: woke up; clock says I've slept 1.0365006048232317 years
task1: sleeping for 1 year, 100 times
task2: woke up; clock says I've slept 5.0572111969813704 years
task2: sleeping for 500 years
task1: slept 104.77677842136472 years total
task2: slept 505.25014589075 years total
Total real time elapsed: 5.053582429885864 seconds

Clock where time automatically skips past the boring parts:

task2: sleeping for 5 years
task1: sleeping for 1 year
task1: woke up; clock says I've slept 1.0 years
task1: sleeping for 1 year, 100 times
task2: woke up; clock says I've slept 5.0 years
task2: sleeping for 500 years
task1: slept 101.0 years total
task2: slept 505.0 years total
Total real time elapsed: 0.019298791885375977 seconds

class trio.testing.MockClock(rate: float = 0.0, autojump_threshold: float = inf)

Управляемый пользователем механизм часов, подходящий для написания тестов.

Параметры:

• rate (float) – начальная rate.

• autojump_threshold (float) – начальный autojump_threshold.

rate

Количество секунд времени по часам, проходящих за секунду реального времени. Значение по умолчанию равно 0.0, т.е. часы увеличиваются только при ручных вызовах jump() или при срабатывании autojump_threshold. Вы можете изменить это значение.

autojump_threshold

Часы отслеживают цикл обработки событий, и если в какой-то момент обнаруживают, что все задачи заблокированы в течение этого количества реальных секунд (т.е. по реальным часам, а не по этим часам), то часы автоматически переходят к следующему запланированному таймауту цикла обработки событий. По умолчанию это math.inf, т.е. никогда не происходит автоматического перехода. Вы можете изменить это значение.

В принципе, идея заключается в том, что если у вас есть код или тесты, использующие паузы и таймауты, то вы можете использовать это, чтобы запустить их гораздо быстрее, полностью автоматически. (По крайней мере, пока эти паузы/таймауты происходят внутри Trio; если ваш тест включает взаимодействие с внешней службой и ожидание её таймаута, то, очевидно, мы вам не поможем.)

Вы должны установить это значение на минимальное значение, которое позволит избежать «ложных срабатываний», где какое-то ввод/вывод (например, между двумя половинами пары сокетов) находится в процессе, но порог срабатывает, и время всё равно увеличивается. Это будет зависеть от деталей ваших тестов и тестовой среды. Если вы не выполняете ввод/вывод (как в приведённом выше примере с ожиданием), то установите его в ноль, и часы будут переключаться всякий раз, когда все задачи заблокированы.

Примечание

Если вы используете autojump_threshold и wait_all_tasks_blocked одновременно, то вы можете задаться вопросом, как они взаимодействуют, так как оба вызывают события после того, как цикл обработки событий становится неактивным в течение некоторого времени. Ответ: wait_all_tasks_blocked имеет приоритет. Если есть задача, заблокированная в wait_all_tasks_blocked, то функция автоматического перехода рассматривает её как активную задачу и не переводит часы.

jump(seconds: float) → None

Ручное продвижение часов на заданное количество секунд.

Параметры:

seconds (float) – количество секунд, на которое нужно продвинуть часы вперёд.

Исключения:

ValueError – если вы пытаетесь передать отрицательное значение для seconds.

Порядок выполнения задач

class trio.testing.Sequencer

Удобный класс для принудительного выполнения кода в разных задачах в явном линейном порядке.

Экземпляры этого класса реализуют метод __call__, который возвращает асинхронный менеджер контекста. Идея состоит в том, что вы передаете порядковый номер в __call__, чтобы указать, куда этот блок кода должен поместиться в линейной последовательности. Блок 0 запускается немедленно, а затем блок N не запускается до тех пор, пока блок N-1 не завершит работу.

Пример

Крайне сложный способ вывести числа от 0 до 5 в порядке:

async def worker1(seq):
    async with seq(0):
        print(0)
    async with seq(4):
        print(4)

async def worker2(seq):
    async with seq(2):
        print(2)
    async with seq(5):
        print(5)

async def worker3(seq):
    async with seq(1):
        print(1)
    async with seq(3):
        print(3)

async def main():
   seq = trio.testing.Sequencer()
   async with trio.open_nursery() as nursery:
       nursery.start_soon(worker1, seq)
       nursery.start_soon(worker2, seq)
       nursery.start_soon(worker3, seq)

await trio.testing.wait_all_tasks_blocked(cushion: float = 0.0) → None

Ожидание, пока не останется задач, готовых к выполнению.

Это полезно при тестировании кода, когда вы хотите дать другим задачам возможность «успокоиться». Вызывающая задача блокируется и не разблокируется до тех пор, пока все другие задачи также не заблокированы не менее чем на cushion секунды. (Установка ненулевого значения cushion предназначена для обработки случаев, подобных двум задачам, общающимся друг с другом через локальный сокет, где мы хотим проигнорировать потенциальный краткий момент между отправкой и получением, когда все задачи заблокированы.)

Обратите внимание, что измерение cushion выполняется в реальном времени, а не в времени трио.

Если несколько задач заблокированы в wait_all_tasks_blocked(), то разблокируется та, у которой наименьшее время cushion, и разблокирование этой задачи сбрасывает таймеры для оставшихся задач. Если у нескольких задач одинаковое время cushion, то все они разблокируются.

Вы также можете рассмотреть trio.testing.Sequencer, который предоставляет более явный способ управления порядком выполнения в рамках теста и часто создает более читаемые тесты.

Пример

Вот пример того, как можно проверить, что блокировки трио справедливы: мы берем блокировку в родительском процессе, запускаем дочерний процесс, ждем, пока дочерний процесс заблокируется, ожидая блокировки (!), и затем проверяем, что мы не можем освободить и сразу же повторно получить блокировку:

async def lock_taker(lock):
    await lock.acquire()
    lock.release()

async def test_lock_fairness():
    lock = trio.Lock()
    await lock.acquire()
    async with trio.open_nursery() as nursery:
        nursery.start_soon(lock_taker, lock)
        # child hasn't run yet, we have the lock
        assert lock.locked()
        assert lock._owner is trio.lowlevel.current_task()
        await trio.testing.wait_all_tasks_blocked()
        # now the child has run and is blocked on lock.acquire(), we
        # still have the lock
        assert lock.locked()
        assert lock._owner is trio.lowlevel.current_task()
        lock.release()
        try:
            # The child has a prior claim, so we can't have it
            lock.acquire_nowait()
        except trio.WouldBlock:
            assert lock._owner is not trio.lowlevel.current_task()
            print("PASS")
        else:
            print("FAIL")

await trio.testing.wait_all_threads_completed() → None

Ожидание, пока ни одна нить не будет продолжать работу с задачами.

Это предназначено для использования при тестировании кода с помощью `trio.to_thread`, чтобы убедиться, что ни одна задача не делает дальнейшего прогресса в потоке. Пример использования приведен в следующем коде:

async def wait_all_settled():
    while True:
        await trio.testing.wait_all_threads_complete()
        await trio.testing.wait_all_tasks_blocked()
        if trio.testing.active_thread_count() == 0:
            break

trio.testing.active_thread_count() → int

Возвращает количество потоков, которые в данный момент выполняют задачу.

См. trio.testing.wait_all_threads_completed

Потоки

Виртуальная сеть для тестирования

В предыдущем разделе вы узнали, как использовать виртуальные потоки в памяти для тестирования протоколов, написанных с использованием абстракции Trio’s Stream. Но что делать, если у вас есть более сложный сетевой код – такой, который устанавливает соединения с несколькими хостами, открывает сокет для прослушивания или отправляет пакеты UDP?

Сам Trio не предоставляет виртуальную сетевую реализацию в памяти для тестирования – но модуль trio.socket предоставляет необходимые крючки для написания собственной! Если вы заинтересованы в реализации многоразовой виртуальной сети для тестирования, свяжитесь с нами здесь.

Обратите внимание, что эти API фактически находятся в trio.socket и trio.abc, но мы документируем их здесь, потому что они предназначены в первую очередь для тестирования.

trio.socket.set_custom_hostname_resolver(hostname_resolver: HostnameResolver | None) → HostnameResolver | None

Установите пользовательский разрешитель имен хоста.

По умолчанию функции Trio’s getaddrinfo() и getnameinfo() используют стандартные системные функции разрешения. Эта функция позволяет настроить это поведение. Основной случай использования – для тестирования, но она также может быть полезна для использования сторонних разрешителей, таких как c-ares (хотя будьте осторожны, они редко являются полными аналогами системного разрешителя). Подробнее см. trio.abc.HostnameResolver.

Установка пользовательского разрешителя имен хоста влияет на все последующие вызовы getaddrinfo() и getnameinfo() внутри вложенного вызова trio.run(). Всё остальное разрешение имён хостов в Trio реализуется через эти функции.

Как правило, вы должны вызвать эту функцию только один раз, в самом начале вашей программы.

Параметры:

hostname_resolver (trio.abc.HostnameResolver или None) – Новый пользовательский разрешитель имен хоста или None для восстановления поведения по умолчанию.

Возвращаемое значение:

Предыдущий разрешитель имен хоста (который может быть None).

class trio.abc.HostnameResolver

Если у вас есть пользовательский разрешитель имен хоста, то реализация HostnameResolver позволяет зарегистрировать его для использования в Trio.

См. trio.socket.set_custom_hostname_resolver().

abstractmethod await getaddrinfo(host: bytes | None, port: bytes | str | int | None, family: int = 0, type: int = 0, proto: int = 0, flags: int = 0) → list[tuple[AddressFamily, SocketKind, int, str, tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes]]]

Пользовательская реализация getaddrinfo().

Вызывается trio.socket.getaddrinfo().

Если host задан как числовой IP-адрес, то getaddrinfo() может обработать запрос сам, а не вызывать этот метод.

Любое необходимое кодирование IDNA выполняется перед вызовом этой функции; ваша реализация может считать, что никогда не увидит U-метки, например, "café.com", и ей нужно обрабатывать только A-метки, например, b"xn--caf-dma.com".

abstractmethod await getnameinfo(sockaddr: tuple[str, int] | tuple[str, int, int, int], flags: int) → tuple[str, str]

Пользовательская реализация getnameinfo().

Вызывается trio.socket.getnameinfo().

trio.socket.set_custom_socket_factory(socket_factory: SocketFactory | None) → SocketFactory | None

Установите пользовательский фабричный метод для создания сокетов.

Эта функция позволяет заменить стандартный класс сокетов Trio на пользовательский. Это очень полезно для тестирования, и, вероятно, плохая идея в других случаях. Подробнее см. trio.abc.HostnameResolver.

Установка пользовательского фабричного метода для создания сокетов влияет на все последующие вызовы socket() внутри вложенного вызова trio.run().

Как правило, вы должны вызвать эту функцию только один раз, в самом начале вашей программы.

Параметры:

socket_factory (trio.abc.SocketFactory или None) – Новый пользовательский фабричный метод или None для восстановления поведения по умолчанию.

Возвращаемое значение:

Предыдущий фабричный метод (который может быть None).

class trio.abc.SocketFactory

Если вы пишете пользовательский класс, реализующий интерфейс сокетов Trio, то вы можете использовать SocketFactory для того, чтобы Trio использовал его.

См. trio.socket.set_custom_socket_factory().

abstractmethod socket(family: socket.AddressFamily | int = AddressFamily.AF_INET, type: socket.SocketKind | int = SocketKind.SOCK_STREAM, proto: int = 0) → SocketType

Создайте и верните объект сокета.

Ваш объект сокета должен наследоваться от trio.socket.SocketType, который является пустым классом, единственной целью которого является «разметка» классов, которые должны считаться валидными сокетами Trio.

Вызывается trio.socket.socket().

Обратите внимание, что в отличие от trio.socket.socket(), это не принимает fileno= аргумент. Если fileno= задан, то trio.socket.socket() возвращает обычный объект сокета Trio вместо вызова этого метода.

Тестирование контрольных точек

with trio.testing.assert_checkpoints() → AbstractContextManager[None]

Используйте в качестве менеджера контекста, чтобы проверить, что код внутри блока with завершается с исключением или выполняет по крайней мере одну контрольную точку.

Исключения:

AssertionError – если никакая контрольная точка не была выполнена.

Пример

Проверка того, что trio.sleep() является контрольной точкой, даже если она не блокирует:

with trio.testing.assert_checkpoints():
    await trio.sleep(0)

with trio.testing.assert_no_checkpoints() → AbstractContextManager[None]

Используйте в качестве менеджера контекста, чтобы проверить, что код внутри блока with не выполняет никаких контрольных точек.

Исключения:

AssertionError – если была выполнена контрольная точка.

Пример

Синхронный код никогда не содержит контрольных точек, но мы можем это проверить:

send_channel, receive_channel = trio.open_memory_channel(10)
with trio.testing.assert_no_checkpoints():
    send_channel.send_nowait(None)

Справочные материалы по ExceptionGroup

class trio.testing.RaisesGroup(exception: type[BaseExcT_1] | Matcher[BaseExcT_1] | RaisesGroup[BaseExcT_2], *other_exceptions: type[BaseExcT_1] | Matcher[BaseExcT_1] | RaisesGroup[BaseExcT_2], allow_unwrapped: bool = False, flatten_subgroups: bool = False, match: str | Pattern[str] | None = None, check: Callable[[BaseExceptionGroup[BaseExcT_1]], bool] | Callable[[ExceptionGroup[ExcT_1]], bool] | None = None)

Менеджер контекста для проверки ожидаемой ExceptionGroup. Он работает аналогично pytest.raises, и версия этой функции, надеемся, будет добавлена в исходный код, после чего эту функцию можно будет сделать устаревшей и удалить. См. https://github.com/pytest-dev/pytest/issues/11538

Поведение при перехвате отличается от except* по многим аспектам, будучи по умолчанию более жёстким. Используя allow_unwrapped=True и flatten_subgroups=True, можно полностью сопоставить except*, когда ожидается единственная ошибка.

1. Все указанные исключения должны присутствовать, и никаких других.

   • Если вы ожидаете переменное количество исключений, вам необходимо использовать pytest.raises(ExceptionGroup) и вручную проверить содержащиеся исключения. Рассмотрите возможность использования Matcher.matches().

2. По умолчанию он будет перехватывать только исключения, заключенные в группу исключений.

   • С помощью allow_unwrapped=True можно указать единственное ожидаемое исключение или Matcher, и оно будет соответствовать исключению, даже если оно не находится внутри ExceptionGroup. Если вы ожидаете один из нескольких различных типов исключений, вам необходимо использовать объект Matcher.

3. По умолчанию он учитывает полную структуру с вложенными ExceptionGroup. Можно указать вложенные ExceptionGroup, передав объекты RaisesGroup в качестве ожидаемых исключений.

   • С помощью flatten_subgroups=True он «развернёт» поднятую ExceptionGroup, извлекая все исключения внутри любых вложенных ExceptionGroup, прежде чем выполнять сопоставление.

Порядок исключений не важен, поэтому RaisesGroups(ValueError, TypeError) эквивалентно RaisesGroups(TypeError, ValueError).

Примеры:

with RaisesGroups(ValueError):
    raise ExceptionGroup("", (ValueError(),))
with RaisesGroups(ValueError, ValueError, Matcher(TypeError, match="expected int")):
    ...
with RaisesGroups(KeyboardInterrupt, match="hello", check=lambda x: type(x) is BaseExceptionGroup):
    ...
with RaisesGroups(RaisesGroups(ValueError)):
    raise ExceptionGroup("", (ExceptionGroup("", (ValueError(),)),))

# flatten_subgroups
with RaisesGroups(ValueError, flatten_subgroups=True):
    raise ExceptionGroup("", (ExceptionGroup("", (ValueError(),)),))

# allow_unwrapped
with RaisesGroups(ValueError, allow_unwrapped=True):
    raise ValueError

RaisesGroup.matches также можно использовать непосредственно для проверки автономной группы исключений.

Алгоритм сопоставления жадный, что может привести к ошибкам в таких случаях:

with RaisesGroups(ValueError, Matcher(ValueError, match="hello")):
    raise ExceptionGroup("", (ValueError("hello"), ValueError("goodbye")))

хотя в целом он не учитывает порядок исключений в группе. Чтобы избежать этого, вы должны указать первый ValueError с помощью Matcher.

Подсказка: если вы установите hypothesis и импортируете его в conftest.py, вы получите удобочитаемые repr``s of ``check вызываемые функции в выводе.

fail_reason

Устанавливается после вызова matches для предоставления удобочитаемого объяснения причины неудачи сопоставления. При использовании в качестве менеджера контекста строка будет использоваться в качестве текста AssertionError

matches(exc_val: BaseException | None) → TypeGuard[BaseExceptionGroup[BaseExcT_co]]

Проверка, соответствует ли исключение требованиям этого RaisesGroup. Если произошла неудача, RaisesGroup.fail_reason будет установлено.

Пример:

with pytest.raises(TypeError) as excinfo:
    ...
assert RaisesGroups(ValueError).matches(excinfo.value.__cause__)
# the above line is equivalent to
myexc = excinfo.value.__cause
assert isinstance(myexc, BaseExceptionGroup)
assert len(myexc.exceptions) == 1
assert isinstance(myexc.exceptions[0], ValueError)

class trio.testing.Matcher(exception_type: type[MatchE] | None = None, match: str | Pattern[str] | None = None, check: Callable[[MatchE], bool] | None = None)

Вспомогательный класс для использования совместно с RaisesGroups, когда необходимо указать требования к подисключениям. Указание только типа избыточно, а также не нужно, когда тип является вложенной RaisesGroup, так как она поддерживает те же аргументы. Тип проверяется с помощью isinstance и не обязательно должен быть точным. Если нужно точное соответствие, можно использовать параметр check. Matcher.matches() также можно использовать автономно для проверки отдельных исключений.

Примеры:

with RaisesGroups(Matcher(ValueError, match="string"))
    ...
with RaisesGroups(Matcher(check=lambda x: x.args == (3, "hello"))):
    ...
with RaisesGroups(Matcher(check=lambda x: type(x) is ValueError)):
    ...

Подсказка: если вы установите hypothesis и импортируете его в conftest.py, вы получите удобочитаемые repr``s of ``check вызываемые функции в выводе.

fail_reason

Устанавливается после вызова matches для предоставления удобочитаемого объяснения причины неудачи сопоставления. При использовании в качестве менеджера контекста строка будет использоваться в качестве текста AssertionError

matches(exception: BaseException) → TypeGuard[MatchE]

Проверка, соответствует ли исключение требованиям этого Matcher. Если произошла неудача, Matcher.fail_reason будет установлено.

Примеры:

assert Matcher(ValueError).matches(my_exception):
# is equivalent to
assert isinstance(my_exception, ValueError)

# this can be useful when checking e.g. the ``__cause__`` of an exception.
with pytest.raises(ValueError) as excinfo:
    ...
assert Matcher(SyntaxError, match="foo").matches(excinfo.value.__cause__)
# above line is equivalent to
assert isinstance(excinfo.value.__cause__, SyntaxError)
assert re.search("foo", str(excinfo.value.__cause__)

class trio.testing._raises_group._ExceptionInfo(excinfo: tuple[type[MatchE], MatchE, types.TracebackType] | None)

Минимальная реализация pytest.ExceptionInfo, используемая только в том случае, если pytest недоступен. Поддерживает подмножество его функций, необходимых для работы trio.testing.RaisesGroup и trio.testing.Matcher.

fill_unfilled(exc_info: tuple[type[MatchE], MatchE, types.TracebackType]) → None

Заполнение незаполненной ExceptionInfo, созданной с помощью for_later().

classmethod for_later() → _ExceptionInfo[MatchE]

Возвращает незаполненную ExceptionInfo.

property type: type[MatchE]

Класс исключения.

property value: MatchE

Значение исключения.

property tb: types.TracebackType

Исходный стек вызовов исключения.