Skip to main content

Asyncio loop proxy for testing purposes

Project description

A proxy for asyncio.AbstractEventLoop for testing purposes.

When tests writing for asyncio based code, there are controversial requirements.

First, a signle event loop for the whole test session (or test subset) is desired. For example, if web server starts slowly, there is a temptation to create a server only once and access to the single web server instance from each test.

Second, each test should be isolated. It means that asyncio tasks (timers, connections, etc.) created by test A should be finished at test A finalization and should not affect test B execution.

The library provides loop proxy class that fully implements asyncio.AbstractEventLoop interface but redirects all actual work to the proxied parent loop. It allows to check that all activities created with the proxy are finished on the proxy finalization. In turn, all tasks created with the parent loop are still keep working during the proxy execution.

Loop proxies can be nested, e.g. global-loop -> module-loop -> test-loop is supported.

The library is test tool agnostic, e.g. it can be integrated with unittest and pytest easily (the actual integraion is out of the project scope).

Installation

pip install aioloop-proxy

Usage

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))

Sure, each test system (unittest, pytest, name it) should not run the code snippet above as-is but incorporate it as a dedicates test-case class or plugin.

Extra loop methods

LoopProxy implements all asyncio.AbstractEventLoop public methods. Additionally, it provides two proxy-specific ones: loop.check_and_shutdown() and loop.advance_time().

await proxy.check_and_shutdown(kind=CheckKind.ALL) can be used for checking if the proxy finished without active tasks, open transports etc.

kind is a enum.Flag described as the following:

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

All checks are performed by default. A specific test can omit some check if it raises a false positive warning.

N.B. Dangling resources are always closed even if corresponding kind is omitted. A proxy loop should cleanup all acquired resources at the test finish for the sake of tests isolation principle.

proxy.advance_time(offset) is a perk that helps with writing tests for scenarios that uses timeouts, delays, etc.

Let’s assume, we have a code that should read data from peer or raise TimeoutError after 15 minute timeout. It can be done by shifting the proxy local time (proxy.time() returned value) to 15 minutes forward artificially:

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

In the example above, await task is resumed immediatelly because the test wall-clock is shifted by 15 minutes two lines above, and all timers created by the proxy are adjusted accordingly.

The parent loop wall-clock is not touched.

The method complexity is O(N) where N is amount of active timers created by proxy.call_later() or proxy.call_at() methods.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aioloop-proxy-0.0.1.tar.gz (11.8 kB view details)

Uploaded Source

Built Distribution

aioloop_proxy-0.0.1-py3-none-any.whl (11.1 kB view details)

Uploaded Python 3

File details

Details for the file aioloop-proxy-0.0.1.tar.gz.

File metadata

  • Download URL: aioloop-proxy-0.0.1.tar.gz
  • Upload date:
  • Size: 11.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.7.1 importlib_metadata/4.10.0 pkginfo/1.8.2 requests/2.26.0 requests-toolbelt/0.9.1 tqdm/4.62.3 CPython/3.10.1

File hashes

Hashes for aioloop-proxy-0.0.1.tar.gz
Algorithm Hash digest
SHA256 51fedcb7b7d8c735486b7482f7949e92af94ee2c13e788b568d6aad34a6f090d
MD5 96534d230d090400a83bd98e1d98a6ed
BLAKE2b-256 22c69d087e5a091d999587a8e2fe00566142e18f2864e326da37ddf0942237d0

See more details on using hashes here.

Provenance

File details

Details for the file aioloop_proxy-0.0.1-py3-none-any.whl.

File metadata

  • Download URL: aioloop_proxy-0.0.1-py3-none-any.whl
  • Upload date:
  • Size: 11.1 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/3.7.1 importlib_metadata/4.10.0 pkginfo/1.8.2 requests/2.26.0 requests-toolbelt/0.9.1 tqdm/4.62.3 CPython/3.10.1

File hashes

Hashes for aioloop_proxy-0.0.1-py3-none-any.whl
Algorithm Hash digest
SHA256 b8bc2b0fe4687e1ae3dce8f4efb5a2fded511016170fb9ea31036b84b1f536e4
MD5 fef332d8a7194bcfce957e324c23dc48
BLAKE2b-256 67fa888f7f250df72f89256db4d9819660441e08c631c89643cc0db98f5ca135

See more details on using hashes here.

Provenance

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page