diff --git a/dbusmock/__init__.py b/dbusmock/__init__.py index 7d406b03..ff8a7b33 100644 --- a/dbusmock/__init__.py +++ b/dbusmock/__init__.py @@ -16,6 +16,9 @@ from dbusmock.mockobject import (DBusMockObject, MOCK_IFACE, 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'] + 'DBusTestCase', 'get_object', 'get_objects', + '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]) 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() 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