tests: add a decorator for keeping event listener alive
This is useful for tests causing events in qube manager itsef (not just external qvm-prefs calls). If event listener isn't connected at this time, qube manager won't receive them and also qubesadmin internal cache won't be refreshed.
This commit is contained in:
parent
1e36eb9d30
commit
66e478d968
@ -21,6 +21,7 @@
|
|||||||
#
|
#
|
||||||
import asyncio
|
import asyncio
|
||||||
import contextlib
|
import contextlib
|
||||||
|
import functools
|
||||||
import logging.handlers
|
import logging.handlers
|
||||||
import unittest
|
import unittest
|
||||||
import unittest.mock
|
import unittest.mock
|
||||||
@ -41,6 +42,24 @@ from qubesmanager.tests import init_qtapp
|
|||||||
icon_size = qube_manager.icon_size
|
icon_size = qube_manager.icon_size
|
||||||
|
|
||||||
|
|
||||||
|
def listen_for_events(func):
|
||||||
|
"""Wrapper for a test that needs events listener to be registered all the time.
|
||||||
|
Note the test still needs to yield to the event loop to actually handle events.
|
||||||
|
"""
|
||||||
|
@functools.wraps(func)
|
||||||
|
def wrapper(self, *args, **kwargs):
|
||||||
|
events_listener = \
|
||||||
|
asyncio.ensure_future(self.dispatcher.listen_for_events())
|
||||||
|
# let it connect (run until first yield/await)
|
||||||
|
self.loop.run_until_complete(asyncio.sleep(0))
|
||||||
|
try:
|
||||||
|
return func(self, *args, **kwargs)
|
||||||
|
finally:
|
||||||
|
events_listener.cancel()
|
||||||
|
self.loop.call_soon(self.loop.stop)
|
||||||
|
self.loop.run_forever()
|
||||||
|
return wrapper
|
||||||
|
|
||||||
class QubeManagerTest(unittest.TestCase):
|
class QubeManagerTest(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
super(QubeManagerTest, self).setUp()
|
super(QubeManagerTest, self).setUp()
|
||||||
|
Loading…
Reference in New Issue
Block a user