擅长:python、mysql、java
<p>我用一个装饰师临时解决了这个问题,他的灵感来自于龙卷风的<a href="http://www.tornadoweb.org/en/stable/testing.html#tornado.testing.gen_test">gen_test</a>:</p>
<pre><code>def async_test(f):
def wrapper(*args, **kwargs):
coro = asyncio.coroutine(f)
future = coro(*args, **kwargs)
loop = asyncio.get_event_loop()
loop.run_until_complete(future)
return wrapper
</code></pre>
<p>正如J.F.Sebastian建议的那样,这个装饰器将阻塞,直到测试方法coroutine完成为止。这允许我编写这样的测试用例:</p>
<pre><code>class TestSocketConnection(unittest.TestCase):
def setUp(self):
self.mock_server = MockServer("localhost", 1337)
self.socket_connection = SocketConnection("localhost", 1337)
@async_test
def test_sends_handshake_after_connect(self):
yield from self.socket_connection.connect()
self.assertTrue(self.mock_server.received_handshake())
</code></pre>
<p>这个解决方案可能遗漏了一些边缘情况。</p>
<p>我认为这样的工具应该添加到Python的标准库中,使<code>asyncio</code>和<code>unittest</code>的交互更加方便。</p>