cli: Add support for running custom commands on the mock environment

It may be used to quickly debug interfaces, running specific tools.

README.md

@@ -165,6 +165,20 @@ call.
 You can do the same operations in e. g. d-feet or any other D-Bus
 language binding.
 
+## Interactive debugging
+
+It's possible to use dbus-mock to run interactive sessions using something like:
+
+    python3 -m dbusmock com.example.Foo / com.example.Foo.Manager -e $SHELL
+
+Where a shell session with the defined mocks is set and others can be added.
+
+Or more complex ones such as:
+
+    python3 -m dbusmock --session -t upower -e \
+      python3 -m dbusmock com.example.Foo / com.example.Foo.Manager -e \
+        gdbus introspect --session -d com.example.Foo -o /
+
 ## Logging
 
 Usually you want to verify which methods have been called on the mock

dbusmock/__main__.py

@@ -15,6 +15,8 @@
 
 import argparse
 import json
+import os
+import subprocess
 import sys
 
 import dbusmock.mockobject
@@ -43,6 +45,8 @@ def parse_args():
                         help='automatically implement the org.freedesktop.DBus.ObjectManager interface')
     parser.add_argument('-p', '--parameters',
                         help='JSON dictionary of parameters to pass to the template')
+    parser.add_argument('-e', '--exec', nargs=argparse.REMAINDER,
+                        help='Command to run in the mock environment')
 
     arguments = parser.parse_args()
 
@@ -120,8 +124,34 @@ def parse_args():
     if args.template:
         main_object.AddTemplate(args.template, parameters)
 
+    libglib = ctypes.cdll.LoadLibrary('libglib-2.0.so.0')
+
     dbusmock.mockobject.objects[args.path] = main_object
 
-    libglib = ctypes.cdll.LoadLibrary('libglib-2.0.so.0')
-    while should_run:
-        libglib.g_main_context_iteration(None, True)
+    if args.exec:
+        with subprocess.Popen(args.exec) as exec_proc:
+            exit_status = set()
+
+            @ctypes.CFUNCTYPE(None, ctypes.c_int, ctypes.c_int)
+            def on_process_watch(_pid, status):
+                """ Check if the launched process is still alive """
+                if os.WIFEXITED(status):
+                    exit_status.add(os.WEXITSTATUS(status))
+                else:
+                    exit_status.add(1)
+                should_run.pop()
+
+            libglib.g_child_watch_add(exec_proc.pid, on_process_watch)
+
+            while should_run:
+                libglib.g_main_context_iteration(None, True)
+
+            try:
+                exec_proc.terminate()
+                exec_proc.wait()
+            except ProcessLookupError:
+                pass
+            sys.exit(exit_status.pop() if exit_status else exec_proc.returncode)
+    else:
+        while should_run:
+            libglib.g_main_context_iteration(None, True)

tests/test_cli.py

@@ -25,6 +25,7 @@
 
 tracemalloc.start(25)
 have_upower = shutil.which('upower')
+have_gdbus = shutil.which('gdbus')
 
 
 class TestCLI(dbusmock.DBusTestCase):
@@ -57,6 +58,10 @@ def start_mock(self, args, wait_name, wait_path, wait_system=False):
                                        universal_newlines=True)
         self.wait_for_bus_object(wait_name, wait_path, wait_system)
 
+    def start_mock_process(self, args):
+        return subprocess.check_output([sys.executable, '-m', 'dbusmock'] + args,
+                                       universal_newlines=True)
+
     def test_session_bus(self):
         self.start_mock(['com.example.Test', '/', 'TestIface'],
                         'com.example.Test', '/')
@@ -141,6 +146,27 @@ def test_template_parameters_not_dict(self):
         self.assertEqual(err.returncode, 2)
         self.assertEqual(err.output, 'JSON parameters must be a dictionary\n')
 
+    @unittest.skipIf(not have_upower, 'No upower installed')
+    def test_template_upower_exec(self):
+        out = self.start_mock_process(
+            ['-t', 'upower', '--exec', 'upower', '--dump'])
+        self.assertRegex(out, r'on-battery:\s+no')
+        self.assertRegex(out, r'daemon-version:\s+0\.99')
+
+    @unittest.skipIf(not have_gdbus, 'No gdbus installed')
+    def test_manual_upower_exec(self):
+        out = self.start_mock_process(
+            ['--system',
+             'org.freedesktop.UPower',
+             '/org/freedesktop/UPower',
+             'org.freedesktop.UPower',
+             '--exec',
+             'gdbus', 'introspect', '--system',
+             '--dest', 'org.freedesktop.UPower',
+             '--object-path', '/org/freedesktop/UPower'])
+        self.assertRegex(out, r'AddMethod\(')
+        self.assertRegex(out, r'AddMethods\(')
+
     def test_template_local(self):
         with tempfile.NamedTemporaryFile(prefix='answer_', suffix='.py') as my_template:
             my_template.write(b'''import dbus