functional-tests: Start daemons through D-Bus autolaunch

Instead of manually running and managing Tracker daemon processes
manually in the test, we now rely on our private D-Bus daemon to
do so.

This makes the test environment more like a real Tracker deployment.

Log output from the D-Bus daemon is now captured and output through
the Python logging system. This allows for finer-grained filtering
of output from the tests themselves and from the Tracker daemons.

Some test code is changed to support the new model.

5 files changed, 288 insertions, 343 deletions

diff --git a/utils/sandbox/tracker-sandbox.py b/utils/sandbox/tracker-sandbox.py
index cc8ebd786..34ad09ed5 100755
--- a/utils/sandbox/tracker-sandbox.py
+++ b/utils/sandbox/tracker-sandbox.py
@@ -24,6 +24,7 @@
 #
 
 import argparse
+import configparser
 import locale
 import logging
 import os
@@ -33,10 +34,10 @@ import subprocess
 import sys
 import threading
 
-import configparser
-
 from gi.repository import GLib
 
+import trackertestutils.dbusdaemon
+
 # Script
 script_name = 'tracker-sandbox'
 script_version = '1.0'
@@ -84,116 +85,6 @@ log = logging.getLogger('sandbox')
 dbuslog = logging.getLogger('dbus')
 
 
-# Private DBus daemon
-
-class DBusDaemon:
-    """The private D-Bus instance that provides the sandbox's session bus.
-
-    We support reading and writing the session information to a file. This
-    means that if the user runs two sandbox instances on the same data
-    directory at the same time, they will share the same message bus.
-    """
-
-    def __init__(self, session_file=None):
-        self.session_file = session_file
-        self.existing_session = False
-        self.process = None
-
-        try:
-            self.address, self.pid = self.read_session_file(session_file)
-            self.existing_session = True
-        except FileNotFoundError:
-            log.debug("No existing D-Bus session file was found.")
-
-            self.address = None
-            self.pid = None
-
-    def get_session_file(self):
-        """Returns the path to the session file if we created it, or None."""
-        if self.existing_session:
-            return None
-        return self.session_file
-
-    def get_address(self):
-        return self.address
-
-    @staticmethod
-    def read_session_file(session_file):
-        with open(session_file, 'r') as f:
-            content = f.read()
-
-        try:
-            address = content.splitlines()[0]
-            pid = int(content.splitlines()[1])
-        except ValueError:
-            raise RuntimeError(f"D-Bus session file {session_file} is not valid. "
-                                "Remove this file to start a new session.")
-
-        return address, pid
-
-    @staticmethod
-    def write_session_file(session_file, address, pid):
-        os.makedirs(os.path.dirname(session_file), exist_ok=True)
-
-        content = '%s\n%s' % (address, pid)
-        with open(session_file, 'w') as f:
-            f.write(content)
-
-    def start_if_needed(self):
-        if self.existing_session:
-            log.debug('Using existing D-Bus session from file "%s" with address "%s"'
-                      ' with PID %d' % (self.session_file, self.address, self.pid))
-        else:
-            dbus_command = ['dbus-daemon', '--session', '--print-address=1', '--print-pid=1']
-            self.process = subprocess.Popen(dbus_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-
-            try:
-                self.address = self.process.stdout.readline().strip().decode('ascii')
-                self.pid = int(self.process.stdout.readline().strip().decode('ascii'))
-            except ValueError:
-                error = self.process.stderr.read().strip().decode('unicode-escape')
-                raise RuntimeError(f"Failed to start D-Bus daemon.\n{error}")
-
-            log.debug("Using new D-Bus session with address '%s' with PID %d",
-                      self.address, self.pid)
-
-            self.write_session_file(self.session_file, self.address, self.pid)
-            log.debug("Wrote D-Bus session file at %s", self.session_file)
-
-            # We must read from the pipes continuously, otherwise the daemon
-            # process will block.
-            self._threads=[threading.Thread(target=self.pipe_to_log, args=(self.process.stdout, 'stdout'), daemon=True),
-                           threading.Thread(target=self.pipe_to_log, args=(self.process.stderr, 'stderr'), daemon=True)]
-            self._threads[0].start()
-            self._threads[1].start()
-
-    def stop(self):
-        if self.process:
-            log.debug("  Stopping DBus daemon")
-            self.process.terminate()
-            self.process.wait()
-
-    def pipe_to_log(self, pipe, source):
-        """This function processes the output from our dbus-daemon instance."""
-        while True:
-            line_raw = pipe.readline()
-
-            if len(line_raw) == 0:
-                break
-
-            line = line_raw.decode('utf-8').rstrip()
-
-            if line.startswith('(tracker-'):
-                # We set G_MESSAGES_PREFIXED=all, meaning that all log messages
-                # output by Tracker processes have a prefix. Note that
-                # g_print() will NOT be captured here.
-                dbuslog.info(line)
-            else:
-                # Log messages from other daemons, including the dbus-daemon
-                # itself, go here. Any g_print() messages also end up here.
-                dbuslog.debug(line)
-
-
 # Environment / Clean up
 
 def environment_unset(dbus):
@@ -260,7 +151,7 @@ def environment_set(index_location, prefix, verbosity=0):
     dbus_session_file = os.path.join(
         os.environ['XDG_RUNTIME_DIR'], 'dbus-session')
 
-    dbus = DBusDaemon(dbus_session_file)
+    dbus = trackertestutils.dbusdaemon.DBusDaemon(dbus_session_file)
     dbus.start_if_needed()
 
     # Important, other subprocesses must use our new bus
diff --git a/utils/trackertestutils/dbusdaemon.py b/utils/trackertestutils/dbusdaemon.py
new file mode 100644
index 000000000..43fe8f146
--- /dev/null
+++ b/utils/trackertestutils/dbusdaemon.py
@@ -0,0 +1,198 @@
+# Copyright (C) 2018,2019, Sam Thursfield <sam@afuera.me.uk>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.
+
+
+from gi.repository import Gio
+
+import logging
+import os
+import signal
+import subprocess
+import threading
+
+log = logging.getLogger(__name__)
+dbus_stderr_log = logging.getLogger(__name__ + '.stderr')
+dbus_stdout_log = logging.getLogger(__name__ + '.stdout')
+
+
+class DaemonNotStartedError(Exception):
+    pass
+
+
+class DBusDaemon:
+    """The private D-Bus instance that provides the sandbox's session bus.
+
+    We support reading and writing the session information to a file. This
+    means that if the user runs two sandbox instances on the same data
+    directory at the same time, they will share the same message bus.
+
+    """
+
+    def __init__(self, session_file=None):
+        self.session_file = session_file
+        self.existing_session = False
+        self.process = None
+
+        self.address = None
+        self.pid = None
+
+        self._gdbus_connection = None
+        self._previous_sigterm_handler = None
+
+        self._threads = []
+
+        if session_file:
+            try:
+                self.address, self.pid = self.read_session_file(session_file)
+                self.existing_session = True
+            except FileNotFoundError:
+                log.debug("No existing D-Bus session file was found.")
+
+    def get_session_file(self):
+        """Returns the path to the session file if we created it, or None."""
+        if self.existing_session:
+            return None
+        return self.session_file
+
+    def get_address(self):
+        if self.address is None:
+            raise DaemonNotStartedError()
+        return self.address
+
+    def get_connection(self):
+        if self._gdbus_connection is None:
+            raise DaemonNotStartedError()
+        return self._gdbus_connection
+
+    @staticmethod
+    def read_session_file(session_file):
+        with open(session_file, 'r') as f:
+            content = f.read()
+
+        try:
+            address = content.splitlines()[0]
+            pid = int(content.splitlines()[1])
+        except ValueError:
+            raise RuntimeError(f"D-Bus session file {session_file} is not valid. "
+                                "Remove this file to start a new session.")
+
+        return address, pid
+
+    @staticmethod
+    def write_session_file(session_file, address, pid):
+        os.makedirs(os.path.dirname(session_file), exist_ok=True)
+
+        content = '%s\n%s' % (address, pid)
+        with open(session_file, 'w') as f:
+            f.write(content)
+
+    def start_if_needed(self, config_file=None, env=None):
+        if self.existing_session:
+            log.debug('Using existing D-Bus session from file "%s" with address "%s"'
+                      ' with PID %d' % (self.session_file, self.address, self.pid))
+        else:
+            dbus_command = ['dbus-daemon', '--print-address=1', '--print-pid=1']
+            if config_file:
+                dbus_command += ['--config-file=' + config_file]
+            else:
+                dbus_command += ['--session']
+            log.debug("Running: %s", dbus_command)
+            self.process = subprocess.Popen(
+                dbus_command, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+
+            self._previous_sigterm_handler = signal.signal(
+                signal.SIGTERM, self._sigterm_handler)
+
+            try:
+                self.address = self.process.stdout.readline().strip().decode('ascii')
+                self.pid = int(self.process.stdout.readline().strip().decode('ascii'))
+            except ValueError:
+                error = self.process.stderr.read().strip().decode('unicode-escape')
+                raise RuntimeError(f"Failed to start D-Bus daemon.\n{error}")
+
+            log.debug("Using new D-Bus session with address '%s' with PID %d",
+                      self.address, self.pid)
+
+            if self.session_file:
+                self.write_session_file(self.session_file, self.address, self.pid)
+                log.debug("Wrote D-Bus session file at %s", self.session_file)
+
+            # We must read from the pipes continuously, otherwise the daemon
+            # process will block.
+            self._threads=[threading.Thread(target=self.pipe_to_log, args=(self.process.stdout, dbus_stdout_log), daemon=True),
+                           threading.Thread(target=self.pipe_to_log, args=(self.process.stderr, dbus_stdout_log), daemon=True)]
+            self._threads[0].start()
+            self._threads[1].start()
+
+        self._gdbus_connection = Gio.DBusConnection.new_for_address_sync(
+            self.address,
+            Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT |
+            Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION, None, None)
+
+        log.debug("Pinging the new D-Bus daemon...")
+        self.ping_sync()
+
+    def stop(self):
+        if self.process:
+            log.debug("  Stopping DBus daemon")
+            self.process.terminate()
+            self.process.wait()
+            self.process = None
+        if len(self._threads) > 0:
+            log.debug("  Stopping %i pipe reader threads", len(self._threads))
+            for thread in self._threads:
+                thread.join()
+            self.threads = []
+        if self._previous_sigterm_handler:
+            signal.signal(signal.SIGTERM, self._previous_sigterm_handler)
+            self._previous_sigterm_handler = None
+
+    def pipe_to_log(self, pipe, dbuslog):
+        """This function processes the output from our dbus-daemon instance."""
+        while True:
+            line_raw = pipe.readline()
+
+            if len(line_raw) == 0:
+                break
+
+            line = line_raw.decode('utf-8').rstrip()
+
+            if line.startswith('(tracker-'):
+                # We set G_MESSAGES_PREFIXED=all, meaning that all log messages
+                # output by Tracker processes have a prefix. Note that
+                # g_print() will NOT be captured here.
+                dbuslog.info(line)
+            else:
+                # Log messages from other daemons, including the dbus-daemon
+                # itself, go here. Any g_print() messages also end up here.
+                dbuslog.debug(line)
+        log.debug("Thread stopped")
+
+        # I'm not sure why this is needed, or if it's correct, but without it
+        # we see warnings like this:
+        #
+        #    ResourceWarning: unclosed file <_io.BufferedReader name=3>
+        pipe.close()
+
+    def _sigterm_handler(self, signal, frame):
+        log.info("Received signal %s", signal)
+        self.stop()
+
+    def ping_sync(self):
+        self._gdbus_connection.call_sync(
+            'org.freedesktop.DBus', '/', 'org.freedesktop.DBus', 'GetId',
+            None, None, Gio.DBusCallFlags.NONE, 10000, None)
diff --git a/utils/trackertestutils/dconf.py b/utils/trackertestutils/dconf.py
index 4ad0e88e9..fe6d981fb 100644
--- a/utils/trackertestutils/dconf.py
+++ b/utils/trackertestutils/dconf.py
@@ -18,11 +18,9 @@
 # 02110-1301, USA.
 #
 
-from gi.repository import GLib
-from gi.repository import Gio
-
 import logging
 import os
+import subprocess
 
 log = logging.getLogger(__name__)
 
@@ -36,28 +34,23 @@ class DConfClient(object):
     this reason, and the constructor will fail if this isn't the profile in
     use, to avoid any risk of modifying or removing your real configuration.
 
-    The constructor will fail if DConf is not the default backend, because this
-    probably indicates that the memory backend is in use. Without DConf the
-    required configuration changes will not take effect, causing many tests to
-    break.
+    We use the `gsettings` binary rather than using the Gio.Settings API.
+    This is to avoid the need to set DCONF_PROFILE in our own process
+    environment.
     """
 
-    def __init__(self, schema):
-        self._settings = Gio.Settings.new(schema)
-
-        backend = self._settings.get_property('backend')
-        self._check_settings_backend_is_dconf(backend)
-        self._check_using_correct_dconf_profile()
-
-    def _check_settings_backend_is_dconf(self, backend):
-        typename = type(backend).__name__.split('.')[-1]
-        if typename != 'DConfSettingsBackend':
-            raise Exception(
-                "The functional tests require DConf to be the default "
-                "GSettings backend. Got %s instead." % typename)
+    def __init__(self, sandbox):
+        self.env = os.environ
+        self.env.update(sandbox.extra_env)
+        self.env['DBUS_SESSION_BUS_ADDRESS'] = sandbox.daemon.get_address()
 
     def _check_using_correct_dconf_profile(self):
-        profile = os.environ["DCONF_PROFILE"]
+        profile = self.env.get("DCONF_PROFILE")
+        if not profile:
+            raise Exception(
+                "DCONF_PROFILE is not set in the environment. This class must "
+                "be created inside a TrackerDBussandbox to avoid risk of "
+                "interfering with real settings.")
         if not os.path.exists(profile):
             raise Exception(
                 "Unable to find DConf profile '%s'. Check that Tracker and "
@@ -66,35 +59,11 @@ class DConfClient(object):
 
         assert os.path.basename(profile) == "trackertest"
 
-    def write(self, key, value):
+    def write(self, schema, key, value):
         """
         Write a settings value.
         """
-        self._settings.set_value(key, value)
-
-    def read(self, schema, key):
-        """
-        Read a settings value.
-        """
-        return self._settings.get_value(key)
-
-    def reset(self):
-        """
-        Remove all stored values, resetting configuration to the default.
-
-        This can be done by removing the entire 'trackertest' configuration
-        database.
-        """
-
-        self._check_using_correct_dconf_profile()
-
-        # XDG_CONFIG_HOME is useless, so we use HOME. This code should not be
-        # needed unless for some reason the test is not being run via the
-        # 'test-runner.sh' script.
-        dconf_db = os.path.join(os.environ["HOME"],
-                                ".config",
-                                "dconf",
-                                "trackertest")
-        if os.path.exists(dconf_db):
-            log.debug("[Conf] Removing dconf database: %s", dconf_db)
-            os.remove(dconf_db)
+        subprocess.run(['gsettings', 'set', schema, key, value.print_(False)],
+                       env=self.env,
+                       stdout=subprocess.PIPE,
+                       stderr=subprocess.PIPE)
diff --git a/utils/trackertestutils/helpers.py b/utils/trackertestutils/helpers.py
index e6219be3c..45fa67242 100644
--- a/utils/trackertestutils/helpers.py
+++ b/utils/trackertestutils/helpers.py
@@ -24,8 +24,8 @@ from gi.repository import GLib
 import atexit
 import logging
 import os
-import subprocess
 
+from . import dbusdaemon
 from . import mainloop
 
 log = logging.getLogger(__name__)
@@ -54,172 +54,9 @@ def _cleanup_processes():
 atexit.register(_cleanup_processes)
 
 
-class Helper:
+class StoreHelper():
     """
-    Abstract helper for Tracker processes. Launches the process
-    and waits for it to appear on the session bus.
-
-    The helper will fail if the process is already running. Use
-    test-runner.sh to ensure the processes run inside a separate DBus
-    session bus.
-
-    The process is watched using a timed GLib main loop source. If the process
-    exits with an error code, the test will abort the next time the main loop
-    is entered (or straight away if currently running the main loop).
-    """
-
-    STARTUP_TIMEOUT = 200   # milliseconds
-    SHUTDOWN_TIMEOUT = 200  #
-
-    def __init__(self, helper_name, bus_name, process_path):
-        self.name = helper_name
-        self.bus_name = bus_name
-        self.process_path = process_path
-
-        self.log = logging.getLogger(f'{__name__}.{self.name}')
-
-        self.process = None
-        self.available = False
-
-        self.loop = mainloop.MainLoop()
-
-        self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
-
-    def _start_process(self, command_args=None, extra_env=None):
-        global _process_list
-        _process_list.append(self)
-
-        command = [self.process_path] + (command_args or [])
-        self.log.debug("Starting %s.", ' '.join(command))
-
-        env = os.environ
-        if extra_env:
-            self.log.debug("  starting with extra environment: %s", extra_env)
-            env.update(extra_env)
-
-        try:
-            return subprocess.Popen(command, env=env)
-        except OSError as e:
-            raise RuntimeError("Error starting %s: %s" % (self.process_path, e))
-
-    def _bus_name_appeared(self, connection, name, owner):
-        self.log.debug("%s appeared on the message bus, owned by %s", name, owner)
-        self.available = True
-        self.loop.quit()
-
-    def _bus_name_vanished(self, connection, name):
-        self.log.debug("%s vanished from the message bus", name)
-        self.available = False
-        self.loop.quit()
-
-    def _process_watch_cb(self):
-        if self.process_watch_timeout == 0:
-            # GLib seems to call the timeout after we've removed it
-            # sometimes, which causes errors unless we detect it.
-            return False
-
-        status = self.process.poll()
-
-        if status is None:
-            return True    # continue
-        elif status == 0 and not self.abort_if_process_exits_with_status_0:
-            return True    # continue
-        else:
-            self.process_watch_timeout = 0
-            raise RuntimeError(f"{self.name} exited with status: {self.status}")
-
-    def _process_startup_timeout_cb(self):
-        self.log.debug(f"Process timeout of {self.STARTUP_TIMEOUT}ms was called")
-        self.loop.quit()
-        self.timeout_id = None
-        return False
-
-    def start(self, command_args=None, extra_env=None):
-        """
-        Start an instance of process and wait for it to appear on the bus.
-        """
-        if self.process is not None:
-            raise RuntimeError("%s: already started" % self.name)
-
-        self._bus_name_watch_id = Gio.bus_watch_name_on_connection(
-            self.bus, self.bus_name, Gio.BusNameWatcherFlags.NONE,
-            self._bus_name_appeared, self._bus_name_vanished)
-
-        # We expect the _bus_name_vanished callback to be called here,
-        # causing the loop to exit again.
-        self.loop.run_checked()
-
-        if self.available:
-            # It's running, but we didn't start it...
-            raise RuntimeError("Unable to start test instance of %s: "
-                               "already running" % self.name)
-
-        self.process = self._start_process(command_args=command_args,
-                                           extra_env=extra_env)
-        self.log.debug('Started with PID %i', self.process.pid)
-
-        self.process_startup_timeout = GLib.timeout_add(
-            self.STARTUP_TIMEOUT, self._process_startup_timeout_cb)
-
-        self.abort_if_process_exits_with_status_0 = True
-
-        # Run the loop until the bus name appears, or the process dies.
-        self.loop.run_checked()
-
-        self.abort_if_process_exits_with_status_0 = False
-
-    def stop(self):
-        global _process_list
-
-        if self.process is None:
-            # Seems that it didn't even start...
-            return
-
-        if self.process_startup_timeout != 0:
-            GLib.source_remove(self.process_startup_timeout)
-            self.process_startup_timeout = 0
-
-        if self.process.poll() == None:
-            self.process.terminate()
-            returncode = self.process.wait(timeout=self.SHUTDOWN_TIMEOUT * 1000)
-            if returncode is None:
-                self.log.debug("Process failed to terminate in time, sending kill!")
-                self.process.kill()
-                self.process.wait()
-            elif returncode > 0:
-                self.log.warn("Process returned error code %s", returncode)
-
-        self.log.debug("Process stopped.")
-
-        # Run the loop to handle the expected name_vanished signal.
-        self.loop.run_checked()
-        Gio.bus_unwatch_name(self._bus_name_watch_id)
-
-        self.process = None
-        _process_list.remove(self)
-
-    def kill(self):
-        global _process_list
-
-        if self.process_watch_timeout != 0:
-            GLib.source_remove(self.process_watch_timeout)
-            self.process_watch_timeout = 0
-
-        self.process.kill()
-
-        # Name owner changed callback should take us out from this loop
-        self.loop.run_checked()
-        Gio.bus_unwatch_name(self._bus_name_watch_id)
-
-        self.process = None
-        _process_list.remove(self)
-
-        self.log.debug("Process killed.")
-
-
-class StoreHelper (Helper):
-    """
-    Helper for starting and testing the tracker-store daemon.
+    Helper for testing the tracker-store daemon.
     """
 
     TRACKER_BUSNAME = 'org.freedesktop.Tracker1'
@@ -235,32 +72,41 @@ class StoreHelper (Helper):
     TRACKER_STATUS_OBJ_PATH = "/org/freedesktop/Tracker1/Status"
     STATUS_IFACE = "org.freedesktop.Tracker1.Status"
 
-    def __init__(self, process_path):
-        Helper.__init__(self, "tracker-store", self.TRACKER_BUSNAME, process_path)
+    def __init__(self, dbus_connection):
+        self.log = logging.getLogger(__name__)
+        self.loop = mainloop.MainLoop()
 
-    def start(self, command_args=None, extra_env=None):
-        Helper.start(self, command_args, extra_env)
+        self.bus = dbus_connection
+        self.graph_updated_handler_id = 0
 
         self.resources = Gio.DBusProxy.new_sync(
-            self.bus, Gio.DBusProxyFlags.DO_NOT_AUTO_START, None,
+            self.bus, Gio.DBusProxyFlags.DO_NOT_AUTO_START_AT_CONSTRUCTION, None,
             self.TRACKER_BUSNAME, self.TRACKER_OBJ_PATH, self.RESOURCES_IFACE)
 
         self.backup_iface = Gio.DBusProxy.new_sync(
-            self.bus, Gio.DBusProxyFlags.DO_NOT_AUTO_START, None,
+            self.bus, Gio.DBusProxyFlags.DO_NOT_AUTO_START_AT_CONSTRUCTION, None,
             self.TRACKER_BUSNAME, self.TRACKER_BACKUP_OBJ_PATH, self.BACKUP_IFACE)
 
         self.stats_iface = Gio.DBusProxy.new_sync(
-            self.bus, Gio.DBusProxyFlags.DO_NOT_AUTO_START, None,
+            self.bus, Gio.DBusProxyFlags.DO_NOT_AUTO_START_AT_CONSTRUCTION, None,
             self.TRACKER_BUSNAME, self.TRACKER_STATS_OBJ_PATH, self.STATS_IFACE)
 
         self.status_iface = Gio.DBusProxy.new_sync(
-            self.bus, Gio.DBusProxyFlags.DO_NOT_AUTO_START, None,
+            self.bus, Gio.DBusProxyFlags.DO_NOT_AUTO_START_AT_CONSTRUCTION, None,
             self.TRACKER_BUSNAME, self.TRACKER_STATUS_OBJ_PATH, self.STATUS_IFACE)
 
+    def start_and_wait_for_ready(self):
+        # The daemon is autostarted as soon as a method is called.
+        #
+        # We set a big timeout to avoid interfering when a daemon is being
+        # interactively debugged.
         self.log.debug("Calling %s.Wait() method", self.STATUS_IFACE)
-        self.status_iface.Wait()
+        self.status_iface.call_sync('Wait', None, Gio.DBusCallFlags.NONE, 1000000, None)
         self.log.debug("Ready")
 
+    def start_watching_updates(self):
+        assert self.graph_updated_handler_id == 0
+
         self.reset_graph_updates_tracking()
 
         def signal_handler(proxy, sender_name, signal_name, parameters):
@@ -269,12 +115,13 @@ class StoreHelper (Helper):
 
         self.graph_updated_handler_id = self.resources.connect(
             'g-signal', signal_handler)
+        self.log.debug("Watching for updates from Resources interface")
 
-    def stop(self):
-        Helper.stop(self)
-
+    def stop_watching_updates(self):
         if self.graph_updated_handler_id != 0:
+            self.log.debug("No longer watching for updates from Resources interface")
             self.resources.disconnect(self.graph_updated_handler_id)
+            self.graph_updated_handler_id = 0
 
     # A system to follow GraphUpdated and make sure all changes are tracked.
     # This code saves every change notification received, and exposes methods
@@ -329,6 +176,7 @@ class StoreHelper (Helper):
         """
         assert (self.inserts_match_function == None)
         assert (self.class_to_track == None), "Already waiting for resource of type %s" % self.class_to_track
+        assert (self.graph_updated_handler_id != 0), "You must call start_watching_updates() first."
 
         self.class_to_track = rdf_class
 
@@ -413,6 +261,7 @@ class StoreHelper (Helper):
         """
         assert (self.deletes_match_function == None)
         assert (self.class_to_track == None)
+        assert (self.graph_updated_handler_id != 0), "You must call start_watching_updates() first."
 
         def find_resource_deletion(deletes_list):
             self.log.debug("find_resource_deletion: looking for %i in %s", id, deletes_list)
@@ -444,8 +293,8 @@ class StoreHelper (Helper):
             # Run the event loop until the correct notification arrives
             try:
                 self.loop.run_checked()
-            except GraphUpdateTimeoutException:
-                raise GraphUpdateTimeoutException("Resource %i has not been deleted." % id)
+            except GraphUpdateTimeoutException as e:
+                raise GraphUpdateTimeoutException("Resource %i has not been deleted." % id) from e
             self.deletes_match_function = None
             self.class_to_track = None
 
@@ -458,6 +307,7 @@ class StoreHelper (Helper):
         assert (self.inserts_match_function == None)
         assert (self.deletes_match_function == None)
         assert (self.class_to_track == None)
+        assert (self.graph_updated_handler_id != 0), "You must call start_watching_updates() first."
 
         self.log.debug("Await change to %i %s (%i, %i existing)", subject_id, property_uri, len(self.inserts_list), len(self.deletes_list))
 
@@ -582,3 +432,39 @@ class StoreHelper (Helper):
             return False
         else:
             raise Exception("Something fishy is going on")
+
+
+class TrackerDBusSandbox:
+    """
+    Private D-Bus session bus which executes a sandboxed Tracker instance.
+
+    """
+    def __init__(self, dbus_daemon_config_file, extra_env=None):
+        self.dbus_daemon_config_file = dbus_daemon_config_file
+        self.extra_env = extra_env or {}
+
+        self.daemon = dbusdaemon.DBusDaemon()
+
+    def start(self):
+        env = os.environ
+        env.update(self.extra_env)
+        env['G_MESSAGES_PREFIXED'] = 'all'
+
+        # Precreate runtime dir, to avoid this warning from dbus-daemon:
+        #
+        #    Unable to set up transient service directory: XDG_RUNTIME_DIR "/home/sam/tracker-tests/tmp_59i3ev1/run" not available: No such file or directory
+        #
+        xdg_runtime_dir = env.get('XDG_RUNTIME_DIR')
+        if xdg_runtime_dir:
+            os.makedirs(xdg_runtime_dir, exist_ok=True)
+
+        log.info("Starting D-Bus daemon for sandbox.")
+        log.debug("Added environment variables: %s", self.extra_env)
+        self.daemon.start_if_needed(self.dbus_daemon_config_file, env=env)
+
+    def stop(self):
+        log.info("Stopping D-Bus daemon for sandbox.")
+        self.daemon.stop()
+
+    def get_connection(self):
+        return self.daemon.get_connection()
diff --git a/utils/trackertestutils/meson.build b/utils/trackertestutils/meson.build
index 99573e323..a144794d5 100644
--- a/utils/trackertestutils/meson.build
+++ b/utils/trackertestutils/meson.build
@@ -1,5 +1,6 @@
 sources = [
   '__init__.py',
+  'dbusdaemon.py',
   'dconf.py',
   'helpers.py',
   'mainloop.py'