aio.testing 0.0.9

Testing utils for aio asyncio framework

aio.testing

Test utils for the aio asyncio framework

Build status

Installation

Install with:

pip install aio.testing

Aio testing provides 2 decorators for running asyncio tests

• aio.testing.run_until_complete:

  • creates a test loop

  • calls the test with loop.run_until_complete

• aio.testing.run_forever:

  • creates a test loop

  • calls test using loop.run_forever

  • waits for number of seconds specified in “timeout” (default = 1)

  • if test returns a callable, calls it as a coroutine

  • waits for number of seconds specified in “sleep” (default = 0)

@aio.testing.run_until_complete

aio.testing provides a method decorator for running asyncio-based tests

import unittest
import asyncio

import aio.testing


class MyTestCase(unittest.TestCase):

    @aio.testing.run_until_complete
    def test_example():
        yield from asyncio.sleep(2)
        self.assertTrue(True)

Prior to the test running asyncio.get_new_loop() is called and set using asyncio.set_event_loop().

On completion of the test asyncio.set_event_loop() is again called with the original event loop.

@aio.testing.run_forever

If your code needs to test long-running tasks, you can use the @aio.testing.run_forever decorator.

The @aio.testing.run_forever decorator uses loop.run_forever to run the test.

Any setup required can be done in the body of the test function which can optionally return a test callback

The callback is wrapped in a coroutine, and called after 1 second

import unittest
import asyncio

import aio.testing


class MyFutureTestCase(unittest.TestCase):

    @aio.testing.run_forever
    def test_example():
        yield from asyncio.sleep(2)

        def callback_test(self):
            yield from asyncio.sleep(2)
            self.assertTrue(True)

        # this function is called 1 second after being returned
        return callback_test

As with aio.testing.run_until_complete, the test is run in a separate loop.

@aio.testing.run_forever with timeout

You can specify how many seconds to wait before running the callback tests by setting the timeout value

import unittest
import asyncio

from aio.testing import aio.testing.run_forever


class MyFutureTestCase(unittest.TestCase):

    @aio.testing.run_forever(timeout=10)
    def test_example():
        yield from asyncio.sleep(2)

        def callback_test(self):
            yield from asyncio.sleep(2)
            self.assertTrue(True)

        # this function is called 10 seconds after being returned
        return callback_test

@aio.testing.run_forever with sleep

Sometimes a test needs to wait for some time after services have been stopped and the test loop has been destroyed.

You can specify how many seconds to wait after running the callback tests by setting the sleep value

import unittest
import asyncio

from aio.testing import aio.testing.run_forever


class MyFutureTestCase(unittest.TestCase):

    @aio.testing.run_forever(sleep=10)
    def test_example():
        yield from asyncio.sleep(2)

        def callback_test(self):
            yield from asyncio.sleep(2)
            self.assertTrue(True)

        return callback_test

aio.testing usage

aio.testing.run_until_complete

Lets create a test

>>> import asyncio
>>> import aio.testing

>>> @aio.testing.run_until_complete
... def run_test(parent_loop):
...     yield from asyncio.sleep(1)
...
...     print(asyncio.get_event_loop() != parent_loop)

And lets check that the test loop is not the same as the current one

>>> loop_before_test = asyncio.get_event_loop()
>>> run_test(loop_before_test)
True

After the test has run we have the original event loop back

>>> asyncio.get_event_loop() == loop_before_test
True

We can raise an error in the test

>>> @aio.testing.run_until_complete
... def run_test():
...     assert(True == False)

>>> try:
...     run_test()
... except Exception as e:
...     print(repr(e))
AssertionError()

aio.testing.run_forever

Lets create a future test

>>> import asyncio

>>> @aio.testing.run_forever
... def run_test(parent_loop):
...     yield from asyncio.sleep(1)
...
...     print(asyncio.get_event_loop() != parent_loop)

Just like with aio.testing.run_until_complete, the test is run in a separate loop

>>> loop_before_test = asyncio.get_event_loop()
>>> run_test(loop_before_test)
True

And again, after the test has run we have the original event loop back

>>> asyncio.get_event_loop() == loop_before_test
  True

If the test returns a callable, its called 1 second later.

The test_callback runs in the same loop as the test

>>> @aio.testing.run_forever
... def run_test():
...     test_loop = asyncio.get_event_loop()
...
...     @asyncio.coroutine
...     def test_callback():
...         print(
...             asyncio.get_event_loop() == test_loop)
...
...     return test_callback

>>> run_test()
True

The test_callback is always wrapped in asyncio.coroutine if its not one already

>>> @aio.testing.run_forever
... def run_test():
...
...     def test_callback():
...         yield from asyncio.sleep(1)
...         print("test_callback is always wrapped in a coroutine!")
...
...     return test_callback

>>> run_test()
test_callback is always wrapped in a coroutine!

We can raise an error in the test

>>> @aio.testing.run_forever
... def run_test():
...     assert(True == False)

>>> try:
...     run_test()
... except Exception as e:
...     print(repr(e))
AssertionError()

And we can raise an error in the test callback

>>> @aio.testing.run_forever
... def run_test():
...
...     def test_callback():
...         assert(True == False)
...
...     return test_callback

>>> try:
...     run_test()
... except Exception as e:
...     print(repr(e))
AssertionError()

By default the test_callback is called 1 second after being returned

>>> import time

>>> @aio.testing.run_forever
... def run_test():
...     test_run_at = int(time.time())
...
...     return lambda: (
...         print("callback called %s second(s) after test" % (
...             int(time.time()) - test_run_at)))

>>> run_test()
callback called 1 second(s) after test

You can set the amount of time to wait before calling the test_callback by setting the “timeout” argument in the decorator

>>> import time

>>> @aio.testing.run_forever(timeout=3)
... def run_test():
...     test_run_at = int(time.time())
...
...     return lambda: print(
...         "callback called %s second(s) after test" % (
...             int(time.time()) - test_run_at))

>>> run_test()
callback called 3 second(s) after test

You can also set the amount of time to wait after the test has completely finished, by setting the “sleep” argument on the decorator

>>> @aio.testing.run_forever(sleep=3)
... def run_test(test_time):
...     return lambda: (
...         test_time.__setitem__('completed_at', int(time.time())))

>>> test_time = {}
>>> run_test(test_time)

>>> print("test waited %s second(s) after completing" % (
...     int(time.time()) - test_time['completed_at']))
test waited 3 second(s) after completing