reliability: guard BlueZ base async lifecycles

blueman/bluez/Base.py

@@ -1,19 +1,20 @@
-from typing import Any
+import logging
 from collections.abc import Callable
-from blueman.bluemantyping import GSignals, ObjectPath
+from typing import Any
+from weakref import WeakValueDictionary
 
-from gi.repository import Gio, GLib, GObject
+from blueman.bluez.errors import BluezDBusException, parse_dbus_error
+from blueman.bluemantyping import GSignals, ObjectPath
+from gi.repository import GLib, GObject, Gio
 from gi.types import GObjectMeta
-from blueman.bluez.errors import parse_dbus_error, BluezDBusException
-import logging
 
 DBUS_TIMEOUT = 10 * 1_000
 
 
 class BaseMeta(GObjectMeta):
     def __call__(cls, *args: object, **kwargs: str) -> "Base":
-        if not hasattr(cls, "__instances__"):
-            cls.__instances__: dict[str, "Base"] = {}
+        if "__instances__" not in cls.__dict__:
+            cls.__instances__: WeakValueDictionary[str, "Base"] = WeakValueDictionary()
 
         path = kwargs.get('obj_path')
         if path is None:
@@ -36,7 +37,7 @@ class Base(GObject.Object, metaclass=BaseMeta):
     __gsignals__: GSignals = {
         'property-changed': (GObject.SignalFlags.NO_HOOKS, None, (str, object, str))
     }
-    __instances__: dict[str, "Base"]
+    __instances__: WeakValueDictionary[str, "Base"]
 
     _interface_name: str
 
@@ -61,6 +62,7 @@ def __init__(self, *, obj_path: ObjectPath):
         self.__fallback = {'Icon': 'blueman', 'Class': 0, 'Appearance': 0}
 
         self.__variant_map = {str: 's', int: 'u', bool: 'b'}
+        self.__set_cancellables: set[Gio.Cancellable] = set()
 
     def _properties_changed(self, _proxy: Gio.DBusProxy, changed_properties: GLib.Variant,
                             invalidated_properties: list[str]) -> None:
@@ -115,14 +117,27 @@ def get(self, name: str) -> Any:
             else:
                 raise parse_dbus_error(e)
 
-    def set(self, name: str, value: str | int | bool) -> None:
+    def set(self, name: str, value: str | int | bool, cancellable: Gio.Cancellable | None = None) -> None:
+        def on_set_done(proxy: Gio.DBusProxy, result: Gio.AsyncResult, pending: Gio.Cancellable) -> None:
+            self.__set_cancellables.discard(pending)
+            try:
+                proxy.call_finish(result)
+            except GLib.Error:
+                logging.exception(f"Unhandled error for {proxy.get_interface_name()}.Properties.Set")
+
         v = GLib.Variant(self.__variant_map[type(value)], value)
         param = GLib.Variant('(ssv)', (self._interface_name, name, v))
+        if cancellable is None:
+            cancellable = Gio.Cancellable()
+
+        self.__set_cancellables.add(cancellable)
         self.__proxy.call('org.freedesktop.DBus.Properties.Set',
                           param,
                           Gio.DBusCallFlags.NONE,
                           DBUS_TIMEOUT,
-                          None)
+                          cancellable,
+                          on_set_done,
+                          cancellable)
 
     def get_object_path(self) -> ObjectPath:
         return ObjectPath(self.__proxy.get_object_path())
@@ -145,7 +160,13 @@ def get_properties(self) -> dict[str, Any]:
         return props
 
     def destroy(self) -> None:
+        for cancellable in self.__set_cancellables:
+            cancellable.cancel()
+        self.__set_cancellables.clear()
         if self.__proxy:
+            object_path = self.get_object_path()
+            if self.__class__.__instances__.get(object_path) is self:
+                del self.__class__.__instances__[object_path]
             del self.__proxy
 
     def __getitem__(self, key: str) -> Any:

test/bluez/Makefile.am

@@ -3,5 +3,6 @@ SUBDIRS =   \
 
 EXTRA_DIST =    \
     __init__.py \
+    test_base.py \
     test_imports.py \
     test_manager.py

test/bluez/test_base.py

@@ -0,0 +1,97 @@
+import gc
+import weakref
+from unittest import TestCase
+from unittest.mock import patch
+
+from blueman.bluemantyping import ObjectPath
+from blueman.bluez.Base import Base
+from gi.repository import Gio
+
+
+class FakeProxy:
+    def __init__(self, object_path: str = "/org/test/fake") -> None:
+        self.calls = []
+        self.object_path = object_path
+
+    def connect(self, *_args):
+        return 1
+
+    def call(self, *args):
+        self.calls.append(args)
+
+    def call_finish(self, _result):
+        return None
+
+    def get_interface_name(self):
+        return "org.test.Interface"
+
+    def get_object_path(self):
+        return self.object_path
+
+
+class TestBaseObject(Base):
+    _interface_name = "org.test.Interface"
+
+
+class OtherTestBaseObject(Base):
+    _interface_name = "org.test.OtherInterface"
+
+
+class TestBaseConcurrency(TestCase):
+    def setUp(self):
+        for cls in (TestBaseObject, OtherTestBaseObject):
+            if hasattr(cls, "__instances__"):
+                cls.__instances__.clear()
+
+    def test_instance_cache_uses_weak_values(self):
+        with patch("blueman.bluez.Base.Gio.DBusProxy.new_for_bus_sync", return_value=FakeProxy()):
+            obj = TestBaseObject(obj_path=ObjectPath("/org/test/device0"))
+            self.assertIs(TestBaseObject(obj_path=ObjectPath("/org/test/device0")), obj)
+            obj_ref = weakref.ref(obj)
+
+            del obj
+            gc.collect()
+
+        self.assertIsNone(obj_ref())
+
+    def test_instance_cache_is_per_subclass(self):
+        with patch("blueman.bluez.Base.Gio.DBusProxy.new_for_bus_sync", return_value=FakeProxy()):
+            obj = TestBaseObject(obj_path=ObjectPath("/org/test/shared"))
+            other = OtherTestBaseObject(obj_path=ObjectPath("/org/test/shared"))
+
+            self.assertIs(TestBaseObject(obj_path=ObjectPath("/org/test/shared")), obj)
+            self.assertIs(OtherTestBaseObject(obj_path=ObjectPath("/org/test/shared")), other)
+
+    def test_destroy_removes_instance_from_cache(self):
+        with patch("blueman.bluez.Base.Gio.DBusProxy.new_for_bus_sync") as new_proxy:
+            new_proxy.side_effect = [FakeProxy("/org/test/destroyed"), FakeProxy("/org/test/destroyed")]
+            obj = TestBaseObject(obj_path=ObjectPath("/org/test/destroyed"))
+
+            obj.destroy()
+            next_obj = TestBaseObject(obj_path=ObjectPath("/org/test/destroyed"))
+
+        self.assertIsNot(obj, next_obj)
+
+    def test_set_passes_cancellable_to_dbus_call(self):
+        proxy = FakeProxy()
+        cancellable = Gio.Cancellable()
+
+        with patch("blueman.bluez.Base.Gio.DBusProxy.new_for_bus_sync", return_value=proxy):
+            obj = TestBaseObject(obj_path=ObjectPath("/org/test/device1"))
+            obj.set("Powered", True, cancellable=cancellable)
+
+        call = proxy.calls[0]
+        self.assertEqual(call[0], "org.freedesktop.DBus.Properties.Set")
+        self.assertIs(call[4], cancellable)
+        self.assertIs(call[6], cancellable)
+
+    def test_destroy_cancels_pending_set_calls(self):
+        cancellable = Gio.Cancellable()
+
+        with patch("blueman.bluez.Base.Gio.DBusProxy.new_for_bus_sync", return_value=FakeProxy()):
+            obj = TestBaseObject(obj_path=ObjectPath("/org/test/device2"))
+            obj.set("Powered", True, cancellable=cancellable)
+
+            obj.destroy()
+
+        self.assertTrue(cancellable.is_cancelled())