From 66e478d968a96c255e9fff1732e7fe79a9a07145 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marek=20Marczykowski-G=C3=B3recki?= Date: Thu, 25 Feb 2021 02:28:38 +0100 Subject: [PATCH] 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. --- qubesmanager/tests/test_qube_manager.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/qubesmanager/tests/test_qube_manager.py b/qubesmanager/tests/test_qube_manager.py index ab2e907..7a7de77 100644 --- a/qubesmanager/tests/test_qube_manager.py +++ b/qubesmanager/tests/test_qube_manager.py @@ -21,6 +21,7 @@ # import asyncio import contextlib +import functools import logging.handlers import unittest import unittest.mock @@ -41,6 +42,24 @@ from qubesmanager.tests import init_qtapp 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): def setUp(self): super(QubeManagerTest, self).setUp()