aioloop-proxy 0.0.13

安装

pip install aioloop-proxy

用法

import asyncio
import aioloop_proxy

loop = asyncio.new_event_loop()
server_addr = loop.run_until_complete(setup_and_run_test_server())
...

with aioloop_proxy(loop, strict=True) as loop_proxy:
   loop_proxy.run_until_complete(test_func(server_addr))

当然，每个测试系统（unittest、pytest，等等）不应直接运行上面的代码片段，而应将其作为专门的测试用例类或插件整合。

额外的循环方法

LoopProxy实现了所有asyncio.AbstractEventLoop的公共方法。此外，它还提供了两个特定的代理方法：loop.check_and_shutdown()和loop.advance_time()。

await proxy.check_and_shutdown(kind=CheckKind.ALL)可用于检查代理是否在没有活动任务、打开传输等情况下完成。

kind是一个enum.Flag，描述如下

class CheckKind(enum.Flag):
    TASKS = enum.auto()
    SIGNALS = enum.auto()
    SERVERS = enum.auto()
    TRANSPORTS = enum.auto()
    READERS = enum.auto()
    WRITERS = enum.auto()
    HANDLES = enum.auto()

    ALL = TASKS | SIGNALS | SERVERS | TRANSPORTS | READERS | WRITERS

默认情况下，所有检查都执行。如果某个特定测试引发了误报警告，则可以省略某些检查。

注意即使省略了相应的kind，悬空资源也会始终关闭。为了保持测试的隔离原则，代理循环应在测试结束时清理所有获取的资源。

proxy.advance_time(offset)是一个特性，有助于编写使用超时、延迟等场景的测试。

假设我们有一个代码，应该从对等方读取数据或在对15分钟超时后引发TimeoutError。这可以通过将代理本地时间（proxy.time()返回的值）人工向前推进15分钟来实现。

task = asyncio.create_task(fetch_or_timeout())
loop.advance_time(15 * 60)
try:
    await task
except TimeoutError:
    ...

在上面的示例中，await task立即恢复，因为测试的wall-clock被向前推进了15分钟（在上面的两行中），并且所有由代理创建的计时器都相应地调整。

父循环的wall-clock不会被触及。

该方法的时间复杂度是O(N)，其中N是proxy.call_later()或proxy.call_at()方法创建的激活计时器的数量。