aio.testing 0.1.0

构建状态

安装

需要python>；=3.4

安装时使用：

pip install aio.testing

aio测试提供两个decorator来运行异步测试

• aio.testing.run直至完成：
  • 创建测试循环
  • 使用loop.run_调用测试直到完成
• aio.testing.run永远运行：
  • 创建测试循环
  • 使用loop.run_forever调用测试
  • 等待“超时”中指定的秒数（默认值为1）
  • 如果test返回一个可调用的，则将其作为协程调用
  • 等待“睡眠”中指定的秒数（默认值为0）

@所有测试。运行直至完成

testing为运行基于异步的测试提供了方法装饰器

importunittestimportasyncioimportaio.testingclassMyTestCase(unittest.TestCase):@aio.testing.run_until_completedeftest_example():yield fromasyncio.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装饰程序使用loop.run_forever运行测试。

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

回调被包装在一个协程中，并在1秒后调用

importunittestimportasyncioimportaio.testingclassMyFutureTestCase(unittest.TestCase):@aio.testing.run_foreverdeftest_example():yield fromasyncio.sleep(2)defcallback_test(self):yield fromasyncio.sleep(2)self.assertTrue(True)# this function is called 1 second after being returnedreturncallback_test

与aio.testing.run_直到完成一样，测试在单独的循环中运行。

@aio.testing.run_永远超时

通过设置超时值，可以指定在运行回调测试之前等待的秒数

importunittestimportasyncioimportaio.testingclassMyFutureTestCase(unittest.TestCase):@aio.testing.run_forever(timeout=10)deftest_example():yield fromasyncio.sleep(2)defcallback_test(self):yield fromasyncio.sleep(2)self.assertTrue(True)# this function is called 10 seconds after being returnedreturncallback_test

@所有的测试。在睡眠

importunittestimportasyncioimportaio.testingclassMyFutureTestCase(unittest.TestCase):@aio.testing.run_forever(sleep=10)deftest_example():yield fromasyncio.sleep(2)defcallback_test(self):yield fromasyncio.sleep(2)self.assertTrue(True)returncallback_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)

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

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

所有测试。永远运行

让我们创建一个未来的测试

>>> 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_回调与测试在同一个循环中运行

>>> @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_回调不是asyncio.coroutine，则它始终包装在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()

默认情况下，返回后1秒调用test_回调

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

通过在decorator中设置“timeout”参数，可以设置调用test_回调之前等待的时间量

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

您还可以通过在decorator上设置“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