mirror of
https://github.com/brl/mutter.git
synced 2024-11-23 08:30:42 -05:00
e21f839776
The test and dist CI jobs run wrap the meson calls in dbus-runner to avoid setting up dbus servers and mocking services for every test but the dbus-runner invocation from meson test didn't actually skip all the setup. This nested mocking also doesn't work because the system bus is assumed to be the host system bus and not a mocked one. Part-of: <https://gitlab.gnome.org/GNOME/mutter/-/merge_requests/2618>
315 lines
11 KiB
Python
315 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import dbus
|
|
import sys
|
|
import os
|
|
import fcntl
|
|
import subprocess
|
|
import getpass
|
|
import argparse
|
|
import logind_helpers
|
|
from collections import OrderedDict
|
|
from dbusmock import DBusTestCase
|
|
from dbus.mainloop.glib import DBusGMainLoop
|
|
from pathlib import Path
|
|
from gi.repository import Gio
|
|
|
|
|
|
def escape_object_path(path):
|
|
b = bytearray()
|
|
b.extend(path.encode())
|
|
path = Gio.dbus_escape_object_path_bytestring(b)
|
|
if path[0].isdigit():
|
|
path = "_{0:02x}{1}".format(ord(path[0]), path[1:])
|
|
return os.path.basename(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, launch):
|
|
klass.templates_dirs = [klass.__get_templates_dir()]
|
|
|
|
klass.mocks = OrderedDict()
|
|
|
|
klass.host_system_bus_address = os.getenv('DBUS_SYSTEM_BUS_ADDRESS')
|
|
if klass.host_system_bus_address is None:
|
|
klass.host_system_bus_address = 'unix:path=/run/dbus/system_bus_socket'
|
|
|
|
try:
|
|
dbus.bus.BusConnection(klass.host_system_bus_address)
|
|
klass.has_host_system_bus = True
|
|
except:
|
|
klass.has_host_system_bus = False
|
|
|
|
print('Starting D-Bus daemons (session & system)...', file=sys.stderr)
|
|
DBusTestCase.setUpClass()
|
|
klass.start_session_bus()
|
|
klass.start_system_bus()
|
|
|
|
print('Launching required services...', file=sys.stderr)
|
|
klass.service_processes = []
|
|
for service in launch:
|
|
klass.service_processes += [subprocess.Popen(service)]
|
|
|
|
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.start_from_local_template('rtkit')
|
|
klass.start_from_local_template('screensaver')
|
|
|
|
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()
|
|
|
|
for process in klass.service_processes:
|
|
process.terminate()
|
|
process.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
|
|
|
|
def wrap_logind_call(call):
|
|
code = \
|
|
f'''
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, '{os.path.dirname(__file__)}')
|
|
import logind_helpers
|
|
|
|
{call}
|
|
'''
|
|
return code
|
|
|
|
@classmethod
|
|
def forward_to_host(klass, object_path, interface, method, in_type, out_type):
|
|
proxy = klass.system_bus_con.get_object('org.freedesktop.login1',
|
|
object_path)
|
|
proxy.AddMethod(interface, method, in_type, out_type,
|
|
f'''
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, '{os.path.dirname(__file__)}')
|
|
import logind_helpers
|
|
|
|
ret = logind_helpers.call_host('{klass.host_system_bus_address}',
|
|
'{object_path}',
|
|
'{interface}',
|
|
'{method}',
|
|
'{in_type}',
|
|
args)
|
|
''')
|
|
|
|
@classmethod
|
|
def init_logind_forward(klass, session_path, seat_path):
|
|
klass.forward_to_host(session_path, 'org.freedesktop.login1.Session',
|
|
'TakeDevice',
|
|
'uu', 'hb')
|
|
klass.forward_to_host(session_path, 'org.freedesktop.login1.Session',
|
|
'ReleaseDevice',
|
|
'uu', '')
|
|
klass.forward_to_host(session_path, 'org.freedesktop.login1.Session',
|
|
'TakeDevice',
|
|
'uu', 'hb')
|
|
klass.forward_to_host(session_path, 'org.freedesktop.login1.Session',
|
|
'TakeControl',
|
|
'b', '')
|
|
klass.forward_to_host(seat_path, 'org.freedesktop.login1.Seat',
|
|
'SwitchTo',
|
|
'u', '')
|
|
|
|
@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',
|
|
klass.wrap_logind_call(
|
|
f'''
|
|
major = args[0]
|
|
minor = args[1]
|
|
ret = logind_helpers.open_file_direct(major, minor)
|
|
'''))
|
|
session_obj.AddMethods('org.freedesktop.login1.Session', [
|
|
('ReleaseDevice', 'uu', '', ''),
|
|
('TakeControl', 'b', '', ''),
|
|
])
|
|
|
|
|
|
@classmethod
|
|
def find_host_session_name(klass):
|
|
if 'XDG_SESSION_ID' in os.environ:
|
|
return escape_object_path(os.environ['XDG_SESSION_ID'])
|
|
|
|
bus = dbus.bus.BusConnection(klass.host_system_bus_address)
|
|
session_auto_proxy = bus.get_object('org.freedesktop.login1',
|
|
'/org/freedesktop/login1/session/auto')
|
|
props = dbus.Interface(session_auto_proxy,
|
|
dbus_interface='org.freedesktop.DBus.Properties')
|
|
session_id = props.Get('org.freedesktop.login1.Session', 'Id')
|
|
manager_proxy = bus.get_object('org.freedesktop.login1',
|
|
'/org/freedesktop/login1')
|
|
manager = dbus.Interface(manager_proxy,
|
|
dbus_interface='org.freedesktop.login1.Manager')
|
|
session_path = manager.GetSession(session_id)
|
|
return os.path.basename(session_path)
|
|
|
|
@classmethod
|
|
def init_logind(klass, enable_kvm):
|
|
logind = klass.start_from_template('logind')
|
|
|
|
[p_mock, obj] = logind
|
|
|
|
mock_iface = 'org.freedesktop.DBus.Mock'
|
|
seat_path = obj.AddSeat('seat0', dbus_interface=mock_iface)
|
|
session_name = 'dummy'
|
|
if klass.has_host_system_bus:
|
|
session_name = klass.find_host_session_name()
|
|
|
|
session_path = obj.AddSession(session_name, 'seat0',
|
|
dbus.types.UInt32(os.getuid()),
|
|
getpass.getuser(),
|
|
True,
|
|
dbus_interface=mock_iface)
|
|
|
|
if enable_kvm:
|
|
klass.init_logind_kvm(session_path)
|
|
elif klass.has_host_system_bus:
|
|
klass.init_logind_forward(session_path, seat_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(args, wrapper):
|
|
env = {}
|
|
env.update(os.environ)
|
|
env['NO_AT_BRIDGE'] = '1'
|
|
env['GTK_A11Y'] = 'none'
|
|
env['GSETTINGS_BACKEND'] = 'memory'
|
|
env['META_DBUS_RUNNER_ACTIVE'] = '1'
|
|
|
|
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)
|
|
parser.add_argument('--launch', action='append', default=[])
|
|
(args, rest) = parser.parse_known_args(sys.argv)
|
|
|
|
rest.pop(0)
|
|
if rest[0] == '--':
|
|
rest.pop(0)
|
|
|
|
result = 1
|
|
|
|
if os.getenv('META_DBUS_RUNNER_ACTIVE') == None:
|
|
klass.setUpClass(args.kvm, args.launch)
|
|
runner = klass()
|
|
runner.assertGreater(len(rest), 0)
|
|
wrapper = os.getenv('META_DBUS_RUNNER_WRAPPER')
|
|
|
|
try:
|
|
print('Running test case...', file=sys.stderr)
|
|
result = wrap_call(rest, wrapper)
|
|
finally:
|
|
MutterDBusRunner.tearDownClass()
|
|
else:
|
|
try:
|
|
print(('Inside a nested meta-dbus-runner: '
|
|
'Not re-creating mocked environment.'),
|
|
file=sys.stderr)
|
|
print('Running test case...', file=sys.stderr)
|
|
result = wrap_call(rest, None)
|
|
finally:
|
|
pass
|
|
|
|
return result
|