tests: Extract D-Bus runner as reusable python module

The D-Bus runner used by tests, including installed tests, is made to be
reusable from GNOME Shell. To do this, install it and the templates in
the pkgdatadir (e.g. /usr/share/mutter-APIVERSION/tests/), generate a
custom runner for the installed tests that uses the installed script and
templates, and change the non-installed original runner to use the
non-installed templates.

The end goal is to reuse the D-Bus session runner and templates used for
mutter when test running GNOME Shell.

Part-of: <https://gitlab.gnome.org/GNOME/mutter/-/merge_requests/1354>

src/tests/meson.build

@@ -68,6 +68,7 @@ pkg.generate(libmutter_test,
   version: meson.project_version(),
   variables: [
     'apiversion=' + libmutter_api_version,
+    'tests_datadir=${prefix}/share/mutter-' + libmutter_api_version + '/tests',
   ],
   install_dir: pcdir,
 )
@@ -148,11 +149,22 @@ test_runner = executable('mutter-test-runner',
 
 meta_dbus_runner = find_program('meta-dbus-runner.py')
 
-if have_installed_tests
+tests_datadir = join_paths(pkgdatadir, 'tests')
-  install_data('meta-dbus-runner.py',
+
-    install_dir: mutter_installed_tests_libexecdir,
+install_data('mutter_dbusrunner.py',
+  install_dir: tests_datadir,
 )
 install_subdir('dbusmock-templates',
+  install_dir: tests_datadir,
+)
+
+if have_installed_tests
+  configure_file(
+    input: 'mutter-installed-dbus-session.py.in',
+    output: 'mutter-installed-dbus-session.py',
+    configuration: {'tests_datadir': tests_datadir},
+    install: true,
+    install_mode: 'rwxr-xr-x',
     install_dir: mutter_installed_tests_libexecdir,
   )
 endif

src/tests/meta-dbus-runner.py

@@ -1,194 +1,9 @@
 #!/usr/bin/env python3
 
-import dbus
 import sys
-import os
-import fcntl
-import subprocess
-import getpass
-import argparse
-from collections import OrderedDict
-from dbusmock import DBusTestCase
-from dbus.mainloop.glib import DBusGMainLoop
-
-DBusGMainLoop(set_as_default=True)
-
-
-def get_templates_dir():
-    return os.path.join(os.path.dirname(__file__), 'dbusmock-templates')
-
-def get_template_path(template_name):
-    return os.path.join(get_templates_dir(), template_name + '.py')
-
-def get_subprocess_stdout():
-    if os.getenv('META_DBUS_RUNNER_VERBOSE') == '1':
-        return sys.stderr
-    else:
-        return subprocess.DEVNULL;
-
-
-class MutterDBusTestCase(DBusTestCase):
-    @classmethod
-    def setUpClass(klass, enable_kvm):
-        klass.mocks = OrderedDict()
-
-        print('Starting D-Bus daemons (session & system)...', file=sys.stderr)
-        DBusTestCase.setUpClass()
-        klass.start_session_bus()
-        klass.start_system_bus()
-
-        print('Starting mocked services...', file=sys.stderr)
-        (klass.mocks_manager, klass.mock_obj) = klass.start_from_local_template(
-            'meta-mocks-manager', {'templates-dir': get_templates_dir()})
-
-        klass.start_from_local_template('localed')
-        klass.start_from_local_template('colord')
-        klass.start_from_local_template('gsd-color')
-
-        klass.system_bus_con = klass.get_dbus(system_bus=True)
-        klass.session_bus_con = klass.get_dbus(system_bus=False)
-
-        klass.init_logind(enable_kvm)
-
-        if klass.session_bus_con.name_has_owner('org.gnome.Mutter.DisplayConfig'):
-            raise Exception(
-                'org.gnome.Mutter.DisplayConfig already has owner on the session bus, bailing')
-
-    @classmethod
-    def tearDownClass(klass):
-        klass.mock_obj.Cleanup()
-
-        for (mock_server, mock_obj) in reversed(klass.mocks.values()):
-            mock_server.terminate()
-            mock_server.wait()
-
-        DBusTestCase.tearDownClass()
-
-    @classmethod
-    def start_from_template(klass, template, params={}):
-        mock_server, mock_obj = \
-            klass.spawn_server_template(template,
-                                        params,
-                                        get_subprocess_stdout())
-
-        mocks = (mock_server, mock_obj)
-        assert klass.mocks.setdefault(template, mocks) == mocks
-        return mocks
-
-    @classmethod
-    def start_from_local_template(klass, template_file_name, params={}):
-        template = get_template_path(template_file_name)
-        return klass.start_from_template(template, params)
-
-    @classmethod
-    def start_from_template_managed(klass, template):
-        klass.mock_obj.StartFromTemplate(template)
-
-    @classmethod
-    def start_from_local_template_managed(klass, template_file_name):
-        template = get_template_path(template_file_name)
-        klass.mock_obj.StartFromLocalTemplate(template)
-
-    @classmethod
-    def start_from_class(klass, mock_class, params={}):
-        mock_server = \
-            klass.spawn_server(mock_class.BUS_NAME,
-                               mock_class.MAIN_OBJ,
-                               mock_class.MAIN_IFACE,
-                               mock_class.SYSTEM_BUS,
-                               stdout=get_subprocess_stdout())
-
-        bus = klass.get_dbus(system_bus=mock_class.SYSTEM_BUS)
-        mock_obj = bus.get_object(mock_class.BUS_NAME, mock_class.MAIN_OBJ)
-        mock_class.load(mock_obj, params)
-
-        mocks = (mock_server, mock_obj)
-        assert klass.mocks.setdefault(mock_class, mocks) == mocks
-        return mocks
-
-    @classmethod
-    def init_logind_kvm(klass, session_path):
-        session_obj = klass.system_bus_con.get_object('org.freedesktop.login1', session_path)
-        session_obj.AddMethod('org.freedesktop.login1.Session',
-                              'TakeDevice',
-                              'uu', 'hb',
-'''
-import re
-
-major = args[0]
-minor = args[1]
-
-sysfs_uevent_path = '/sys/dev/char/{}:{}/uevent'.format(major, minor)
-sysfs_uevent = open(sysfs_uevent_path, 'r')
-devname = None
-for line in sysfs_uevent.readlines():
-    match = re.match('DEVNAME=(.*)', line)
-    if match:
-        devname = match[1]
-        break
-sysfs_uevent.close()
-if not devname:
-    raise dbus.exceptions.DBusException(f'Device file {major}:{minor} doesn\\\'t exist',
-                                        major=major, minor=minor)
-fd = os.open('/dev/' + devname, os.O_RDWR | os.O_CLOEXEC)
-unix_fd = dbus.types.UnixFd(fd)
-os.close(fd)
-ret = (unix_fd, False)
-''')
-        session_obj.AddMethods('org.freedesktop.login1.Session', [
-            ('ReleaseDevice', 'uu', '', ''),
-            ('TakeControl', 'b', '', ''),
-        ])
-
-    @classmethod
-    def init_logind(klass, enable_kvm):
-        logind = klass.start_from_template('logind')
-
-        [p_mock, obj] = logind
-
-        mock_iface = 'org.freedesktop.DBus.Mock'
-        obj.AddSeat('seat0', dbus_interface=mock_iface)
-        session_path = obj.AddSession('dummy', 'seat0',
-                                      dbus.types.UInt32(os.getuid()),
-                                      getpass.getuser(),
-                                      True,
-                                      dbus_interface=mock_iface)
-
-        if enable_kvm:
-            klass.init_logind_kvm(session_path)
-
-    def wrap_call(self, args):
-        env = {}
-        env.update(os.environ)
-        env['NO_AT_BRIDGE'] = '1'
-        env['GSETTINGS_BACKEND'] = 'memory'
-
-        wrapper = env.get('META_DBUS_RUNNER_WRAPPER')
-        if wrapper == 'gdb':
-            args = ['gdb', '-ex', 'r', '-ex', 'bt full', '--args'] + args
-        elif wrapper:
-            args = wrapper.split(' ') + args
-
-        p = subprocess.Popen(args, env=env)
-        return p.wait()
 
+from mutter_dbusrunner import MutterDBusRunner, meta_run
 
 if __name__ == '__main__':
-    parser = argparse.ArgumentParser()
+    result = meta_run(MutterDBusRunner)
-    parser.add_argument('--kvm', action='store_true', default=False)
-    (args, rest) = parser.parse_known_args(sys.argv)
-
-    rest.pop(0)
-    if rest[0] == '--':
-      rest.pop(0)
-
-    MutterDBusTestCase.setUpClass(args.kvm)
-    test_case = MutterDBusTestCase()
-    test_case.assertGreater(len(rest), 0)
-    result = 1
-    try:
-        print('Running test case...', file=sys.stderr)
-        result = test_case.wrap_call(rest)
-    finally:
-        MutterDBusTestCase.tearDownClass()
     sys.exit(result)

src/tests/mutter-all.test.in

@@ -4,6 +4,6 @@ Description=All Mutter tests
 # a solution for
 # https://gitlab.gnome.org/GNOME/gnome-desktop-testing/-/issues/1,
 # and anyway that wouldn't be sufficient to handle XDG_RUNTIME_DIR
-Exec=sh -ec 'env GSETTINGS_BACKEND=memory XDG_RUNTIME_DIR="$(mktemp -d -t mutter-@apiversion@-all-tests-XXXXXX)" @libexecdir@/installed-tests/mutter-@apiversion@/meta-dbus-runner.py xvfb-run -a -s "+iglx -noreset" -- @libexecdir@/installed-tests/mutter-@apiversion@/mutter-test-runner --all'
+Exec=sh -ec 'env GSETTINGS_BACKEND=memory XDG_RUNTIME_DIR="$(mktemp -d -t mutter-@apiversion@-all-tests-XXXXXX)" @libexecdir@/installed-tests/mutter-@apiversion@/mutter-installed-dbus-session.py xvfb-run -a -s "+iglx -noreset" -- @libexecdir@/installed-tests/mutter-@apiversion@/mutter-test-runner --all'
 Type=session
 Output=TAP

src/tests/mutter-installed-dbus-session.py.in

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+
+import sys
+
+sys.path.insert(1, '@tests_datadir@')
+
+from mutter_dbusrunner import MutterDBusRunner, meta_run
+
+if __name__ == '__main__':
+    result = meta_run(MutterDBusRunner)
+    sys.exit(result)

src/tests/mutter_dbusrunner.py

@@ -0,0 +1,208 @@
+#!/usr/bin/env python3
+
+import dbus
+import sys
+import os
+import fcntl
+import subprocess
+import getpass
+import argparse
+from collections import OrderedDict
+from dbusmock import DBusTestCase
+from dbus.mainloop.glib import DBusGMainLoop
+from pathlib import Path
+
+
+def get_subprocess_stdout():
+    if os.getenv('META_DBUS_RUNNER_VERBOSE') == '1':
+        return sys.stderr
+    else:
+        return subprocess.DEVNULL;
+
+
+class MutterDBusRunner(DBusTestCase):
+    @classmethod
+    def __get_templates_dir(klass):
+            return os.path.join(os.path.dirname(__file__), 'dbusmock-templates')
+
+    @classmethod
+    def setUpClass(klass, enable_kvm):
+        klass.templates_dirs = [klass.__get_templates_dir()]
+
+        klass.mocks = OrderedDict()
+
+        print('Starting D-Bus daemons (session & system)...', file=sys.stderr)
+        DBusTestCase.setUpClass()
+        klass.start_session_bus()
+        klass.start_system_bus()
+
+        print('Starting mocked services...', file=sys.stderr)
+        (klass.mocks_manager, klass.mock_obj) = klass.start_from_local_template(
+            'meta-mocks-manager', {'templates-dir': klass.__get_templates_dir()})
+
+        klass.start_from_local_template('localed')
+        klass.start_from_local_template('colord')
+        klass.start_from_local_template('gsd-color')
+
+        klass.system_bus_con = klass.get_dbus(system_bus=True)
+        klass.session_bus_con = klass.get_dbus(system_bus=False)
+
+        klass.init_logind(enable_kvm)
+
+        if klass.session_bus_con.name_has_owner('org.gnome.Mutter.DisplayConfig'):
+            raise Exception(
+                'org.gnome.Mutter.DisplayConfig already has owner on the session bus, bailing')
+
+    @classmethod
+    def tearDownClass(klass):
+        klass.mock_obj.Cleanup()
+
+        for (mock_server, mock_obj) in reversed(klass.mocks.values()):
+            mock_server.terminate()
+            mock_server.wait()
+
+        DBusTestCase.tearDownClass()
+
+    @classmethod
+    def start_from_template(klass, template, params={}, system_bus=None):
+        mock_server, mock_obj = \
+            klass.spawn_server_template(template,
+                                        params,
+                                        get_subprocess_stdout(),
+                                        system_bus=system_bus)
+
+        mocks = (mock_server, mock_obj)
+        return mocks
+
+    @classmethod
+    def start_from_local_template(klass, template_file_name, params={}, system_bus=None):
+        template = klass.find_template(template_file_name)
+        return klass.start_from_template(template, params, system_bus=system_bus)
+
+    @classmethod
+    def start_from_template_managed(klass, template):
+        klass.mock_obj.StartFromTemplate(template)
+
+    @classmethod
+    def start_from_local_template_managed(klass, template_file_name):
+        template = klass.find_template(template_file_name)
+        klass.mock_obj.StartFromLocalTemplate(template)
+
+    @classmethod
+    def start_from_class(klass, mock_class, params={}):
+        mock_server = \
+            klass.spawn_server(mock_class.BUS_NAME,
+                               mock_class.MAIN_OBJ,
+                               mock_class.MAIN_IFACE,
+                               mock_class.SYSTEM_BUS,
+                               stdout=get_subprocess_stdout())
+
+        bus = klass.get_dbus(system_bus=mock_class.SYSTEM_BUS)
+        mock_obj = bus.get_object(mock_class.BUS_NAME, mock_class.MAIN_OBJ)
+        mock_class.load(mock_obj, params)
+
+        mocks = (mock_server, mock_obj)
+        return mocks
+
+    @classmethod
+    def init_logind_kvm(klass, session_path):
+        session_obj = klass.system_bus_con.get_object('org.freedesktop.login1', session_path)
+        session_obj.AddMethod('org.freedesktop.login1.Session',
+                              'TakeDevice',
+                              'uu', 'hb',
+'''
+import re
+
+major = args[0]
+minor = args[1]
+
+sysfs_uevent_path = '/sys/dev/char/{}:{}/uevent'.format(major, minor)
+sysfs_uevent = open(sysfs_uevent_path, 'r')
+devname = None
+for line in sysfs_uevent.readlines():
+    match = re.match('DEVNAME=(.*)', line)
+    if match:
+        devname = match[1]
+        break
+sysfs_uevent.close()
+if not devname:
+    raise dbus.exceptions.DBusException(f'Device file {major}:{minor} doesn\\\'t exist',
+                                        major=major, minor=minor)
+fd = os.open('/dev/' + devname, os.O_RDWR | os.O_CLOEXEC)
+unix_fd = dbus.types.UnixFd(fd)
+os.close(fd)
+ret = (unix_fd, False)
+''')
+        session_obj.AddMethods('org.freedesktop.login1.Session', [
+            ('ReleaseDevice', 'uu', '', ''),
+            ('TakeControl', 'b', '', ''),
+        ])
+
+    @classmethod
+    def init_logind(klass, enable_kvm):
+        logind = klass.start_from_template('logind')
+
+        [p_mock, obj] = logind
+
+        mock_iface = 'org.freedesktop.DBus.Mock'
+        obj.AddSeat('seat0', dbus_interface=mock_iface)
+        session_path = obj.AddSession('dummy', 'seat0',
+                                      dbus.types.UInt32(os.getuid()),
+                                      getpass.getuser(),
+                                      True,
+                                      dbus_interface=mock_iface)
+
+        if enable_kvm:
+            klass.init_logind_kvm(session_path)
+
+    @classmethod
+    def add_template_dir(klass, templates_dir):
+        klass.templates_dirs += [templates_dir]
+
+    @classmethod
+    def find_template(klass, template_name):
+        for templates_dir in klass.templates_dirs:
+            template_path = os.path.join(templates_dir, template_name + '.py')
+            template_file = Path(template_path)
+            if template_file.is_file():
+                return template_path
+        raise FileNotFoundError(f'Couldnt find a {template_name} template')
+
+    def wrap_call(self, args):
+        env = {}
+        env.update(os.environ)
+        env['NO_AT_BRIDGE'] = '1'
+        env['GSETTINGS_BACKEND'] = 'memory'
+
+        wrapper = env.get('META_DBUS_RUNNER_WRAPPER')
+        if wrapper == 'gdb':
+            args = ['gdb', '-ex', 'r', '-ex', 'bt full', '--args'] + args
+        elif wrapper:
+            args = wrapper.split(' ') + args
+
+        p = subprocess.Popen(args, env=env)
+        return p.wait()
+
+
+def meta_run(klass):
+    DBusGMainLoop(set_as_default=True)
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--kvm', action='store_true', default=False)
+    (args, rest) = parser.parse_known_args(sys.argv)
+
+    rest.pop(0)
+    if rest[0] == '--':
+      rest.pop(0)
+
+    klass.setUpClass(args.kvm)
+    runner = klass()
+    runner.assertGreater(len(rest), 0)
+    result = 1
+
+    try:
+        print('Running test case...', file=sys.stderr)
+        result = runner.wrap_call(rest)
+    finally:
+        MutterDBusRunner.tearDownClass()
+    return result