From 799b051f5bface9cc3ae5f30d6120bc229192893 Mon Sep 17 00:00:00 2001 From: Benjamin Berg Date: Thu, 5 Apr 2018 15:47:59 +0200 Subject: [PATCH 1/3] Add X11SessionTestCase This test will automatically start up Xvfb and both a system and session bus. This can be useful for testing UI application. --- dbusmock/__init__.py | 4 +- dbusmock/x11session.py | 115 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 118 insertions(+), 1 deletion(-) create mode 100644 dbusmock/x11session.py diff --git a/dbusmock/__init__.py b/dbusmock/__init__.py index 7d406b03..943ffcfb 100644 --- a/dbusmock/__init__.py +++ b/dbusmock/__init__.py @@ -16,6 +16,8 @@ from dbusmock.mockobject import (DBusMockObject, MOCK_IFACE, OBJECT_MANAGER_IFACE, get_object, get_objects) from dbusmock.testcase import DBusTestCase +from dbusmock.x11session import X11SessionTestCase __all__ = ['DBusMockObject', 'MOCK_IFACE', 'OBJECT_MANAGER_IFACE', - 'DBusTestCase', 'get_object', 'get_objects'] + 'DBusTestCase', 'get_object', 'get_objects', + 'X11SessionTestCase'] diff --git a/dbusmock/x11session.py b/dbusmock/x11session.py new file mode 100644 index 00000000..58ddee82 --- /dev/null +++ b/dbusmock/x11session.py @@ -0,0 +1,115 @@ +import os +import sys +import subprocess +from .testcase import DBusTestCase + +if sys.version_info[0] < 3: + PIPE_DEVNULL = open(os.devnull, 'wb') + + def Popen(*args, **kwargs): + if 'pass_fds' in kwargs: + pass_fds = kwargs['pass_fds'] + del kwargs['pass_fds'] + else: + pass_fds = [] + + orig_preexec_fn = kwargs.get('preexec_fn', None) + + def _setup(): + for fd in range(3, subprocess.MAXFD): + if fd in pass_fds: + continue + + try: + os.close(fd) + except OSError: + pass + + if orig_preexec_fn: + orig_preexec_fn() + + # Don't let subprocess close FDs for us + kwargs['close_fds'] = False + kwargs['preexec_fn'] = _setup + + return subprocess.Popen(*args, **kwargs) + +else: + PIPE_DEVNULL = subprocess.DEVNULL + Popen = subprocess.Popen + + +class X11SessionTestCase(DBusTestCase): + #: The display the X server is running on + X_display = -1 + #: The X server to start + Xserver = 'Xvfb' + #: Further parameters for the X server + Xserver_args = ['-screen', '0', '1280x1024x24', '+extension', 'GLX'] + #: Where to redirect the X stdout and stderr to. Set to None for debugging + #: purposes if the X server is failing for some reason. + Xserver_output = PIPE_DEVNULL + + @classmethod + def setUpClass(klass): + klass.start_xorg() + klass.start_system_bus() + klass.start_session_bus() + + @classmethod + def start_xorg(klass): + r, w = os.pipe() + + # Xvfb seems to randomly crash in some workloads if "-noreset" is not given. + # https://bugzilla.redhat.com/show_bug.cgi?id=1565847 + klass.xorg = Popen([klass.Xserver, '-displayfd', "%d" % w, '-noreset'] + klass.Xserver_args, + pass_fds=(w,), + stdout=klass.Xserver_output, + stderr=subprocess.STDOUT) + os.close(w) + + # The X server will write "%d\n", we need to make sure to read the "\n". + # If the server dies we get zero bytes reads as EOF is reached. + display = b'' + while True: + tmp = os.read(r, 1024) + display += tmp + + # Break out if the read was empty or the line feed was read + if not tmp or tmp[-1] == b'\n': + break + + os.close(r) + + try: + display = int(display.strip()) + except ValueError: + # This should never happen, the X server didn't return a proper integer. + # Most likely it died for some odd reason. + # Note: Set Xserver_output to None to debug Xvfb startup issues. + klass.stop_xorg() + raise AssertionError('X server reported back no or an invalid display number (%s)' % (display)) + + klass.X_display = display + # Export information into our environment for tests to use + os.environ['DISPLAY'] = ":%d" % klass.X_display + os.environ['WAYLAND'] = '' + + # Server should still be up and running at this point + assert klass.xorg.poll() is None + + return klass.X_display + + @classmethod + def stop_xorg(klass): + if hasattr(klass, 'xorg'): + klass.X_display = -1 + klass.xorg.terminate() + klass.xorg.wait() + del klass.xorg + + @classmethod + def tearDownClass(klass): + DBusTestCase.tearDownClass() + + klass.stop_xorg() From 1e24b958f5bce81bbd21be5dcc42621dcb03478c Mon Sep 17 00:00:00 2001 From: Benjamin Berg Date: Thu, 5 Apr 2018 17:34:11 +0200 Subject: [PATCH 2/3] Add python-future to dependency list for testing --- tests/run-ubuntu-chroot | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/run-ubuntu-chroot b/tests/run-ubuntu-chroot index f09c8223..6dade85d 100755 --- a/tests/run-ubuntu-chroot +++ b/tests/run-ubuntu-chroot @@ -49,7 +49,7 @@ if [ -n "${PROPOSED:-}" ]; then fi apt-get update $UPGRADE -apt-get install -y python-all python-setuptools python3-all python3-setuptools python-nose python-dbus python-gi python3-nose python3-dbus python3-gi gir1.2-glib-2.0 dbus libnotify-bin upower network-manager pyflakes3 bluez +apt-get install -y python-all python-setuptools python3-all python3-setuptools python-nose python-dbus python-gi python-future python3-nose python3-dbus python3-gi python3-future gir1.2-glib-2.0 dbus libnotify-bin upower network-manager pyflakes3 bluez apt-get install -y pycodestyle || true # run build and tests as user From 1a89bcd4e8c081c47cfe4f8665fb778e6be9b37c Mon Sep 17 00:00:00 2001 From: Benjamin Berg Date: Thu, 5 Apr 2018 15:49:08 +0200 Subject: [PATCH 3/3] Add a GTest mixin for use with glib based test applications This is just a simple wrapper that allows calling test binaries created with the GLib testing infrastructure and produces nicer output. In particular, the output will only be printed if the test failed. There is fallback code which will simply run the binary as is. --- dbusmock/__init__.py | 3 +- dbusmock/gtest.py | 108 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 1 deletion(-) create mode 100644 dbusmock/gtest.py diff --git a/dbusmock/__init__.py b/dbusmock/__init__.py index 943ffcfb..ff8a7b33 100644 --- a/dbusmock/__init__.py +++ b/dbusmock/__init__.py @@ -17,7 +17,8 @@ OBJECT_MANAGER_IFACE, get_object, get_objects) from dbusmock.testcase import DBusTestCase from dbusmock.x11session import X11SessionTestCase +from dbusmock.gtest import GTest __all__ = ['DBusMockObject', 'MOCK_IFACE', 'OBJECT_MANAGER_IFACE', 'DBusTestCase', 'get_object', 'get_objects', - 'X11SessionTestCase'] + 'X11SessionTestCase', 'GTest'] diff --git a/dbusmock/gtest.py b/dbusmock/gtest.py new file mode 100644 index 00000000..96d0d330 --- /dev/null +++ b/dbusmock/gtest.py @@ -0,0 +1,108 @@ +#!/usr/bin/python3 + +import os +import sys +import subprocess +import functools +from future.utils import with_metaclass + +if sys.version_info[0] < 3: + PIPE_DEVNULL = open(os.devnull, 'wb') +else: + PIPE_DEVNULL = subprocess.DEVNULL + + +class _GTestSingleProp(object): + """Property which creates a bound method for calling the specified test.""" + def __init__(self, test): + self.test = test + + @staticmethod + def __func(self, test): + self._gtest_single(test) + + def __get__(self, obj, cls): + bound_method = self.__func.__get__(obj, cls) + partial_method = functools.partial(bound_method, self.test) + partial_method.__doc__ = bound_method.__doc__ + + return partial_method + + +class _GTestMeta(type): + def __new__(cls, name, bases, namespace, **kwds): + result = type.__new__(cls, name, bases, dict(namespace)) + + if result.g_test_exe is not None: + try: + _GTestMeta.make_tests(result.g_test_exe, result) + except Exception as e: + print('') + print(e) + print('Error generating separate test funcs, will call binary once.') + result.test_all = result._gtest_all + + return result + + @staticmethod + def make_tests(exe, result): + test = subprocess.Popen([exe, '-l'], stdout=subprocess.PIPE, stderr=PIPE_DEVNULL) + stdout, stderr = test.communicate() + + if test.returncode != 0: + raise AssertionError('Execution of GTest executable to query the tests returned non-zero exit code!') + + stdout = stdout.decode('utf-8') + + for i, test in enumerate(stdout.split('\n')): + if not test: + continue + + # Number it and make sure the function name is prefixed with 'test'. + # Keep the rest as is, we don't care about the fact that the function + # names cannot be typed in. + name = 'test_%03d_' % (i + 1) + test + setattr(result, name, _GTestSingleProp(test)) + + +class GTest(with_metaclass(_GTestMeta)): + """Helper class to run GLib test. A test function will be created for each + test from the executable. + + Use by using this class as a mixin and setting g_test_exe to an appropriate + value. + """ + + #: The GTest based executable + g_test_exe = None + #: Timeout when running a single test, only set when using python 3! + g_test_single_timeout = None + #: Timeout when running all tests in one go, only set when using python 3! + g_test_all_timeout = None + + def _gtest_single(self, test): + assert(test) + p = subprocess.Popen([self.g_test_exe, '-q', '-p', test], stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + try: + if self.g_test_single_timeout: + stdout, stderr = p.communicate(timeout=self.g_test_single_timeout) + else: + stdout, stderr = p.communicate() + except subprocess.TimeoutExpired: + p.kill() + stdout, stderr = p.communicate() + stdout += b'\n\nTest was aborted due to timeout' + + try: + stdout = stdout.decode('utf-8') + except UnicodeDecodeError: + pass + + if p.returncode != 0: + self.fail(stdout) + + def _gtest_all(self): + if self.g_test_all_timeout: + subprocess.check_call([self.g_test_exe], timeout=self.g_test_all_timeout) + else: + subprocess.check_call([self.g_test_exe])