Add helpers for application testing

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']

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])

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()

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