diff --git a/README.md b/README.md index 45f38384..780662b4 100644 --- a/README.md +++ b/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 diff --git a/dbusmock/__main__.py b/dbusmock/__main__.py index ecc7fb2c..b982301d 100644 --- a/dbusmock/__main__.py +++ b/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) diff --git a/tests/test_cli.py b/tests/test_cli.py index e2248b49..b0ba8ab0 100644 --- a/tests/test_cli.py +++ b/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