Run tornado.testing.AsyncTestCase mit asyncio Ereignisschleife

Question

Ich habe eine asyncio-basierte Klasse, die ich Unit-Test wollen. Mit tornado.testing.AsyncTestCase funktioniert das ganz gut und einfach. Eine bestimmte Methode meiner Klasse verwendet jedoch asyncio.ensure_future, um die Ausführung einer anderen Methode zu planen. Dies endet nie in der AsyncTestCase, da der Standard-Test-Runner die Tornado KQueueIOLoop Ereignisschleife verwendet, keine asyncio Ereignisschleife.

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 

Wie kann ich eine andere Ereignisschleife verwenden, um die Tests an, um sicherzustellen, meine Aufgabe wird es tatsächlich ausgeführt laufen? Oder wie kann ich mein Implementierungsereignis loop-unabhängig und kompatibel machen?

Answer 1

stellte sich heraus, einfach genug zu sein ...

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

ich das schon versuchte, kehrte aber direkt asyncio.get_event_loop(), was nicht funktioniert hat. Die Rückkehr von Tornados asyncio loop wrapper macht den Trick.