aio.testing 0.1.0

构建状态

安装

需要python >= 3.4

安装方式

pip install aio.testing

Aio测试提供了2个装饰器来运行asyncio测试

• aio.testing.run_until_complete:

  • 创建一个测试循环

  • 使用loop.run_until_complete调用测试

• aio.testing.run_forever:

  • 创建一个测试循环

  • 使用loop.run_forever调用测试

  • 等待“timeout”中指定的秒数（默认=1）

  • 如果测试返回一个可调用对象，则将其作为协程调用

  • 等待“sleep”中指定的秒数（默认=0）

@aio.testing.run_until_complete

aio.testing提供了一个方法装饰器来运行基于asyncio的测试

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)

在测试运行之前，会调用asyncio.get_new_loop()并将其设置为asyncio.set_event_loop().

在测试完成后，会再次调用asyncio.set_event_loop()，并使用原始的事件循环。

@aio.testing.run_forever

如果您的代码需要测试长时间运行的任务，您可以使用@aio.testing.run_forever装饰器。

@aio.testing.run_forever装饰器使用loop.run_forever来运行测试。

任何所需的设置都可以在测试函数的主体中完成，该函数可以可选地返回一个测试回调

回调被封装在协程中，并在1秒后调用

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

与aio.testing.run_until_complete一样，测试是在一个单独的循环中运行的。

@aio.testing.run_forever带超时

您可以通过设置超时值来指定在运行回调测试之前等待多少秒

import unittest
import asyncio

import aio.testing


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带sleep

有时测试需要在服务停止并且测试循环被销毁之后等待一段时间。

您可以通过设置休眠值来指定在运行回调测试后等待多少秒。

import unittest
import asyncio

import aio.testing


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.run_until_complete

让我们创建一个测试

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

然后检查测试循环是否与当前循环不同

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

测试运行后，我们恢复了原始的事件循环

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

我们可以在测试中引发错误

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

让我们创建一个未来测试

>>> import asyncio

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

与 aio.testing.run_until_complete 类似，测试在一个单独的循环中运行

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

同样，测试运行后，我们恢复了原始的事件循环

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

如果测试返回一个可调用对象，它将在1秒后调用。

test_callback 在与测试相同的循环中运行

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

如果 test_callback 不是一个协程，它将始终被 asyncio.coroutine 包装

>>> @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!

我们可以在测试中引发错误

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

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

我们可以在测试回调中引发错误

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

默认情况下，test_callback 在返回后1秒被调用

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

您可以通过在装饰器中设置“timeout”参数来设置在调用 test_callback 之前等待的时间

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

您还可以通过在装饰器中设置“sleep”参数来设置测试完成后等待的时间

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