bus: Bind name owning/releasing

This is the synchronous API, which should usually be good enough (not
much else can happen in parallel anyway), and easier to use than the
async one.

Add correspoinding IntFlag for SD_BUS_NAME_* constants. Introduce an
additional `DEFAULT` value, to make the API calls nicer to read.

examples/dbus-service.py

@@ -0,0 +1,46 @@
+# systemd_ctypes
+#
+# Copyright (C) 2023 Martin Pitt <mpitt@redhat.com>
+#
+# 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 <http://www.gnu.org/licenses/>.
+
+# 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())

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

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]),

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