diff --git a/test/test_oeffis.py b/test/test_oeffis.py index 1650b34..6eeefc5 100644 --- a/test/test_oeffis.py +++ b/test/test_oeffis.py @@ -31,10 +31,9 @@ # ```python # class TestOeffis(): # .... -# def test_foo(self): -# params = {} -# self.setup_daemon(params) +# def test_foo(self, daemon, mainloop): # # now you can talk to the RemoteDesktop portal +# ... # ``` # See the RemoteDesktop template for parameters that can be passed in. # @@ -47,10 +46,9 @@ from ctypes import c_char_p, c_int, c_uint32, c_void_p -from typing import Dict, List, Tuple, Type, Optional, TextIO +from typing import Iterator, List, Tuple, Type, Optional, TextIO from gi.repository import GLib # type: ignore from dbus.mainloop.glib import DBusGMainLoop -import unittest import attr import ctypes @@ -66,6 +64,12 @@ DBusGMainLoop(set_as_default=True) PREFIX = "oeffis_" +# Uncomment this to have dbus-monitor listen on the normal session address +# rather than the test DBus. This can be useful for cases where *something* +# messes up and tests run against the wrong bus. +# +# session_dbus_address = os.environ["DBUS_SESSION_BUS_ADDRESS"] + def version_at_least(have, required) -> bool: for h, r in zip(have.split("."), required.split(".")): @@ -269,14 +273,25 @@ def test_error_out(): assert oeffis.error_message is not None -# Uncomment this to have dbus-monitor listen on the normal session address -# rather than the test DBus. This can be useful for cases where *something* -# messes up and tests run against the wrong bus. -# -# session_dbus_address = os.environ["DBUS_SESSION_BUS_ADDRESS"] +@pytest.fixture() +def session_bus_unmonitored() -> Iterator[dbusmock.DBusTestCase]: + """ + Fixture that yields a newly created session bus + """ + bus = dbusmock.DBusTestCase() + bus.start_session_bus() + bus.setUp() + yield bus + bus.tearDown() + bus.tearDownClass() -def start_dbus_monitor() -> "subprocess.Process": +@pytest.fixture() +def session_bus(session_bus_unmonitored) -> Iterator[dbusmock.DBusTestCase]: + """ + Fixture that yields a newly created session bus + with dbus-monitor running on that bus (printing to stdout). + """ import subprocess env = os.environ.copy() @@ -294,99 +309,77 @@ def start_dbus_monitor() -> "subprocess.Process": mon.wait() GLib.timeout_add(2000, stop_dbus_monitor) - return mon + + yield session_bus_unmonitored + + mon.terminate() + mon.wait() -class TestOeffis(dbusmock.DBusTestCase): +@pytest.fixture() +def daemon( + session_bus, portal_name="RemoteDesktop", params=None, extra_templates=[] +) -> Iterator[subprocess.Popen]: + """ + Fixture that starts a DBusMock daemon in a separate process. + + If extra_templates is specified, it is a list of tuples with the + portal name as first value and the param dict to be passed to that + template as second value, e.g. ("ScreenCast", {...}). + """ + p_mock, obj_portal = session_bus.spawn_server_template( + template=f"templates/{portal_name.lower()}.py", + parameters=params or {}, + stdout=subprocess.PIPE, + ) + + flags = fcntl.fcntl(p_mock.stdout, fcntl.F_GETFL) + fcntl.fcntl(p_mock.stdout, fcntl.F_SETFL, flags | os.O_NONBLOCK) + + for t, tparams in extra_templates: + template = f"templates/{t.lower()}.py" + obj_portal.AddTemplate( + template, + dbus.Dictionary(tparams, signature="sv"), + dbus_interface=dbusmock.MOCK_IFACE, + ) + + yield p_mock + + if p_mock.stdout: + out = (p_mock.stdout.read() or b"").decode("utf-8") + if out: + print(out) + p_mock.stdout.close() + p_mock.terminate() + p_mock.wait() + + +@pytest.fixture() +def mainloop() -> Iterator[GLib.MainLoop]: + """ + Yields a mainloop that automatically quits after a + fixed timeout, but only on the first run. That's usually enough for + tests, if you need to call mainloop.run() repeatedly ensure that a + timeout handler is set to ensure quick test case failure in case of + error. + """ + + loop = GLib.MainLoop() + GLib.timeout_add(2000, loop.quit) + yield loop + + +@pytest.mark.skipif( + not version_at_least(dbusmock.__version__, "0.28.5"), + reason="dbusmock >= 0.28.5 required", +) +class TestOeffis: """ Test class that sets up a mocked DBus session bus to be used by liboeffis.so. """ - @unittest.skipIf( - not version_at_least(dbusmock.__version__, "0.28.5"), - "dbusmock >= 0.28.5 required", - ) - @classmethod - def setUpClass(cls): - cls.PORTAL_NAME = "RemoteDesktop" - cls.INTERFACE_NAME = f"org.freedesktop.portal.{cls.PORTAL_NAME}" - - def setUp(self): - self.p_mock = None - self._mainloop = None - self.dbus_monitor = None - - def setup_daemon(self, params=None, extra_templates: List[Tuple[str, Dict]] = []): - """ - Start a DBusMock daemon in a separate process. - - If extra_templates is specified, it is a list of tuples with the - portal name as first value and the param dict to be passed to that - template as second value, e.g. ("ScreenCast", {...}). - """ - self.start_session_bus() - self.p_mock, self.obj_portal = self.spawn_server_template( - template=f"templates/{self.PORTAL_NAME.lower()}.py", - parameters=params or {}, - stdout=subprocess.PIPE, - ) - - flags = fcntl.fcntl(self.p_mock.stdout, fcntl.F_GETFL) - fcntl.fcntl(self.p_mock.stdout, fcntl.F_SETFL, flags | os.O_NONBLOCK) - self.mock_interface = dbus.Interface(self.obj_portal, dbusmock.MOCK_IFACE) - self.properties_interface = dbus.Interface( - self.obj_portal, dbus.PROPERTIES_IFACE - ) - self.portal_interface = dbus.Interface(self.obj_portal, self.INTERFACE_NAME) - - for t, tparams in extra_templates: - template = f"templates/{t.lower()}.py" - self.obj_portal.AddTemplate( - template, - dbus.Dictionary(tparams, signature="sv"), - dbus_interface=dbusmock.MOCK_IFACE, - ) - - self.dbus_monitor = start_dbus_monitor() - self._mainloop = None - - def tearDown(self): - if self.p_mock: - if self.p_mock.stdout: - out = (self.p_mock.stdout.read() or b"").decode("utf-8") - if out: - print(out) - self.p_mock.stdout.close() - self.p_mock.terminate() - self.p_mock.wait() - - if self.dbus_monitor: - self.dbus_monitor.terminate() - self.dbus_monitor.wait() - - @property - def mainloop(self): - """ - The mainloop for this test. This mainloop automatically quits after a - fixed timeout, but only on the first run. That's usually enough for - tests, if you need to call mainloop.run() repeatedly ensure that a - timeout handler is set to ensure quick test case failure in case of - error. - """ - if self._mainloop is None: - - def quit(): - self._mainloop.quit() # type: ignore - self._mainloop = None - - self._mainloop = GLib.MainLoop() - GLib.timeout_add(2000, quit) - - return self._mainloop - - def test_create_session(self): - self.setup_daemon() - + def test_create_session(self, daemon, mainloop): oeffis = Oeffis() oeffis.create_session(oeffis.DEVICE_POINTER | oeffis.DEVICE_KEYBOARD) # type: ignore oeffis.dispatch() # type: ignore @@ -397,7 +390,7 @@ class TestOeffis(dbusmock.DBusTestCase): GLib.io_add_watch(oeffis.fd, 0, GLib.IO_IN, _dispatch) - self.mainloop.run() + mainloop.run() e = oeffis.get_event() # type: ignore assert e == oeffis.EVENT_CONNECTED_TO_EIS, oeffis.error_message # type: ignore