Skip to content

Commit b1db762

Browse files
igo95862igo95862
committed
Added sdbus.unittest.IsolatedDbusTestCase
Extension of `unittest.IsolatedAsyncioTestCase` from standard library. Allows to run asyncio tests in isolated D-Bus instance.
1 parent 2776196 commit b1db762

File tree

10 files changed

+186
-123
lines changed

10 files changed

+186
-123
lines changed

‎docs/index.rst‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,9 +39,10 @@ If you are unfamiliar with D-Bus you might want to read following pages:
3939
asyncio_api
4040
exceptions
4141
examples
42-
autodoc
43-
code_generator
4442
proxies
43+
code_generator
44+
autodoc
45+
unittest
4546
api_index
4647

4748

‎docs/unittest.rst‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
Unit testing
2+
============
3+
4+
Python-sdbus provides several utilities to enable unit testing.
5+
6+
.. py:currentmodule:: sdbus.unittest
7+
8+
.. py:class:: IsolatedDbusTestCase
9+
10+
Extension of `unittest.IsolatedAsyncioTestCase
11+
<https://docs.python.org/3/library/unittest.html#unittest.IsolatedAsyncioTestCase>`__
12+
from standard library.
13+
14+
Creates an isolated instance of session D-Bus. The D-Bus will be closed
15+
and cleaned up after tests are finished.
16+
17+
Requires ``dbus-daemon`` executable be installed.
18+
19+
.. py:attribute:: bus
20+
:type: SdBus
21+
22+
Bus instance connected to isolated D-Bus environment.
23+
24+
It is also set as a default bus.
25+
26+
27+
Usage example: ::
28+
29+
from sdbus import DbusInterfaceCommonAsync, dbus_method_async
30+
from sdbus.unittest import IsolatedDbusTestCase
31+
32+
class TestInterface(DbusInterfaceCommonAsync,
33+
interface_name='org.test.test',
34+
):
35+
36+
@dbus_method_async("s", "s")
37+
async def upper(self, string: str) -> str:
38+
"""Uppercase the input"""
39+
return string.upper()
40+
41+
def initialize_object() -> Tuple[TestInterface, TestInterface]:
42+
test_object = TestInterface()
43+
test_object.export_to_dbus('/')
44+
45+
test_object_connection = TestInterface.new_proxy(
46+
"org.example.test", '/')
47+
48+
return test_object, test_object_connection
49+
50+
51+
class TestProxy(IsolatedDbusTestCase):
52+
async def asyncSetUp(self) -> None:
53+
await super().asyncSetUp()
54+
await self.bus.request_name_async("org.example.test", 0)
55+
56+
async def test_method_kwargs(self) -> None:
57+
test_object, test_object_connection = initialize_object()
58+
59+
self.assertEqual(
60+
'TEST',
61+
await test_object_connection.upper('test'),
62+
)

‎src/sdbus/unittest.py‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
# SPDX-License-Identifier: LGPL-2.1-or-later
2+
3+
# Copyright (C) 2022 igo95862
4+
5+
# This file is part of python-sdbus
6+
7+
# This library is free software; you can redistribute it and/or
8+
# modify it under the terms of the GNU Lesser General Public
9+
# License as published by the Free Software Foundation; either
10+
# version 2.1 of the License, or (at your option) any later version.
11+
12+
# This library is distributed in the hope that it will be useful,
13+
# but WITHOUT ANY WARRANTY; without even the implied warranty of
14+
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15+
# Lesser General Public License for more details.
16+
17+
# You should have received a copy of the GNU Lesser General Public
18+
# License along with this library; if not, write to the Free Software
19+
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
20+
from __future__ import annotations
21+
22+
from asyncio.subprocess import DEVNULL, create_subprocess_exec
23+
from os import environ, kill
24+
from pathlib import Path
25+
from signal import SIGTERM
26+
from tempfile import TemporaryDirectory
27+
from typing import ClassVar
28+
from unittest import IsolatedAsyncioTestCase
29+
30+
from sdbus import sd_bus_open_user, set_default_bus
31+
32+
dbus_config = '''
33+
<busconfig>
34+
<type>session</type>
35+
<pidfile>{pidfile_path}</pidfile>
36+
<auth>EXTERNAL</auth>
37+
<listen>unix:path={socket_path}</listen>
38+
<policy context="default">
39+
<allow send_destination="*" eavesdrop="true"/>
40+
<allow eavesdrop="true"/>
41+
<allow own="*"/>
42+
</policy>
43+
</busconfig>
44+
'''
45+
46+
47+
class IsolatedDbusTestCase(IsolatedAsyncioTestCase):
48+
dbus_executable_name: ClassVar[str] = 'dbus-daemon'
49+
50+
async def asyncSetUp(self) -> None:
51+
self.temp_dir = TemporaryDirectory()
52+
self.temp_dir_path = Path(self.temp_dir.name)
53+
54+
self.dbus_socket_path = self.temp_dir_path / 'test_dbus.socket'
55+
self.pid_path = self.temp_dir_path / 'dbus.pid'
56+
57+
self.dbus_config_file = self.temp_dir_path / 'dbus.config'
58+
59+
with open(self.dbus_config_file, mode='x') as conf_file:
60+
conf_file.write(dbus_config.format(
61+
socket_path=self.dbus_socket_path,
62+
pidfile_path=self.pid_path))
63+
64+
self.dbus_process = await create_subprocess_exec(
65+
self.dbus_executable_name,
66+
'--config-file', self.dbus_config_file,
67+
'--fork',
68+
stdin=DEVNULL,
69+
)
70+
error_code = await self.dbus_process.wait()
71+
if error_code != 0:
72+
raise ChildProcessError('Failed to start dbus daemon')
73+
74+
self.old_session_bus_address = environ.get('DBUS_SESSION_BUS_ADDRESS')
75+
environ[
76+
'DBUS_SESSION_BUS_ADDRESS'] = f"unix:path={self.dbus_socket_path}"
77+
78+
self.bus = sd_bus_open_user()
79+
set_default_bus(self.bus)
80+
81+
async def asyncTearDown(self) -> None:
82+
with open(self.pid_path) as pid_file:
83+
dbus_pid = int(pid_file.read())
84+
85+
kill(dbus_pid, SIGTERM)
86+
self.temp_dir.cleanup()
87+
environ.pop('DBUS_SESSION_BUS_ADDRESS')
88+
if self.old_session_bus_address is not None:
89+
environ['DBUS_SESSION_BUS_ADDRESS'] = self.old_session_bus_address

‎test/common_test_util.py‎

Lines changed: 1 addition & 78 deletions
Original file line numberDiff line numberDiff line change
@@ -19,84 +19,7 @@
1919
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
2020
from __future__ import annotations
2121

22-
from asyncio.subprocess import DEVNULL, create_subprocess_exec
23-
from os import environ, kill
24-
from pathlib import Path
25-
from signal import SIGKILL
26-
from tempfile import TemporaryDirectory
27-
from unittest import IsolatedAsyncioTestCase, main
28-
29-
from sdbus import sd_bus_open_user
30-
31-
dbus_config = '''
32-
<!DOCTYPE busconfig PUBLIC
33-
"-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN"
34-
"http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
35-
<busconfig>
36-
37-
<!-- Our well-known bus type, do not change this -->
38-
<type>session</type>
39-
40-
41-
<!-- Write a pid file -->
42-
<pidfile>{pidfile_path}</pidfile>
43-
44-
<!-- Only allow socket-credentials-based authentication -->
45-
<auth>EXTERNAL</auth>
46-
47-
<!-- Only listen on a local socket. (abstract=/path/to/socket
48-
means use abstract namespace, don't really create filesystem
49-
file; only Linux supports this. Use path=/whatever on other
50-
systems.) -->
51-
<listen>unix:path={socket_path}</listen>
52-
53-
<policy context="default">
54-
<!-- Allow everything to be sent -->
55-
<allow send_destination="*" eavesdrop="true"/>
56-
<!-- Allow everything to be received -->
57-
<allow eavesdrop="true"/>
58-
<!-- Allow anyone to own anything -->
59-
<allow own="*"/>
60-
</policy>
61-
62-
</busconfig>
63-
'''
64-
65-
66-
class TempDbusTest(IsolatedAsyncioTestCase):
67-
async def asyncSetUp(self) -> None:
68-
self.temp_dir = TemporaryDirectory()
69-
self.temp_dir_path = Path(self.temp_dir.name)
70-
71-
self.dbus_socket_path = self.temp_dir_path / 'test_dbus.socket'
72-
self.pid_path = self.temp_dir_path / 'dbus.pid'
73-
74-
self.dbus_config_file = self.temp_dir_path / 'dbus.config'
75-
76-
with open(self.dbus_config_file, mode='x') as conf_file:
77-
conf_file.write(dbus_config.format(
78-
socket_path=self.dbus_socket_path,
79-
pidfile_path=self.pid_path))
80-
81-
self.dbus_process = await create_subprocess_exec(
82-
'/usr/bin/dbus-daemon',
83-
f'--config-file={self.dbus_config_file}',
84-
'--fork',
85-
stdin=DEVNULL,
86-
)
87-
await self.dbus_process.wait()
88-
environ[
89-
'DBUS_SESSION_BUS_ADDRESS'] = f"unix:path={self.dbus_socket_path}"
90-
91-
self.bus = sd_bus_open_user()
92-
93-
async def asyncTearDown(self) -> None:
94-
with open(self.pid_path) as pid_file:
95-
dbus_pid = int(pid_file.read())
96-
97-
kill(dbus_pid, SIGKILL)
98-
self.temp_dir.cleanup()
99-
environ.pop('DBUS_SESSION_BUS_ADDRESS')
22+
from unittest import main
10023

10124

10225
def mem_test() -> None:

‎test/leak_tests.py‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,8 @@
3232
from typing import List, cast
3333
from unittest import SkipTest
3434

35-
from .common_test_util import TempDbusTest
35+
from sdbus.unittest import IsolatedDbusTestCase
36+
3637
from .test_read_write_dbus_types import TestDbusTypes
3738
from .test_sd_bus_async import TestPing, TestProxy, initialize_object
3839

@@ -48,7 +49,7 @@ def leak_test_enabled() -> None:
4849
)
4950

5051

51-
class LeakTests(TempDbusTest):
52+
class LeakTests(IsolatedDbusTestCase):
5253
def setUp(self) -> None:
5354
super().setUp()
5455
self.start_mem = getrusage(RUSAGE_SELF).ru_maxrss
@@ -122,7 +123,7 @@ async def test_single_object(self) -> None:
122123
leak_test_enabled()
123124
await self.bus.request_name_async("org.example.test", 0)
124125

125-
test_object, test_object_connection = initialize_object(self.bus)
126+
test_object, test_object_connection = initialize_object()
126127

127128
i = 0
128129
num_of_iterations = 10_000

‎test/test_low_level_api.py‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,11 +22,10 @@
2222
from unittest import main
2323

2424
from sdbus.sd_bus_internals import SdBus
25+
from sdbus.unittest import IsolatedDbusTestCase
2526

26-
from .common_test_util import TempDbusTest
2727

28-
29-
class TestDbusTypes(TempDbusTest):
28+
class TestDbusTypes(IsolatedDbusTestCase):
3029
def test_init_bus(self) -> None:
3130
SdBus()
3231

‎test/test_proxies.py‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,12 +21,11 @@
2121

2222
from unittest import main
2323

24+
from sdbus.unittest import IsolatedDbusTestCase
2425
from sdbus_async.dbus_daemon import FreedesktopDbus
2526

26-
from .common_test_util import TempDbusTest
2727

28-
29-
class TestFreedesktopDbus(TempDbusTest):
28+
class TestFreedesktopDbus(IsolatedDbusTestCase):
3029
async def test_connection(self) -> None:
3130
dbus_object = FreedesktopDbus(self.bus)
3231

‎test/test_read_write_dbus_types.py‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,11 +23,10 @@
2323
from unittest import main
2424

2525
from sdbus.sd_bus_internals import SdBus, SdBusMessage
26+
from sdbus.unittest import IsolatedDbusTestCase
2627

2728
from sdbus import SdBusLibraryError
2829

29-
from .common_test_util import TempDbusTest
30-
3130

3231
def create_message(bus: SdBus) -> SdBusMessage:
3332
return bus.new_method_call_message(
@@ -37,7 +36,7 @@ def create_message(bus: SdBus) -> SdBusMessage:
3736
'GetUnit')
3837

3938

40-
class TestDbusTypes(TempDbusTest):
39+
class TestDbusTypes(IsolatedDbusTestCase):
4140
async def asyncSetUp(self) -> None:
4241
await super().asyncSetUp()
4342

0 commit comments

Comments
 (0)