xibus

experimental pure python async D-Bus library
git clone https://git.ce9e.org/xibus.git

commit
e172df4b756c48c363c6413bf2edc64f9c8d4530
parent
d1b1b6d0956f81cadd0c323ead6e81e4dcd24eb6
Author
Tobias Bengfort <tobias.bengfort@posteo.de>
Date
2026-02-16 09:15
add some client tests

Diffstat

A tests/test_client.py 86 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

1 files changed, 86 insertions, 0 deletions


diff --git a/tests/test_client.py b/tests/test_client.py

@@ -0,0 +1,86 @@
   -1     1 import unittest
   -1     2 from unittest.mock import ANY
   -1     3 
   -1     4 from xibus import DBusError
   -1     5 from xibus import get_client
   -1     6 
   -1     7 
   -1     8 class TestCall(unittest.IsolatedAsyncioTestCase):
   -1     9     async def test_call(self):
   -1    10         async with get_client('session') as client:
   -1    11             response = await client.call(
   -1    12                 'org.freedesktop.DBus',
   -1    13                 '/org/freedesktop/DBus',
   -1    14                 'org.freedesktop.DBus',
   -1    15                 'ListNames',
   -1    16                 (),
   -1    17                 '',
   -1    18             )
   -1    19             self.assertIn('org.freedesktop.DBus', response)
   -1    20 
   -1    21     async def test_fail_on_double_hello(self):
   -1    22         async with get_client('session') as client:
   -1    23             with self.assertRaises(DBusError) as ctx:
   -1    24                 await client.call(
   -1    25                     'org.freedesktop.DBus',
   -1    26                     '/org/freedesktop/DBus',
   -1    27                     'org.freedesktop.DBus',
   -1    28                     'Hello',
   -1    29                     (),
   -1    30                     '',
   -1    31                 )
   -1    32             self.assertEqual(str(ctx.exception), 'org.freedesktop.DBus.Error.Failed')
   -1    33 
   -1    34     async def test_proxy_call(self):
   -1    35         async with get_client('session') as client:
   -1    36             response = await client.bus.call('ListNames', (), '')
   -1    37             self.assertIn('org.freedesktop.DBus', response)
   -1    38 
   -1    39     async def test_magic_call(self):
   -1    40         async with get_client('session') as client:
   -1    41             response = await client.call(
   -1    42                 'org.freedesktop.DBus', None, None, 'ListNames', ()
   -1    43             )
   -1    44             self.assertIn('org.freedesktop.DBus', response)
   -1    45 
   -1    46 
   -1    47 class TestProperties(unittest.IsolatedAsyncioTestCase):
   -1    48     async def test_get_property(self):
   -1    49         async with get_client('session') as client:
   -1    50             response = await client.get_property(
   -1    51                 'org.freedesktop.DBus',
   -1    52                 '/org/freedesktop/DBus',
   -1    53                 'org.freedesktop.DBus',
   -1    54                 'Features',
   -1    55             )
   -1    56             self.assertIn('ActivatableServicesChanged', response)
   -1    57 
   -1    58     async def test_get_property_error(self):
   -1    59         async with get_client('session') as client:
   -1    60             with self.assertRaises(DBusError) as ctx:
   -1    61                 await client.get_property(
   -1    62                     'org.freedesktop.DBus',
   -1    63                     '/org/freedesktop/DBus',
   -1    64                     'org.freedesktop.DBus',
   -1    65                     'DoesNotExist',
   -1    66                 )
   -1    67             self.assertEqual(
   -1    68                 str(ctx.exception), 'org.freedesktop.DBus.Error.UnknownProperty'
   -1    69             )
   -1    70 
   -1    71     async def test_proxy_get_property(self):
   -1    72         async with get_client('session') as client:
   -1    73             response = await client.bus.get_property('Features')
   -1    74             self.assertIn('ActivatableServicesChanged', response)
   -1    75 
   -1    76 
   -1    77 class TestSignals(unittest.IsolatedAsyncioTestCase):
   -1    78     async def test_subscribe_signal(self):
   -1    79         async with get_client('session') as client:
   -1    80             async with client.bus.subscribe_signal('NameOwnerChanged') as queue:
   -1    81                 async with client.acquire_name('xibus.test'):
   -1    82                     pass
   -1    83 
   -1    84                 i = aiter(queue)
   -1    85                 self.assertEqual(await anext(i), ('xibus.test', '', ANY))
   -1    86                 self.assertEqual(await anext(i), ('xibus.test', ANY, ''))