mockobject: Allow adding objects derived from DBusMockObject

dbusmock/mockobject.py

@@ -199,6 +199,7 @@ def __init__(
  props: PropsType,
  logfile: Optional[str] = None,
  is_object_manager: bool = False,
+ mock_data: Any = None,
  ) -> None:
  """Create a new DBusMockObject
 
@@ -234,6 +235,7 @@ def __init__(
  self.logfile = open(logfile, "wb") if logfile else None # noqa: SIM115
  self.is_logfile_owner = True
  self.call_log: List[CallLogType] = []
+ self.mock_data = mock_data
 
  if props is None:
  props = {}
@@ -323,7 +325,9 @@ def Set(self, interface_name: str, property_name: str, value: Any, *_, **__) ->
  )
 
  @dbus.service.method(MOCK_IFACE, in_signature="ssa{sv}a(ssss)", out_signature="")
- def AddObject(self, path: str, interface: str, properties: PropsType, methods: List[MethodType]) -> None:
+ def AddObject(
+ self, path: str, interface: str, properties: PropsType, methods: List[MethodType], **kwargs
+ ) -> None:
  """Dynamically add a new D-Bus object to the mock
 
  :param path: D-Bus object path
@@ -335,6 +339,12 @@ def AddObject(self, path: str, interface: str, properties: PropsType, methods: L
  methods to add to "interface"; see AddMethod() for details of
  the tuple values
 
+ Keyword Arguments:
+ :param mock_class: A DBusMockObject derived class name (this is only possible
+ when the method is not called via dbus)
+ :param mock_data: Additional data which will be passed to the DBusMockObject
+ constructor
+
  If this is a D-Bus ObjectManager instance, the InterfacesAdded signal
  will *not* be emitted for the object automatically; it must be emitted
  manually if desired. This is because AddInterface may be called after
@@ -352,13 +362,33 @@ def AddObject(self, path: str, interface: str, properties: PropsType, methods: L
  ('EchoInt', 'i', 'i', 'ret = args[0]'),
  ('GetClients', '', 'ao', 'ret = ["/com/example/Foo/Client1"]'),
  ])
+
+ Example 2 (in a template)::
+
+ class FooManager(dbusmock.mockobject.DBusMockObject):
+ def __init__(self, *args, **kwargs):
+ super(FooManager, self).__init__(*args, **kwargs)
+ self.magic = kwargs.get('mock_data')
+
+ @dbus.service.method('com.example.Foo.Control', in_signature='i', out_signature='i')
+ def EchoMagic(self, input):
+ return self.magic + input
+
+ dbus_proxy.AddObject('/com/example/Foo/Manager',
+ 'com.example.Foo.Control',
+ {}, [],
+ mock_class=FooManager,
+ mock_data=42)
  """
  if path in objects:
  raise dbus.exceptions.DBusException(
  f"object {path} already exists", name="org.freedesktop.DBus.Mock.NameError"
  )
 
- obj = DBusMockObject(self.bus_name, path, interface, properties)
+ mock_class = kwargs.get("mock_class", DBusMockObject)
+ mock_data = kwargs.get("mock_data")
+
+ obj = mock_class(self.bus_name, path, interface, properties, mock_data=mock_data)
  # make sure created objects inherit the log file stream
  obj.logfile = self.logfile
  obj.object_manager = self.object_manager

pyproject.toml

@@ -43,7 +43,7 @@ version_scheme = 'post-release'
 
 [tool.pylint]
 format = { max-line-length = 130 }
-"messages control" = { disable = ["invalid-name"] }
+"messages control" = { disable = ["invalid-name", "too-many-arguments"] }
 design = { max-args = 7, max-locals = 25, max-public-methods = 25 }
 
 [tool.mypy]

tests/test_api.py

@@ -923,6 +923,45 @@ def load(mock, parameters):
  self.assertIn('<node name="Thing1" />', xml)
  self.assertIn('<node name="Thing2" />', xml)
 
+ def test_add_object_with_subclass(self):
+ """Template with DBusMockObject subclass"""
+
+ with tempfile.NamedTemporaryFile(prefix="objmgr_", suffix=".py") as my_template:
+ my_template.write(
+ b"""
+import dbus
+import dbusmock
+BUS_NAME = 'org.test.Things'
+MAIN_OBJ = '/org/test/Things'
+MAIN_IFACE = 'org.test.Do'
+SYSTEM_BUS = False
+
+class Thing1(dbusmock.mockobject.DBusMockObject):
+ def __init__(self, *args, **kwargs):
+ super(Thing1, self).__init__(*args, **kwargs)
+ self.magic = kwargs.get('mock_data')
+
+ @dbus.service.method(MAIN_IFACE, in_signature='i', out_signature='i')
+ def Do0(self, input):
+ return self.magic + input
+
+def load(mock, parameters):
+ mock.AddObject('/org/test/Things/Thing1', MAIN_IFACE, {}, [],
+ mock_class=Thing1,
+ mock_data=42)
+"""
+ )
+ my_template.flush()
+ (p_mock, _) = self.spawn_server_template(my_template.name, stdout=subprocess.PIPE, system_bus=False)
+ self.addCleanup(p_mock.wait)
+ self.addCleanup(p_mock.terminate)
+ self.addCleanup(p_mock.stdout.close)
+
+ dbus_con = self.get_dbus(system_bus=False)
+ thing1 = dbus_con.get_object("org.test.Things", "/org/test/Things/Thing1")
+ self.assertEqual(thing1.Do0(0), 42)
+ self.assertEqual(thing1.Do0(1), 43)
+
  def test_reset(self):
  """Reset() puts the template back to pristine state"""