1
0
forked from drath/libratbag

tools: share the devel boiler plate in a separate python file

We want to share as much as possible between ratbagctl.devel and
ratbagctl.test, so better have a separate file that knows how to
properly spawn the dbus daemon and terminate it.

Signed-off-by: Benjamin Tissoires <benjamin.tissoires@gmail.com>
This commit is contained in:
Benjamin Tissoires
2017-07-31 10:42:13 +01:00
committed by Peter Hutterer
parent 282398493b
commit 6cfca9e34b
5 changed files with 145 additions and 86 deletions

View File

@@ -519,6 +519,10 @@ configure_file(input : 'tools/ratbagctl.test.in',
ratbagctl_test = find_program(join_paths(meson.build_root(), 'ratbagctl.test'))
test('ratbagctl-test', ratbagctl_test, args: ['-v'])
configure_file(input : 'tools/toolbox.py',
output : 'toolbox.py',
configuration : config_ratbagctl_devel)
#### output files ####
configure_file(output: 'config.h', install: false, configuration: config_h)

View File

@@ -26,27 +26,25 @@
# DEALINGS IN THE SOFTWARE.
import os
import shutil
import toolbox
import subprocess
import sys
if not os.geteuid() == 0:
sys.exit('Script must be run as root')
SCRIPT_DIR = '@MESON_BUILD_ROOT@'
def main(argv):
if not os.geteuid() == 0:
sys.exit('Script must be run as root')
# first copy the policy for the ratbagd daemon to be allowed to run
shutil.copy(os.path.join(SCRIPT_DIR, 'org.freedesktop.ratbag_devel1.conf'),
'/etc/dbus-1/system.d/')
ratbagd = toolbox.start_ratbagd()
ratbagd = subprocess.Popen([os.path.join(SCRIPT_DIR, "ratbagd.devel")], shell=True)
try:
ratbagctl = subprocess.Popen(" ".join([toolbox.RATBAGCTL_PATH, *sys.argv[1:]]))
try:
ratbagctl = subprocess.Popen(" ".join([os.path.join(SCRIPT_DIR, "ratbagctl"), *sys.argv[1:]]),
shell=True,
env={"RATBAGCTL_DEVEL": "org.freedesktop.ratbag_devel1_@ratbagd_sha@"})
ratbagctl.wait()
except KeyboardInterrupt:
pass
toolbox.terminate_ratbagd(ratbagd)
ratbagctl.wait()
except KeyboardInterrupt:
pass
ratbagd.terminate()
if __name__ == "__main__":
main(sys.argv[1:])

View File

@@ -1318,7 +1318,7 @@ def get_parser():
return RatbagParserRoot(parser_def)
def open_ratbagd():
def open_ratbagd(ratbagd_process=None):
try:
r = Ratbagd()
except RatbagdDBusUnavailable:
@@ -1329,6 +1329,14 @@ def open_ratbagd():
print("No devices available.")
return r
if ratbagd_process is not None:
# if some old version of ratbagd is still running, ratbagd_process may
# have never started but our DBus bindings may succeed. Check for the
# return code here, this also gives ratbagd enough time to start and
# die. If we check immediately we may not have terminated yet.
ratbagd_process.poll()
assert ratbagd_process.returncode is None
def main(argv):
if not argv:

View File

@@ -26,43 +26,17 @@
# DEALINGS IN THE SOFTWARE.
import argparse
import imp
import io
import os
import resource
import shutil
import subprocess
import sys
import toolbox
import unittest
# various constants
RATBAGCTL_NAME = 'ratbagctl'
RATBAGCTL_PATH = os.path.join('@MESON_BUILD_ROOT@', RATBAGCTL_NAME)
DBUS_CONF_DIR = '/etc/dbus-1/system.d'
DBUS_CONF_NAME = 'org.freedesktop.ratbag_devel1.conf'
DBUS_CONF_PATH = os.path.join(DBUS_CONF_DIR, DBUS_CONF_NAME)
run_ratbagctl_in_subprocess = False
def import_non_standard_path(name, path):
# Fast path: see if the module has already been imported.
try:
return sys.modules[name]
except KeyError:
pass
# If any of the following calls raises an exception,
# there's a problem we can't handle -- let the caller handle it.
with open(path, 'rb') as fp:
module = imp.load_module(name, fp, os.path.basename(path), ('.py', 'rb', imp.PY_SOURCE))
return module
ratbagctl = import_non_standard_path(RATBAGCTL_NAME, RATBAGCTL_PATH)
class TestRatbagCtl(unittest.TestCase):
@classmethod
def setUpClass(cls):
@@ -71,7 +45,7 @@ class TestRatbagCtl(unittest.TestCase):
@classmethod
def run_ratbagctl_subprocess(cls, params):
ratbagctl = subprocess.Popen(" ".join([RATBAGCTL_PATH, params]),
ratbagctl = subprocess.Popen(" ".join([toolbox.RATBAGCTL_PATH, params]),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
shell=True,
@@ -547,56 +521,19 @@ class TestRatbgagCtlLED(TestRatbagCtl):
def setUpModule():
from gi.repository import Gio
import time
global ratbagd_process, ratbagd, parser
# first copy the policy for the ratbagd daemon to be allowed to run
shutil.copy(os.path.join('@MESON_BUILD_ROOT@', DBUS_CONF_NAME), DBUS_CONF_PATH)
# FIXME: kill any running ratbagd.devel
ratbagd_process = toolbox.start_ratbagd()
ratbagd_process = subprocess.Popen([os.path.join('@MESON_BUILD_ROOT@', "ratbagd.devel")], shell=False)
dbus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
name_owner = None
start_time = time.clock()
while name_owner is None and time.clock() - start_time < 30:
proxy = Gio.DBusProxy.new_sync(dbus,
Gio.DBusProxyFlags.NONE,
None,
"org.freedesktop.ratbag_devel1_@ratbagd_sha@",
"/org/freedesktop/ratbag_devel1_@ratbagd_sha@",
"org.freedesktop.ratbag_devel1_@ratbagd_sha@.Manager",
None)
name_owner = proxy.get_name_owner()
if name_owner is None:
time.sleep(0.2)
os.environ['RATBAGCTL_DEVEL'] = "org.freedesktop.ratbag_devel1_@ratbagd_sha@"
ratbagd = ratbagctl.open_ratbagd()
ratbagd = toolbox.open_ratbagd(ratbagd_process)
assert ratbagd is not None
# if some old version of ratbagd is still running, ratbagd_process may
# have never started but our DBus bindings may succeed. Check for the
# return code here, this also gives ratbagd enough time to start and
# die. If we check immediately we may not have terminated yet.
ratbagd_process.poll()
assert ratbagd_process.returncode is None
parser = ratbagctl.get_parser()
parser = toolbox.get_parser()
def tearDownModule():
global ratbagd_process
try:
ratbagd_process.terminate()
ratbagd_process.wait(5)
except subprocess.TimeoutExpired:
ratbagd_process.kill()
os.unlink(DBUS_CONF_PATH)
toolbox.terminate_ratbagd(ratbagd_process)
def parse(input_string):

112
tools/toolbox.py Normal file
View File

@@ -0,0 +1,112 @@
# This file is part of libratbag.
#
# Copyright 2017 Red Hat, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the "Software"),
# to deal in the Software without restriction, including without limitation
# the rights to use, copy, modify, merge, publish, distribute, sublicense,
# and/or sell copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice (including the next
# paragraph) shall be included in all copies or substantial portions of the
# Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
# THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
import imp
import os
import shutil
import subprocess
import sys
# various constants
RATBAGCTL_NAME = 'ratbagctl'
RATBAGCTL_PATH = os.path.join('@MESON_BUILD_ROOT@', RATBAGCTL_NAME)
RATBAGCTL_DEVEL_NAME = 'ratbagctl.devel'
RATBAGCTL_DEVEL_PATH = os.path.join('@MESON_BUILD_ROOT@', RATBAGCTL_DEVEL_NAME)
DBUS_CONF_DIR = '/etc/dbus-1/system.d'
DBUS_CONF_NAME = 'org.freedesktop.ratbag_devel1.conf'
DBUS_CONF_PATH = os.path.join(DBUS_CONF_DIR, DBUS_CONF_NAME)
def import_non_standard_path(name, path):
# Fast path: see if the module has already been imported.
try:
return sys.modules[name]
except KeyError:
pass
# If any of the following calls raises an exception,
# there's a problem we can't handle -- let the caller handle it.
with open(path, 'rb') as fp:
module = imp.load_module(name, fp, os.path.basename(path), ('.py', 'rb', imp.PY_SOURCE))
return module
def start_ratbagd():
from gi.repository import Gio
import time
# first copy the policy for the ratbagd daemon to be allowed to run
shutil.copy(os.path.join('@MESON_BUILD_ROOT@',
DBUS_CONF_NAME),
DBUS_CONF_DIR)
# FIXME: kill any running ratbagd.devel
ratbagd_process = subprocess.Popen([os.path.join('@MESON_BUILD_ROOT@', "ratbagd.devel")], shell=False)
dbus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
name_owner = None
start_time = time.clock()
while name_owner is None and time.clock() - start_time < 30:
proxy = Gio.DBusProxy.new_sync(dbus,
Gio.DBusProxyFlags.NONE,
None,
"org.freedesktop.ratbag_devel1_@ratbagd_sha@",
"/org/freedesktop/ratbag_devel1_@ratbagd_sha@",
"org.freedesktop.ratbag_devel1_@ratbagd_sha@.Manager",
None)
name_owner = proxy.get_name_owner()
if name_owner is None:
time.sleep(0.2)
os.environ['RATBAGCTL_DEVEL'] = "org.freedesktop.ratbag_devel1_@ratbagd_sha@"
return ratbagd_process
def terminate_ratbagd(ratbagd):
try:
ratbagd.terminate()
ratbagd.wait(5)
except subprocess.TimeoutExpired:
ratbagd.kill()
os.unlink(DBUS_CONF_PATH)
ratbagctl = import_non_standard_path(RATBAGCTL_NAME, RATBAGCTL_PATH)
from ratbagctl import open_ratbagd, get_parser
__all__ = [
RATBAGCTL_NAME,
RATBAGCTL_PATH,
DBUS_CONF_DIR,
DBUS_CONF_NAME,
DBUS_CONF_PATH,
start_ratbagd,
terminate_ratbagd,
open_ratbagd,
get_parser,
]