asyncioイベントループを使用してtornado.testing.AsyncTestCaseを実行します。

Question

私はユニットテストをしたいasyncioベースのクラスを持っています。 tornado.testing.AsyncTestCaseを使用すると、これは非常にうまく簡単に機能します。しかし、私のクラスの特定のメソッドの1つはasyncio.ensure_futureを使って別のメソッドの実行をスケジュールします。これはAsyncTestCaseで決して終了しません。デフォルトのテストランナーは、asyncioイベントループではなく、竜巻KQueueIOLoopイベントループを使用するためです。

class TestSubject: 
    def foo(self): 
     asyncio.ensure_future(self.bar()) 

    async def bar(self): 
     pass 

class TestSubjectTest(AsyncTestCase): 
    def test_foo(self): 
     t = TestSubject() 
     # here be somewhat involved setup with MagicMock and self.stop 
     t.foo() 
     self.wait() 

$ python -m tornado.testing baz.testsubject_test 
... 
[E 160627 17:48:22 testing:731] FAIL 
[E 160627 17:48:22 base_events:1090] Task was destroyed but it is pending! 
    task: <Task pending coro=<TestSubject.bar() running at ...>> 
.../asyncio/base_events.py:362: RuntimeWarning: coroutine 'TestSubject.bar' was never awaited 

どのように私は私のタスクが実際に実行されることを確認するためにテストを実行するためにさまざまなイベントループを使用することができますか？また、実装イベントをループに依存せず相互互換にするにはどうすればよいですか？

Answer 1

は十分に単純であることが判明...

class TestSubjectTest(AsyncTestCase): 
    def get_new_ioloop(self): # override this method 
     return tornado.platform.asyncio.AsyncIOMainLoop() 

私は前にこれをしようとしますが、直接動作しませんでしたasyncio.get_event_loop()を、返されました。 Tornadoのasyncioループラッパーを返すことは、そのトリックです。