test: rework the oeffis dbus tests to be pytest-compatible

DBusMock is unittest based and the documentation points users to that
approach. That approach is limiting however because we can't use all
pytest features (see [1]). Luckily, the parent class in dbusmock doesn't
really do much so we can emulate the functionality ourselves - all we
need to do is call the same setUp/tearDowns and be done with it.

This means we can move the dbus-monitor and mainloop handling into
fixtures too which makes the code a fair bit nicer to read.

[1] https://docs.pytest.org/en/7.1.x/how-to/unittest.html#pytest-features-in-unittest-testcase-subclasses
This commit is contained in:
Peter Hutterer 2023-07-26 16:14:12 +03:00
parent e03c047b5d
commit 7115e9c4c8

View file

@ -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