Skip to content

Commit b6a1974

Browse files
igo95862igo95862
committed
Document IsolatedDbusTestCase.assertDbusSignalEmits
Also the recorder object now only has the `.output` property. The tests can call the assert methods against it to compare the emitted data.
1 parent 72e9d52 commit b6a1974

File tree

3 files changed

+66
-46
lines changed

3 files changed

+66
-46
lines changed

‎docs/unittest.rst‎

Lines changed: 55 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -16,47 +16,76 @@ Python-sdbus provides several utilities to enable unit testing.
1616

1717
Requires ``dbus-daemon`` executable be installed.
1818

19+
Example::
20+
21+
from sdbus import DbusInterfaceCommonAsync, dbus_method_async
22+
from sdbus.unittest import IsolatedDbusTestCase
23+
24+
class TestInterface(DbusInterfaceCommonAsync,
25+
interface_name='org.test.test',
26+
):
27+
28+
@dbus_method_async("s", "s")
29+
async def upper(self, string: str) -> str:
30+
"""Uppercase the input"""
31+
return string.upper()
32+
33+
def initialize_object() -> Tuple[TestInterface, TestInterface]:
34+
test_object = TestInterface()
35+
test_object.export_to_dbus('/')
36+
37+
test_object_connection = TestInterface.new_proxy(
38+
"org.example.test", '/')
39+
40+
return test_object, test_object_connection
41+
42+
43+
class TestProxy(IsolatedDbusTestCase):
44+
async def asyncSetUp(self) -> None:
45+
await super().asyncSetUp()
46+
await self.bus.request_name_async("org.example.test", 0)
47+
48+
async def test_method_kwargs(self) -> None:
49+
test_object, test_object_connection = initialize_object()
50+
51+
self.assertEqual(
52+
'TEST',
53+
await test_object_connection.upper('test'),
54+
)
55+
1956
.. py:attribute:: bus
2057
:type: SdBus
2158

2259
Bus instance connected to isolated D-Bus environment.
2360

2461
It is also set as a default bus.
2562

63+
.. py:method:: assertDbusSignalEmits(signal, timeout=1)
64+
65+
Assert that a given signal was emitted at least once within the
66+
given timeout.
67+
68+
:param signal: D-Bus signal object. Can be a signal from either local or proxy object.
69+
:param Union[int, float] timeout: Maximum wait time until first captured signal.
2670

27-
Usage example: ::
71+
Should be used as an async context manager. The context manager exits as soon
72+
as first signal is captured.
2873

29-
from sdbus import DbusInterfaceCommonAsync, dbus_method_async
30-
from sdbus.unittest import IsolatedDbusTestCase
74+
The object returned by context manager has following attributes:
3175

32-
class TestInterface(DbusInterfaceCommonAsync,
33-
interface_name='org.test.test',
34-
):
76+
.. py:attribute:: output
77+
:type: List[Any]
3578

36-
@dbus_method_async("s", "s")
37-
async def upper(self, string: str) -> str:
38-
"""Uppercase the input"""
39-
return string.upper()
79+
List of captured data.
4080

41-
def initialize_object() -> Tuple[TestInterface, TestInterface]:
42-
test_object = TestInterface()
43-
test_object.export_to_dbus('/')
81+
Example::
4482

45-
test_object_connection = TestInterface.new_proxy(
46-
"org.example.test", '/')
83+
async with self.assertDbusSignalEmits(test_object.test_signal) as signal_record:
84+
test_object.test_signal.emit("test")
4785

48-
return test_object, test_object_connection
86+
self.assertEqual(["test"], signal_record.output)
4987

88+
*New in version 0.12.0.*
5089

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)
5590

56-
async def test_method_kwargs(self) -> None:
57-
test_object, test_object_connection = initialize_object()
5891

59-
self.assertEqual(
60-
'TEST',
61-
await test_object_connection.upper('test'),
62-
)

‎src/sdbus/unittest.py‎

Lines changed: 7 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -75,10 +75,8 @@
7575
class DbusSignalRecorderBase:
7676
def __init__(
7777
self,
78-
testcase: IsolatedDbusTestCase,
7978
timeout: Union[int, float],
8079
):
81-
self._testcase = testcase
8280
self._timeout = timeout
8381
self._captured_data: List[Any] = []
8482
self._ready_event = Event()
@@ -114,25 +112,19 @@ def _callback(self, data: Any) -> None:
114112
self._captured_data.append(data)
115113
self._ready_event.set()
116114

117-
def assert_emitted_once_with(self, data: Any) -> None:
118-
captured_signals_num = len(self._captured_data)
119-
if captured_signals_num != 1:
120-
raise AssertionError(
121-
f"Expected one captured signal got {captured_signals_num}"
122-
)
123-
124-
self._testcase.assertEqual(self._captured_data[0], data)
115+
@property
116+
def output(self) -> List[Any]:
117+
return self._captured_data.copy()
125118

126119

127120
class DbusSignalRecorderRemote(DbusSignalRecorderBase):
128121
def __init__(
129122
self,
130-
testcase: IsolatedDbusTestCase,
131123
timeout: Union[int, float],
132124
bus: SdBus,
133125
remote_signal: DbusSignalAsyncProxyBind[Any],
134126
):
135-
super().__init__(testcase, timeout)
127+
super().__init__(timeout)
136128
self._bus = bus
137129
self._match_slot: Optional[SdBusSlot] = None
138130
self._remote_signal = remote_signal
@@ -161,11 +153,10 @@ async def __aexit__(
161153
class DbusSignalRecorderLocal(DbusSignalRecorderBase):
162154
def __init__(
163155
self,
164-
testcase: IsolatedDbusTestCase,
165156
timeout: Union[int, float],
166157
local_signal: DbusSignalAsyncLocalBind[Any],
167158
):
168-
super().__init__(testcase, timeout)
159+
super().__init__(timeout)
169160
self._local_signal_ref: weak_ref[DbusSignalAsync[Any]] = (
170161
weak_ref(local_signal.dbus_signal)
171162
)
@@ -234,8 +225,8 @@ def assertDbusSignalEmits(
234225
) -> AsyncContextManager[DbusSignalRecorderBase]:
235226

236227
if isinstance(signal, DbusSignalAsyncLocalBind):
237-
return DbusSignalRecorderLocal(self, timeout, signal)
228+
return DbusSignalRecorderLocal(timeout, signal)
238229
elif isinstance(signal, DbusSignalAsyncProxyBind):
239-
return DbusSignalRecorderRemote(self, timeout, self.bus, signal)
230+
return DbusSignalRecorderRemote(timeout, self.bus, signal)
240231
else:
241232
raise TypeError("Unknown or unsupported signal class.")

‎test/test_sdbus_async.py‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -484,8 +484,8 @@ async def test_signal(self) -> None:
484484
) as remote_signals_record:
485485
test_object.test_signal.emit(test_tuple)
486486

487-
local_signals_record.assert_emitted_once_with(test_tuple)
488-
remote_signals_record.assert_emitted_once_with(test_tuple)
487+
self.assertEqual([test_tuple], local_signals_record.output)
488+
self.assertEqual([test_tuple], remote_signals_record.output)
489489

490490
async def test_signal_catch_anywhere(self) -> None:
491491
test_object, test_object_connection = initialize_object()
@@ -779,8 +779,8 @@ async def test_empty_signal(self) -> None:
779779
) as remote_signals_record:
780780
test_object.empty_signal.emit(None)
781781

782-
local_signals_record.assert_emitted_once_with(None)
783-
remote_signals_record.assert_emitted_once_with(None)
782+
self.assertEqual([None], local_signals_record.output)
783+
self.assertEqual([None], remote_signals_record.output)
784784

785785
async def test_properties_changed(self) -> None:
786786
test_object, test_object_connection = initialize_object()

0 commit comments

Comments
 (0)