diff --git a/examples/dbus-service.py b/examples/dbus-service.py new file mode 100644 index 0000000..bc4852f --- /dev/null +++ b/examples/dbus-service.py @@ -0,0 +1,46 @@ +# systemd_ctypes +# +# Copyright (C) 2023 Martin Pitt +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program 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 General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +# once this runs, test with: +# busctl --user call com.example.Test / com.example.Test HelloWorld s "World" + +import asyncio + +from systemd_ctypes import bus, run_async + + +class com_example_Test(bus.Object): + @bus.Interface.Method('s', 's') + def hello_world(self, name): + return f'Hello {name}!' + + +async def main(): + user_bus = bus.Bus.default_user() + + test_object = com_example_Test() + test_slot = user_bus.add_object('/', test_object) + + user_bus.request_name('com.example.Test', 0) + + await asyncio.sleep(30) + + user_bus.release_name('com.example.Test') + del test_slot + + +run_async(main()) diff --git a/src/systemd_ctypes/bus.py b/src/systemd_ctypes/bus.py index 706369c..f90672d 100644 --- a/src/systemd_ctypes/bus.py +++ b/src/systemd_ctypes/bus.py @@ -17,6 +17,7 @@ import asyncio import base64 +import enum import functools import itertools import logging @@ -308,6 +309,12 @@ class Bus(sd.bus): _default_system = None _default_user = None + class NameFlags(enum.IntFlag): + DEFAULT = 0 + REPLACE_EXISTING = 1 << 0 + ALLOW_REPLACEMENT = 1 << 1 + QUEUE = 1 << 2 + @staticmethod def new(fd=None, address=None, bus_client=False, server=False, start=True, attach_event=True): bus = Bus() diff --git a/src/systemd_ctypes/libsystemd.py b/src/systemd_ctypes/libsystemd.py index eb10aa7..5cb2374 100644 --- a/src/systemd_ctypes/libsystemd.py +++ b/src/systemd_ctypes/libsystemd.py @@ -153,6 +153,8 @@ def wrapper(self, char): (instancemethod, negative_errno, 'message_new_method_call', [POINTER(sd.bus_message_p), utf8, utf8, utf8, utf8]), (instancemethod, negative_errno, 'message_new_signal', [POINTER(sd.bus_message_p), utf8, utf8, utf8]), (instancemethod, negative_errno, 'new', [POINTER(sd.bus_p)]), + (instancemethod, negative_errno, 'release_name', [utf8]), + (instancemethod, negative_errno, 'request_name', [utf8, c_uint64]), (instancemethod, negative_errno, 'set_address', [utf8]), (instancemethod, negative_errno, 'set_bus_client', [boolint]), (instancemethod, negative_errno, 'set_fd', [c_int, c_int]), diff --git a/test/test_basic.py b/test/test_basic.py index c4b54b7..f5b6a0d 100644 --- a/test/test_basic.py +++ b/test/test_basic.py @@ -20,12 +20,18 @@ import dbusmock import systemd_ctypes -from systemd_ctypes import introspection +from systemd_ctypes import introspection, bus TEST_ADDR = ('org.freedesktop.Test', '/', 'org.freedesktop.Test.Main') +class Test_Greeter(bus.Object): + @bus.Interface.Method('s', 's') + def say_hello(self, name): + return f'Hello {name}!' + + class TestAPI(dbusmock.DBusTestCase): @classmethod def setUpClass(cls): @@ -237,6 +243,46 @@ def test_introspect(self): } self.assertEqual(parsed['org.freedesktop.Test.Main'], expected) + def check_iface_sayhello(self, service_name): + message = self.bus_user.message_new_method_call( + service_name, '/', 'Test.Greeter', + 'SayHello', 's', 'world') + self.assertEqual(self.async_call(message).get_body(), ('Hello world!',)) + + def test_service(self): + test_object = Test_Greeter() + test_slot = self.bus_user.add_object('/', test_object) + self.bus_user.request_name('com.example.Test', bus.Bus.NameFlags.DEFAULT) + + self.check_iface_sayhello('com.example.Test') + + self.bus_user.release_name('com.example.Test') + del test_slot + + def test_service_replace(self): + test_object = Test_Greeter() + test_slot = self.bus_user.add_object('/', test_object) + self.bus_user.request_name(TEST_ADDR[0], bus.Bus.NameFlags.REPLACE_EXISTING) + + self.check_iface_sayhello(TEST_ADDR[0]) + + self.bus_user.release_name(TEST_ADDR[0]) + del test_slot + + def test_request_name_errors(self): + # name already exists + self.assertRaises(FileExistsError, self.bus_user.request_name, TEST_ADDR[0], bus.Bus.NameFlags.DEFAULT) + + # invalid name + self.assertRaisesRegex(OSError, '.*Invalid argument', + self.bus_user.request_name, '', bus.Bus.NameFlags.DEFAULT) + + # invalid flag + self.assertRaisesRegex(OSError, '.*Invalid argument', self.bus_user.request_name, TEST_ADDR[0], 0xFF) + + # name not taken + self.assertRaises(ProcessLookupError, self.bus_user.release_name, 'com.example.NotThis') + if __name__ == '__main__': unittest.main()