testcase: factor the server spawning into an object

Let's move spawn_server and spawn_server_template into an object with a
context manager, greatly simplifying them where fixtures are being used.

dbusmock/__init__.py

@@ -15,7 +15,7 @@
 
 
 from dbusmock.mockobject import MOCK_IFACE, OBJECT_MANAGER_IFACE, DBusMockObject, get_object, get_objects
-from dbusmock.testcase import BusType, DBusTestCase, PrivateDBus, spawn_server, spawn_server_template
+from dbusmock.testcase import BusType, DBusTestCase, PrivateDBus, SpawnedServer
 
 try:
  # created by setuptools_scm
@@ -25,4 +25,4 @@
 
 
 __all__ = ['DBusMockObject', 'MOCK_IFACE', 'OBJECT_MANAGER_IFACE',
- 'DBusTestCase', 'PrivateDBus', 'BusType', 'get_object', 'get_objects', 'spawn_server', 'spawn_server_template']
+ 'DBusTestCase', 'PrivateDBus', 'BusType', 'SpawnedServer', 'get_object', 'get_objects']

dbusmock/testcase.py

@@ -27,6 +27,7 @@
 from typing import Any, Dict, Optional, Tuple
 
 import dbus
+import dbus.proxies
 
 from dbusmock.mockobject import MOCK_IFACE, OBJECT_MANAGER_IFACE, load_module
 
@@ -242,12 +243,13 @@ def disable_service(self, service):
  if self._daemon:
  self.bustype.reload_configuration()
 
- def spawn_server(self, name: str, path: str, interface: str, system_bus: bool = False, stdout=None):
+ def spawn_server(self, name: str, path: str, interface: str, system_bus: bool = False, stdout=None) -> subprocess.Popen:
  '''
  Wrapper around ``spawn_server`` for backwards compatibility.
  '''
  assert not system_bus or self.bustype == BusType.SYSTEM, "Mismatching bus types"
- return spawn_server(name=name, path=path, interface=interface, bustype=self.bustype, stdout=stdout)
+ server = SpawnedServer.spawn_for_name(name, path, interface, bustype=self.bustype, stdout=stdout)
+ return server.process
 
  def wait_for_bus_object(self, dest: str, path: str, system_bus: bool = False, timeout: int = 600):
  '''
@@ -260,15 +262,16 @@ def spawn_server_template(self,
  template: str,
  parameters: Optional[Dict[str, Any]] = None,
  stdout=None,
- system_bus: Optional[bool] = None):
+ system_bus: Optional[bool] = None) -> Tuple[subprocess.Popen, dbus.proxies.ProxyObject]:
  '''
  Wrapper around ``spawn_server_template`` for backwards compatibility.
  '''
  if system_bus is not None:
  bustype = BusType.SYSTEM if system_bus else BusType.SESSION
  else:
  bustype = None
- return spawn_server_template(template=template, parameters=parameters, bustype=bustype, stdout=stdout)
+ server = SpawnedServer.spawn_server_template(template=template, parameters=parameters, bustype=bustype, stdout=stdout)
+ return server.process, server.obj
 
  def get_dbus(self, system_bus: bool = False) -> dbus.Bus:
  '''
@@ -430,7 +433,7 @@ def wait_for_bus_object(dest: str, path: str, system_bus: bool = False, timeout:
  bustype.wait_for_bus_object(dest, path, timeout)
 
  @staticmethod
- def spawn_server(name: str, path: str, interface: str, system_bus: bool = False, stdout=None):
+ def spawn_server(name: str, path: str, interface: str, system_bus: bool = False, stdout=None) -> subprocess.Popen:
  '''Run a DBusMockObject instance in a separate process
 
  The daemon will terminate automatically when the D-Bus that it connects
@@ -442,16 +445,18 @@ def spawn_server(name: str, path: str, interface: str, system_bus: bool = False,
 
  Returns the Popen object of the spawned daemon.
 
- This is a legacy method kept for backwards compatibility, use ``spawn_server()`` instead.
+ This is a legacy method kept for backwards compatibility,
+ use SpawnedServer.spawn_server() instead.
  '''
  bustype = BusType.SYSTEM if system_bus else BusType.SESSION
- return spawn_server(name, path, interface, bustype, stdout)
+ server = SpawnedServer.spawn_for_name(name, path, interface, bustype, stdout)
+ return server.process
 
  @staticmethod
  def spawn_server_template(template: str,
  parameters: Optional[Dict[str, Any]] = None,
  stdout=None,
- system_bus: Optional[bool] = None):
+ system_bus: Optional[bool] = None) -> Tuple[subprocess.Popen, dbus.proxies.ProxyObject]:
  '''Run a D-Bus mock template instance in a separate process
 
  This starts a D-Bus mock process and loads the given template with
@@ -471,13 +476,15 @@ def spawn_server_template(template: str,
 
  Returns a pair (daemon Popen object, main dbus object).
 
- This is a legacy method kept for backwards compatibility, use ``spawn_server_template()`` instead.
+ This is a legacy method kept for backwards compatibility,
+ use SpawnedServer.spawn_server_template() instead.
  '''
  if system_bus is not None:
  bustype = BusType.SYSTEM if system_bus else BusType.SESSION
  else:
  bustype = None
- return spawn_server_template(template=template, parameters=parameters, stdout=stdout, bustype=bustype)
+ server = SpawnedServer.spawn_server_template(template, parameters, bustype)
+ return server.process, server.obj
 
  @staticmethod
  def enable_service(service, system_bus: bool = False) -> None:
@@ -508,85 +515,112 @@ def disable_service(service, system_bus: bool = False) -> None:
  bus.disable_service(service)
 
 
-def spawn_server(name: str, path: str, interface: str,
-  bustype: BusType = BusType.SESSION,
-  stdout=None):
- '''Run a DBusMockObject instance in a separate process
+class SpawnedServer:
+ def __init__(self, process: subprocess.Popen, obj: dbus.proxies.ProxyObject):
+ self._process = process
+  self._obj = obj
 
- The daemon will terminate automatically when the D-Bus that it connects
- to goes down. If that does not happen (e. g. you test on the actual
- system/session bus), you need to kill it manually.
+ @property
+ def process(self):
+ return self._process
+
+ @property
+ def obj(self):
+ '''
+ The D-Bus object this server was spawned for.
+ '''
+ return self._obj
 
- This function blocks until the spawned DBusMockObject is ready and
- listening on the bus.
+ def __enter__(self) -> "SpawnedServer":
+  return self
 
- Returns the Popen object of the spawned daemon.
- '''
- argv = [sys.executable, '-m', 'dbusmock', f'--{bustype.value}', name, path, interface]
- bus = bustype.get_connection()
- if bus.name_has_owner(name):
- raise AssertionError(f'Trying to spawn a server for name {name} but it is already owned!')
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.terminate()
 
- # pylint: disable=consider-using-with
- daemon = subprocess.Popen(argv, stdout=stdout)
+ def terminate(self):
+ self._process.terminate()
+ try:
+ self._process.wait(timeout=1)
+ except subprocess.TimeoutExpired:
+ self._process.kill()
 
- # wait for daemon to start up
- bustype.wait_for_bus_object(name, path)
+ def stdout(self):
+  return self._process.stdout
 
- return daemon
+ def stderr(self):
+ return self._process.stderr
 
+ @classmethod
+ def spawn_for_name(cls, name: str, path: str, interface: str,
+ bustype: BusType = BusType.SESSION,
+ stdout=None) -> "SpawnedServer":
+ argv = [sys.executable, '-m', 'dbusmock', f'--{bustype.value}', name, path, interface]
+ bus = bustype.get_connection()
+ if bus.name_has_owner(name):
+ raise AssertionError(f'Trying to spawn a server for name {name} but it is already owned!')
 
-def spawn_server_template(template: str,
- parameters: Optional[Dict[str, Any]] = None,
- bustype: Optional[BusType] = None,
- stdout=None):
- '''Run a D-Bus mock template instance in a separate process
+ if stdout is None:
+ stdout = subprocess.PIPE
 
- This starts a D-Bus mock process and loads the given template with
- (optional) parameters into it. For details about templates see
- dbusmock.DBusMockObject.AddTemplate().
+ # pylint: disable=consider-using-with
+ daemon = subprocess.Popen(argv, stdout=stdout, stderr=subprocess.PIPE)
 
- Usually a template should specify SYSTEM_BUS = False/True to select whether it
- gets loaded on the session or system bus. This can be overridden with the system_bus
- parameter. For templates which don't set SYSTEM_BUS, this parameter has to be set.
+  # wait for daemon to start up
+  bustype.wait_for_bus_object(name, path)
+  obj = bus.get_object(name, path)
 
- The daemon will terminate automatically when the D-Bus that it connects
- to goes down. If that does not happen (e. g. you test on the actual
- system/session bus), you need to kill it manually.
+ return cls(
+ process=daemon,
+ obj=obj
+ )
 
- This function blocks until the spawned DBusMockObject is ready and
- listening on the bus.
+ @classmethod
+ def spawn_server_template(cls,
+ template: str,
+ parameters: Optional[Dict[str, Any]] = None,
+ bustype: Optional[BusType] = None,
+ stdout=None):
+ '''Run a D-Bus mock template instance in a separate process
 
- Returns a pair (daemon Popen object, main dbus object).
- '''
+ This starts a D-Bus mock process and loads the given template with
+ (optional) parameters into it. For details about templates see
+ dbusmock.DBusMockObject.AddTemplate().
+
+ Usually a template should specify SYSTEM_BUS = False/True to select whether it
+ gets loaded on the session or system bus. This can be overridden with the system_bus
+ parameter. For templates which don't set SYSTEM_BUS, this parameter has to be set.
+
+ The daemon will terminate automatically when the D-Bus that it connects
+ to goes down. If that does not happen (e. g. you test on the actual
+ system/session bus), you need to kill it manually.
 
- # we need the bus address from the template module
- module = load_module(template)
+  This function blocks until the spawned DBusMockObject is ready and
+  listening on the bus.
 
- if hasattr(module, 'IS_OBJECT_MANAGER'):
- is_object_manager = module.IS_OBJECT_MANAGER
- else:
- is_object_manager = False
+ Returns a pair (daemon Popen object, main dbus object).
+ '''
 
- if is_object_manager and not hasattr(module, 'MAIN_IFACE'):
- interface_name = OBJECT_MANAGER_IFACE
- else:
- interface_name = module.MAIN_IFACE
+ # we need the bus address from the template module
+ module = load_module(template)
 
- if bustype is None:
- bustype = BusType.SYSTEM if module.SYSTEM_BUS else BusType.SESSION
+ if hasattr(module, 'IS_OBJECT_MANAGER'):
+ is_object_manager = module.IS_OBJECT_MANAGER
+ else:
+ is_object_manager = False
 
- assert bustype is not None
- logger.debug(f"spawning template on {bustype.value} bus")
+ if is_object_manager and not hasattr(module, 'MAIN_IFACE'):
+ interface_name = OBJECT_MANAGER_IFACE
+ else:
+ interface_name = module.MAIN_IFACE
 
- daemon = spawn_server(module.BUS_NAME, module.MAIN_OBJ,
-  interface_name, bustype, stdout)
+  if bustype is None:
+ bustype = BusType.SYSTEM if module.SYSTEM_BUS else BusType.SESSION
 
- bus = bustype.get_connection()
- obj = bus.get_object(module.BUS_NAME, module.MAIN_OBJ)
- if not parameters:
- parameters = dbus.Dictionary({}, signature='sv')
- obj.AddTemplate(template, parameters,
- dbus_interface=MOCK_IFACE)
+ assert bustype is not None
+ logger.debug(f"spawning template on {bustype.value} bus")
 
- return (daemon, obj)
+ server = SpawnedServer.spawn_for_name(module.BUS_NAME, module.MAIN_OBJ, interface_name, bustype, stdout)
+ if not parameters:
+ parameters = dbus.Dictionary({}, signature='sv')
+ server.obj.AddTemplate(template, parameters, dbus_interface=MOCK_IFACE)
+ return server

tests/test_api_pytest.py

@@ -45,3 +45,25 @@ def test_dbusmock_test_spawn_system_template(upower_mock):
  out = subprocess.check_output(['upower', '--dump'], universal_newlines=True)
  assert 'version:' in out
  assert '0.99' in out
+
+
+def test_dbusmock_test_spawnedserver(dbusmock_session):
+ test_iface = 'org.freedesktop.Test.Main'
+
+ with dbusmock.SpawnedServer.spawn_for_name('org.freedesktop.Test', '/', test_iface) as server:
+ obj_test = server.obj
+ obj_test.AddMethod('', 'Upper', 's', 's', 'ret = args[0].upper()', interface_name=dbusmock.MOCK_IFACE)
+ assert obj_test.Upper('hello', interface=test_iface) == 'HELLO'
+
+
+@pytest.fixture(name='upower_mock_spawned')
+def fixture_upower_mock_spawned(dbusmock_system):
+ with dbusmock.SpawnedServer.spawn_server_template('upower') as server:
+ yield server.obj
+
+
+def test_dbusmock_test_spawnedserver_template(upower_mock_spawned):
+ assert upower_mock_spawned
+ out = subprocess.check_output(['upower', '--dump'], universal_newlines=True)
+ assert 'version:' in out
+ assert '0.99' in out