Added sdbus.unittest.IsolatedDbusTestCase

Extension of `unittest.IsolatedAsyncioTestCase` from standard
library.
Allows to run asyncio tests in isolated D-Bus instance.

docs/index.rst

@@ -39,9 +39,10 @@ If you are unfamiliar with D-Bus you might want to read following pages:
     asyncio_api
     exceptions
     examples
-    autodoc
-    code_generator
     proxies
+    code_generator
+    autodoc
+    unittest
     api_index
 
 

docs/unittest.rst

@@ -0,0 +1,62 @@
+Unit testing
+============
+
+Python-sdbus provides several utilities to enable unit testing.
+
+.. py:currentmodule:: sdbus.unittest
+
+.. py:class:: IsolatedDbusTestCase
+
+    Extension of `unittest.IsolatedAsyncioTestCase
+    <https://docs.python.org/3/library/unittest.html#unittest.IsolatedAsyncioTestCase>`__
+    from standard library.
+
+    Creates an isolated instance of session D-Bus. The D-Bus will be closed
+    and cleaned up after tests are finished.
+
+    Requires ``dbus-daemon`` executable be installed.
+
+    .. py:attribute:: bus
+        :type: SdBus
+
+        Bus instance connected to isolated D-Bus environment.
+
+        It is also set as a default bus.
+
+
+Usage example: ::
+
+    from sdbus import DbusInterfaceCommonAsync, dbus_method_async
+    from sdbus.unittest import IsolatedDbusTestCase
+
+    class TestInterface(DbusInterfaceCommonAsync,
+                        interface_name='org.test.test',
+                        ):
+
+        @dbus_method_async("s", "s")
+        async def upper(self, string: str) -> str:
+            """Uppercase the input"""
+            return string.upper()
+
+    def initialize_object() -> Tuple[TestInterface, TestInterface]:
+        test_object = TestInterface()
+        test_object.export_to_dbus('/')
+
+        test_object_connection = TestInterface.new_proxy(
+        "org.example.test", '/')
+
+        return test_object, test_object_connection
+
+
+    class TestProxy(IsolatedDbusTestCase):
+        async def asyncSetUp(self) -> None:
+            await super().asyncSetUp()
+            await self.bus.request_name_async("org.example.test", 0)
+
+        async def test_method_kwargs(self) -> None:
+            test_object, test_object_connection = initialize_object()
+
+            self.assertEqual(
+                'TEST',
+                await test_object_connection.upper('test'),
+            )

src/sdbus/unittest.py

@@ -0,0 +1,89 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+
+# Copyright (C) 2022 igo95862
+
+# This file is part of python-sdbus
+
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+
+# This library 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
+# Lesser General Public License for more details.
+
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+from __future__ import annotations
+
+from asyncio.subprocess import DEVNULL, create_subprocess_exec
+from os import environ, kill
+from pathlib import Path
+from signal import SIGTERM
+from tempfile import TemporaryDirectory
+from typing import ClassVar
+from unittest import IsolatedAsyncioTestCase
+
+from sdbus import sd_bus_open_user, set_default_bus
+
+dbus_config = '''
+<busconfig>
+  <type>session</type>
+  <pidfile>{pidfile_path}</pidfile>
+  <auth>EXTERNAL</auth>
+  <listen>unix:path={socket_path}</listen>
+  <policy context="default">
+    <allow send_destination="*" eavesdrop="true"/>
+    <allow eavesdrop="true"/>
+    <allow own="*"/>
+  </policy>
+</busconfig>
+'''
+
+
+class IsolatedDbusTestCase(IsolatedAsyncioTestCase):
+    dbus_executable_name: ClassVar[str] = 'dbus-daemon'
+
+    async def asyncSetUp(self) -> None:
+        self.temp_dir = TemporaryDirectory()
+        self.temp_dir_path = Path(self.temp_dir.name)
+
+        self.dbus_socket_path = self.temp_dir_path / 'test_dbus.socket'
+        self.pid_path = self.temp_dir_path / 'dbus.pid'
+
+        self.dbus_config_file = self.temp_dir_path / 'dbus.config'
+
+        with open(self.dbus_config_file, mode='x') as conf_file:
+            conf_file.write(dbus_config.format(
+                socket_path=self.dbus_socket_path,
+                pidfile_path=self.pid_path))
+
+        self.dbus_process = await create_subprocess_exec(
+            self.dbus_executable_name,
+            '--config-file', self.dbus_config_file,
+            '--fork',
+            stdin=DEVNULL,
+        )
+        error_code = await self.dbus_process.wait()
+        if error_code != 0:
+            raise ChildProcessError('Failed to start dbus daemon')
+
+        self.old_session_bus_address = environ.get('DBUS_SESSION_BUS_ADDRESS')
+        environ[
+            'DBUS_SESSION_BUS_ADDRESS'] = f"unix:path={self.dbus_socket_path}"
+
+        self.bus = sd_bus_open_user()
+        set_default_bus(self.bus)
+
+    async def asyncTearDown(self) -> None:
+        with open(self.pid_path) as pid_file:
+            dbus_pid = int(pid_file.read())
+
+        kill(dbus_pid, SIGTERM)
+        self.temp_dir.cleanup()
+        environ.pop('DBUS_SESSION_BUS_ADDRESS')
+        if self.old_session_bus_address is not None:
+            environ['DBUS_SESSION_BUS_ADDRESS'] = self.old_session_bus_address

test/common_test_util.py

@@ -19,84 +19,7 @@
 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 from __future__ import annotations
 
-from asyncio.subprocess import DEVNULL, create_subprocess_exec
-from os import environ, kill
-from pathlib import Path
-from signal import SIGKILL
-from tempfile import TemporaryDirectory
-from unittest import IsolatedAsyncioTestCase, main
-
-from sdbus import sd_bus_open_user
-
-dbus_config = '''
-<!DOCTYPE busconfig PUBLIC
- "-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN"
- "http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
-<busconfig>
-
-  <!-- Our well-known bus type, do not change this -->
-  <type>session</type>
-
-
-  <!-- Write a pid file -->
-  <pidfile>{pidfile_path}</pidfile>
-
-  <!-- Only allow socket-credentials-based authentication -->
-  <auth>EXTERNAL</auth>
-
-  <!-- Only listen on a local socket. (abstract=/path/to/socket
-       means use abstract namespace, don't really create filesystem
-       file; only Linux supports this. Use path=/whatever on other
-       systems.) -->
-  <listen>unix:path={socket_path}</listen>
-
-  <policy context="default">
-    <!-- Allow everything to be sent -->
-    <allow send_destination="*" eavesdrop="true"/>
-    <!-- Allow everything to be received -->
-    <allow eavesdrop="true"/>
-    <!-- Allow anyone to own anything -->
-    <allow own="*"/>
-  </policy>
-
-</busconfig>
-'''
-
-
-class TempDbusTest(IsolatedAsyncioTestCase):
-    async def asyncSetUp(self) -> None:
-        self.temp_dir = TemporaryDirectory()
-        self.temp_dir_path = Path(self.temp_dir.name)
-
-        self.dbus_socket_path = self.temp_dir_path / 'test_dbus.socket'
-        self.pid_path = self.temp_dir_path / 'dbus.pid'
-
-        self.dbus_config_file = self.temp_dir_path / 'dbus.config'
-
-        with open(self.dbus_config_file, mode='x') as conf_file:
-            conf_file.write(dbus_config.format(
-                socket_path=self.dbus_socket_path,
-                pidfile_path=self.pid_path))
-
-        self.dbus_process = await create_subprocess_exec(
-            '/usr/bin/dbus-daemon',
-            f'--config-file={self.dbus_config_file}',
-            '--fork',
-            stdin=DEVNULL,
-        )
-        await self.dbus_process.wait()
-        environ[
-            'DBUS_SESSION_BUS_ADDRESS'] = f"unix:path={self.dbus_socket_path}"
-
-        self.bus = sd_bus_open_user()
-
-    async def asyncTearDown(self) -> None:
-        with open(self.pid_path) as pid_file:
-            dbus_pid = int(pid_file.read())
-
-        kill(dbus_pid, SIGKILL)
-        self.temp_dir.cleanup()
-        environ.pop('DBUS_SESSION_BUS_ADDRESS')
+from unittest import main
 
 
 def mem_test() -> None:

test/leak_tests.py

@@ -32,7 +32,8 @@
 from typing import List, cast
 from unittest import SkipTest
 
-from .common_test_util import TempDbusTest
+from sdbus.unittest import IsolatedDbusTestCase
+
 from .test_read_write_dbus_types import TestDbusTypes
 from .test_sd_bus_async import TestPing, TestProxy, initialize_object
 
@@ -48,7 +49,7 @@ def leak_test_enabled() -> None:
         )
 
 
-class LeakTests(TempDbusTest):
+class LeakTests(IsolatedDbusTestCase):
     def setUp(self) -> None:
         super().setUp()
         self.start_mem = getrusage(RUSAGE_SELF).ru_maxrss
@@ -122,7 +123,7 @@ async def test_single_object(self) -> None:
         leak_test_enabled()
         await self.bus.request_name_async("org.example.test", 0)
 
-        test_object, test_object_connection = initialize_object(self.bus)
+        test_object, test_object_connection = initialize_object()
 
         i = 0
         num_of_iterations = 10_000

test/test_low_level_api.py

@@ -22,11 +22,10 @@
 from unittest import main
 
 from sdbus.sd_bus_internals import SdBus
+from sdbus.unittest import IsolatedDbusTestCase
 
-from .common_test_util import TempDbusTest
 
-
-class TestDbusTypes(TempDbusTest):
+class TestDbusTypes(IsolatedDbusTestCase):
     def test_init_bus(self) -> None:
         SdBus()
 

test/test_proxies.py

@@ -21,12 +21,11 @@
 
 from unittest import main
 
+from sdbus.unittest import IsolatedDbusTestCase
 from sdbus_async.dbus_daemon import FreedesktopDbus
 
-from .common_test_util import TempDbusTest
 
-
-class TestFreedesktopDbus(TempDbusTest):
+class TestFreedesktopDbus(IsolatedDbusTestCase):
     async def test_connection(self) -> None:
         dbus_object = FreedesktopDbus(self.bus)
 

test/test_read_write_dbus_types.py

@@ -23,11 +23,10 @@
 from unittest import main
 
 from sdbus.sd_bus_internals import SdBus, SdBusMessage
+from sdbus.unittest import IsolatedDbusTestCase
 
 from sdbus import SdBusLibraryError
 
-from .common_test_util import TempDbusTest
-
 
 def create_message(bus: SdBus) -> SdBusMessage:
     return bus.new_method_call_message(
@@ -37,7 +36,7 @@ def create_message(bus: SdBus) -> SdBusMessage:
         'GetUnit')
 
 
-class TestDbusTypes(TempDbusTest):
+class TestDbusTypes(IsolatedDbusTestCase):
     async def asyncSetUp(self) -> None:
         await super().asyncSetUp()