From 1eaba585364d07a989fb5c5e63e6c8aa126a460a Mon Sep 17 00:00:00 2001 From: Geraldo Netto Date: Mon, 15 Jun 2026 06:19:35 +0200 Subject: [PATCH 1/9] perf: optimize manager device row updates --- blueman/gui/manager/ManagerDeviceList.py | 150 ++++++++-------- test/gui/manager/test_manager_device_list.py | 172 +++++++++++++++++++ 2 files changed, 249 insertions(+), 73 deletions(-) create mode 100644 test/gui/manager/test_manager_device_list.py diff --git a/blueman/gui/manager/ManagerDeviceList.py b/blueman/gui/manager/ManagerDeviceList.py index 2514a6ec6..e5f92349e 100644 --- a/blueman/gui/manager/ManagerDeviceList.py +++ b/blueman/gui/manager/ManagerDeviceList.py @@ -1,6 +1,6 @@ from gettext import gettext as _ from typing import TYPE_CHECKING, Any, cast -from collections.abc import Callable +from collections.abc import Callable, Iterable, Mapping import html import logging import cairo @@ -81,7 +81,8 @@ def __init__(self, inst: "Blueman", adapter: str | None = None) -> None: self.props.has_tooltip = True self.Blueman = inst - self._monitored_devices: set[BtAddress] = set() + self._monitored_devices: dict[BtAddress, tuple[Gtk.TreeRowReference, conn_info]] = {} + self._power_levels_timer: int | None = None self.manager.connect_signal("battery-created", self.on_battery_created) self.manager.connect_signal("battery-removed", self.on_battery_removed) @@ -343,14 +344,17 @@ def make_display_name(alias: str, klass: int, address: BtAddress) -> str: return alias @staticmethod - def get_device_class(device: Device) -> str: - klass = get_minor_class(device['Class']) + def get_device_class(device: Device, properties: Mapping[str, Any] | None = None) -> str: + klass_id = device["Class"] if properties is None else cast(int, properties["Class"]) + klass = get_minor_class(klass_id) if klass != _("Uncategorized"): return klass else: - return get_major_class(device['Class']) + return get_major_class(klass_id) def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: + properties = device.get_properties() + if not self.get(tree_iter, "initial_anim")["initial_anim"]: assert self.liststore is not None child_path = self.liststore.get_path(tree_iter) @@ -370,53 +374,43 @@ def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: else: self.set(tree_iter, initial_anim=False) - has_objpush = self._has_objpush(device) - klass = get_minor_class(device['Class']) + has_objpush = self._has_objpush(properties["UUIDs"]) + klass_id = cast(int, properties["Class"]) + klass = get_minor_class(klass_id) # Bluetooth >= 4 devices use Appearance property - appearance = device["Appearance"] + appearance = properties["Appearance"] if klass != _("Uncategorized") and klass != _("Unknown"): description = klass elif klass == _("Unknown") and appearance: description = gatt_appearance_to_name(appearance) else: - description = get_major_class(device['Class']) + description = get_major_class(klass_id) - surface = self._make_device_icon(device["Icon"], device["Paired"], device["Connected"], device["Trusted"], - device["Blocked"]) + surface = self._make_device_icon(properties["Icon"], properties["Paired"], properties["Connected"], + properties["Trusted"], properties["Blocked"]) surface_object = SurfaceObject(surface) - display_name = self.make_display_name(device.display_name, device["Class"], device['Address']) - caption = self.make_caption(display_name, description, device['Address']) + address = cast(BtAddress, properties["Address"]) + display_name = self.make_display_name(device.display_name, klass_id, address) + caption = self.make_caption(display_name, description, address) self.set(tree_iter, caption=caption, alias=display_name, objpush=has_objpush, device_surface=surface_object) + self.set(tree_iter, trusted=properties["Trusted"], paired=properties["Paired"], + connected=properties["Connected"], blocked=properties["Blocked"]) - try: - self.row_update_event(tree_iter, "Trusted", device['Trusted']) - except Exception as e: - logging.exception(e) - try: - self.row_update_event(tree_iter, "Paired", device['Paired']) - except Exception as e: - logging.exception(e) - try: - self.row_update_event(tree_iter, "Connected", device["Connected"]) - except Exception as e: - logging.exception(e) - try: - self.row_update_event(tree_iter, "Blocked", device["Blocked"]) - except Exception as e: - logging.exception(e) - - if device["Connected"]: - self._monitor_power_levels(tree_iter, device) + if properties["Connected"]: + self._monitor_power_levels(tree_iter, device, properties) - def _monitor_power_levels(self, tree_iter: Gtk.TreeIter, device: Device) -> None: - if device["Address"] in self._monitored_devices: + def _monitor_power_levels( + self, tree_iter: Gtk.TreeIter, device: Device, properties: Mapping[str, Any] | None = None + ) -> None: + address = cast(BtAddress, device["Address"] if properties is None else properties["Address"]) + if address in self._monitored_devices: return assert self.Adapter is not None hci_dev = adapter_path_to_name(self.Adapter.get_object_path()) assert hci_dev is not None - cinfo = conn_info(device["Address"], hci_dev) + cinfo = conn_info(address, hci_dev) try: cinfo.init() except ConnInfoReadError: @@ -426,68 +420,80 @@ def _monitor_power_levels(self, tree_iter: Gtk.TreeIter, device: Device) -> None assert isinstance(model, Gtk.TreeModel) r = Gtk.TreeRowReference.new(model, model.get_path(tree_iter)) self._update_power_levels(tree_iter, device, cinfo) - GLib.timeout_add(1000, self._check_power_levels, r, cinfo, device["Address"]) - self._monitored_devices.add(device["Address"]) - - def _check_power_levels(self, row_ref: Gtk.TreeRowReference, cinfo: conn_info, address: BtAddress) -> bool: - if not row_ref.valid(): - logging.warning("stopping monitor (row does not exist)") - cinfo.deinit() - self._monitored_devices.remove(address) - return False + self._monitored_devices[address] = (r, cinfo) + if self._power_levels_timer is None: + self._power_levels_timer = GLib.timeout_add(1000, self._check_power_levels) + + def _check_power_levels(self) -> bool: + for address, (row_ref, cinfo) in list(self._monitored_devices.items()): + if not row_ref.valid(): + logging.warning("stopping monitor (row does not exist)") + cinfo.deinit() + del self._monitored_devices[address] + continue + + tree_iter = self.get_iter(row_ref.get_path()) + assert tree_iter is not None - tree_iter = self.get_iter(row_ref.get_path()) - assert tree_iter is not None + device = self.get(tree_iter, "device")["device"] - device = self.get(tree_iter, "device")["device"] + if device["Connected"]: + self._update_power_levels(tree_iter, device, cinfo) + else: + cinfo.deinit() + self._disable_power_levels(tree_iter) + del self._monitored_devices[address] - if device["Connected"]: - self._update_power_levels(tree_iter, device, cinfo) + if self._monitored_devices: return True - else: - cinfo.deinit() - self._disable_power_levels(tree_iter) - self._monitored_devices.remove(address) - return False + + self._power_levels_timer = None + return False def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> None: logging.info(f"{key} {value}") device = self.get(tree_iter, "device")["device"] + properties = None if key in ("Blocked", "Connected", "Paired", "Trusted"): - surface = self._make_device_icon(device["Icon"], device["Paired"], device["Connected"], device["Trusted"], - device["Blocked"]) + properties = device.get_properties() + properties[key] = value + surface = self._make_device_icon(properties["Icon"], properties["Paired"], properties["Connected"], + properties["Trusted"], properties["Blocked"]) self.set(tree_iter, device_surface=SurfaceObject(surface)) if key == "Trusted": - if value: - self.set(tree_iter, trusted=True) - else: - self.set(tree_iter, trusted=False) + self.set(tree_iter, trusted=value) elif key == "Paired": - if value: - self.set(tree_iter, paired=True) - else: - self.set(tree_iter, paired=False) + self.set(tree_iter, paired=value) elif key == "Alias": - c = self.make_caption(value, self.get_device_class(device), device['Address']) - name = self.make_display_name(device.display_name, device["Class"], device["Address"]) + properties = device.get_properties() + properties[key] = value + address = cast(BtAddress, properties["Address"]) + c = self.make_caption(value, self.get_device_class(device, properties), address) + name = self.make_display_name(device.display_name, properties["Class"], address) self.set(tree_iter, caption=c, alias=name) elif key == "UUIDs": - has_objpush = self._has_objpush(device) + has_objpush = self._has_objpush(value) self.set(tree_iter, objpush=has_objpush) elif key == "Connected": self.set(tree_iter, connected=value) if value: - self._monitor_power_levels(tree_iter, device) + assert properties is not None + self._monitor_power_levels(tree_iter, device, properties) else: self._disable_power_levels(tree_iter) + assert properties is not None + address = cast(BtAddress, properties["Address"]) + monitored = self._monitored_devices.pop(address, None) + if monitored is not None: + monitored[1].deinit() elif key == "Name": self.set(tree_iter, no_name=False) self.filter.refilter() @@ -651,11 +657,9 @@ def tooltip_query(self, _tw: Gtk.Widget, x: int, y: int, _kb: bool, tooltip: Gtk return True return False - def _has_objpush(self, device: Device) -> bool: - if device is None: - return False - - for uuid in device["UUIDs"]: + @staticmethod + def _has_objpush(uuids: Iterable[str]) -> bool: + for uuid in uuids: if ServiceUUID(uuid).short_uuid == OBEX_OBJPUSH_SVCLASS_ID: return True return False diff --git a/test/gui/manager/test_manager_device_list.py b/test/gui/manager/test_manager_device_list.py new file mode 100644 index 000000000..3b97a0f34 --- /dev/null +++ b/test/gui/manager/test_manager_device_list.py @@ -0,0 +1,172 @@ +from pathlib import Path +import sys +import types +from unittest import TestCase +from unittest.mock import Mock, patch + +import gi + +gi.require_version("Gtk", "3.0") +from gi.repository import Gtk + +constants = types.ModuleType("blueman.Constants") +constants.BIN_DIR = Path("/tmp") +constants.BLUETOOTHD_PATH = Path("/tmp/bluetoothd") +constants.ICON_PATH = Path("/tmp") +constants.PIXMAP_PATH = Path("/tmp") +constants.UI_PATH = Path("/tmp") +sys.modules.setdefault("blueman.Constants", constants) + +from blueman.gui.manager.ManagerDeviceList import ManagerDeviceList + + +class FakeDevice: + display_name = "Keyboard" + + def __init__(self) -> None: + self.item_reads: list[str] = [] + self.properties = { + "Address": "AA:BB:CC:DD:EE:FF", + "Alias": "Keyboard", + "Appearance": 0, + "Blocked": False, + "Class": 0, + "Connected": False, + "Icon": "input-keyboard", + "Paired": False, + "Trusted": False, + "UUIDs": [], + } + + def get_properties(self): + return dict(self.properties) + + def __getitem__(self, key): + self.item_reads.append(key) + return self.properties[key] + + +class FakeManagerDeviceList: + Adapter = None + + def __init__(self) -> None: + self.values = {"initial_anim": True, "device": FakeDevice()} + self.set_calls: list[dict[str, object]] = [] + self.monitor_calls = [] + self.disabled = [] + self._monitored_devices = {} + + def get(self, _tree_iter, *keys): + return {key: self.values[key] for key in keys} + + def set(self, _tree_iter, **kwargs): + self.values.update(kwargs) + self.set_calls.append(kwargs) + + def _make_device_icon(self, *args): + self.icon_args = args + return Mock() + + def _monitor_power_levels(self, tree_iter, device, properties=None): + self.monitor_calls.append((tree_iter, device, properties)) + + def _check_power_levels(self): + return False + + def _disable_power_levels(self, tree_iter): + self.disabled.append(tree_iter) + + def make_display_name(self, display_name, _klass, _address): + return display_name + + def make_caption(self, display_name, description, address): + return f"{display_name} {description} {address}" + + _has_objpush = staticmethod(ManagerDeviceList._has_objpush) + + +class TestManagerDeviceListProperties(TestCase): + def test_has_objpush_uses_uuid_iterable(self): + self.assertTrue(ManagerDeviceList._has_objpush(["00001105-0000-1000-8000-00805f9b34fb"])) + self.assertFalse(ManagerDeviceList._has_objpush(["0000110a-0000-1000-8000-00805f9b34fb"])) + + def test_row_setup_uses_get_all_properties(self): + fake = FakeManagerDeviceList() + device = FakeDevice() + device.properties["Connected"] = True + tree_iter = object() + + ManagerDeviceList.row_setup_event(fake, tree_iter, device) + + self.assertEqual(device.item_reads, []) + self.assertEqual(fake.values["trusted"], False) + self.assertEqual(fake.values["paired"], False) + self.assertEqual(fake.values["connected"], True) + self.assertEqual(fake.values["blocked"], False) + self.assertEqual(len(fake.monitor_calls), 1) + self.assertEqual(fake.monitor_calls[0][2]["Address"], device.properties["Address"]) + + def test_row_update_batches_icon_properties(self): + fake = FakeManagerDeviceList() + device = fake.values["device"] + tree_iter = object() + + ManagerDeviceList.row_update_event(fake, tree_iter, "Trusted", True) + + self.assertEqual(device.item_reads, []) + self.assertEqual(fake.icon_args, ("input-keyboard", False, False, True, False)) + self.assertTrue(fake.values["trusted"]) + + def test_row_update_connected_false_uses_batched_address(self): + fake = FakeManagerDeviceList() + device = fake.values["device"] + cinfo = Mock() + fake._monitored_devices = {device.properties["Address"]: (Mock(), cinfo)} + tree_iter = object() + + ManagerDeviceList.row_update_event(fake, tree_iter, "Connected", False) + + self.assertEqual(device.item_reads, []) + self.assertEqual(fake.disabled, [tree_iter]) + cinfo.deinit.assert_called_once_with() + self.assertEqual(fake._monitored_devices, {}) + + +class TestManagerDeviceListPowerTimer(TestCase): + def test_monitor_power_levels_starts_one_timer(self): + fake = FakeManagerDeviceList() + fake.Adapter = Mock() + fake.Adapter.get_object_path.return_value = "/org/bluez/hci0" + fake.liststore = Gtk.ListStore(str) + tree_iter = fake.liststore.append(["Keyboard"]) + fake._monitored_devices = {} + fake._power_levels_timer = None + fake._update_power_levels = Mock() + device = FakeDevice() + + with patch("blueman.gui.manager.ManagerDeviceList.adapter_path_to_name", return_value="hci0"), \ + patch("blueman.gui.manager.ManagerDeviceList.conn_info") as conn_info_cls, \ + patch("blueman.gui.manager.ManagerDeviceList.Gtk.TreeRowReference.new", return_value=Mock()), \ + patch("blueman.gui.manager.ManagerDeviceList.GLib.timeout_add", return_value=12) as timeout_add: + ManagerDeviceList._monitor_power_levels(fake, tree_iter, device, device.get_properties()) + ManagerDeviceList._monitor_power_levels(fake, tree_iter, device, device.get_properties()) + + conn_info_cls.assert_called_once_with("AA:BB:CC:DD:EE:FF", "hci0") + timeout_add.assert_called_once_with(1000, fake._check_power_levels) + self.assertEqual(fake._power_levels_timer, 12) + self.assertEqual(len(fake._monitored_devices), 1) + + def test_check_power_levels_stops_when_no_devices_remain(self): + row_ref = Mock() + row_ref.valid.return_value = False + cinfo = Mock() + fake = FakeManagerDeviceList() + fake._monitored_devices = {"AA:BB:CC:DD:EE:FF": (row_ref, cinfo)} + fake._power_levels_timer = 12 + + keep = ManagerDeviceList._check_power_levels(fake) + + self.assertFalse(keep) + cinfo.deinit.assert_called_once_with() + self.assertEqual(fake._monitored_devices, {}) + self.assertIsNone(fake._power_levels_timer) From e98e9a155a30bfa558682dd455f5b932ac768226 Mon Sep 17 00:00:00 2001 From: Geraldo Netto Date: Mon, 15 Jun 2026 06:22:41 +0200 Subject: [PATCH 2/9] perf: cache manager device uuids --- blueman/gui/manager/ManagerDeviceList.py | 18 ++++++----- blueman/gui/manager/ManagerDeviceMenu.py | 10 +++--- test/gui/manager/test_manager_device_list.py | 33 +++++++++++++++++++- 3 files changed, 48 insertions(+), 13 deletions(-) diff --git a/blueman/gui/manager/ManagerDeviceList.py b/blueman/gui/manager/ManagerDeviceList.py index e5f92349e..f05ab3920 100644 --- a/blueman/gui/manager/ManagerDeviceList.py +++ b/blueman/gui/manager/ManagerDeviceList.py @@ -67,6 +67,7 @@ def __init__(self, inst: "Blueman", adapter: str | None = None) -> None: {"id": "paired", "type": bool}, # used for quick access instead of device.GetProperties {"id": "trusted", "type": bool}, # used for quick access instead of device.GetProperties {"id": "objpush", "type": bool}, # used to set Send File button + {"id": "uuids", "type": object}, {"id": "battery", "type": float}, {"id": "rssi", "type": float}, {"id": "tpl", "type": float}, @@ -193,7 +194,7 @@ def drag_motion(self, _widget: Gtk.Widget, drag_context: Gdk.DragContext, x: int if not self.selection.path_is_selected(path): tree_iter = self.get_iter(path) assert tree_iter is not None - has_obj_push = self._has_objpush(self.get(tree_iter, "device")["device"]) + has_obj_push = self.get(tree_iter, "objpush")["objpush"] if has_obj_push: Gdk.drag_status(drag_context, Gdk.DragAction.COPY, timestamp) self.set_cursor(path) @@ -234,7 +235,7 @@ def _on_event_clicked(self, _widget: Gtk.Widget, event: Gdk.Event) -> bool: assert tree_iter is not None child_iter = self.filter.convert_iter_to_child_iter(tree_iter) assert child_iter is not None - row = self.get(child_iter, "device", "connected") + row = self.get(child_iter, "device", "connected", "uuids") if not row: return False @@ -242,7 +243,7 @@ def _on_event_clicked(self, _widget: Gtk.Widget, event: Gdk.Event) -> bool: self.menu = ManagerDeviceMenu(self.Blueman) if event.type == Gdk.EventType._2BUTTON_PRESS and cast(Gdk.EventButton, event).button == 1: - if self.menu.show_generic_connect_calc(row["device"]['UUIDs']): + if self.menu.show_generic_connect_calc(row["uuids"]): if row["connected"]: self.menu.disconnect_service(row["device"]) elif Adapter(obj_path=row["device"]["Adapter"])["Powered"]: @@ -374,7 +375,8 @@ def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: else: self.set(tree_iter, initial_anim=False) - has_objpush = self._has_objpush(properties["UUIDs"]) + uuids = tuple(cast(Iterable[str], properties["UUIDs"])) + has_objpush = self._has_objpush(uuids) klass_id = cast(int, properties["Class"]) klass = get_minor_class(klass_id) # Bluetooth >= 4 devices use Appearance property @@ -393,7 +395,8 @@ def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: display_name = self.make_display_name(device.display_name, klass_id, address) caption = self.make_caption(display_name, description, address) - self.set(tree_iter, caption=caption, alias=display_name, objpush=has_objpush, device_surface=surface_object) + self.set(tree_iter, caption=caption, alias=display_name, objpush=has_objpush, uuids=uuids, + device_surface=surface_object) self.set(tree_iter, trusted=properties["Trusted"], paired=properties["Paired"], connected=properties["Connected"], blocked=properties["Blocked"]) @@ -478,8 +481,9 @@ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> Non self.set(tree_iter, caption=c, alias=name) elif key == "UUIDs": - has_objpush = self._has_objpush(value) - self.set(tree_iter, objpush=has_objpush) + uuids = tuple(cast(Iterable[str], value)) + has_objpush = self._has_objpush(uuids) + self.set(tree_iter, objpush=has_objpush, uuids=uuids) elif key == "Connected": self.set(tree_iter, connected=value) diff --git a/blueman/gui/manager/ManagerDeviceMenu.py b/blueman/gui/manager/ManagerDeviceMenu.py index eb0645548..b20c10848 100644 --- a/blueman/gui/manager/ManagerDeviceMenu.py +++ b/blueman/gui/manager/ManagerDeviceMenu.py @@ -265,8 +265,8 @@ def generate(self) -> None: selected = self.Blueman.List.selected() if not selected: return - row = self.Blueman.List.get(selected, "alias", "paired", "connected", "trusted", "objpush", "device", - "blocked") + row = self.Blueman.List.get(selected, "alias", "paired", "connected", "trusted", "objpush", "uuids", + "device", "blocked") else: (x, y) = self.Blueman.List.get_pointer() posdata = self.Blueman.List.get_path_at_pos(x, y) @@ -282,8 +282,8 @@ def generate(self) -> None: child_iter = self.Blueman.List.filter.convert_iter_to_child_iter(tree_iter) assert child_iter is not None - row = self.Blueman.List.get(child_iter, "alias", "paired", "connected", "trusted", "objpush", "device", - "blocked") + row = self.Blueman.List.get(child_iter, "alias", "paired", "connected", "trusted", "objpush", "uuids", + "device", "blocked") self.SelectedDevice = row["device"] @@ -296,7 +296,7 @@ def generate(self) -> None: self.append(item) return - show_generic_connect = self.show_generic_connect_calc(self.SelectedDevice['UUIDs']) + show_generic_connect = self.show_generic_connect_calc(row["uuids"]) powered = Adapter(obj_path=self.SelectedDevice["Adapter"])["Powered"] diff --git a/test/gui/manager/test_manager_device_list.py b/test/gui/manager/test_manager_device_list.py index 3b97a0f34..1cc7f8311 100644 --- a/test/gui/manager/test_manager_device_list.py +++ b/test/gui/manager/test_manager_device_list.py @@ -50,7 +50,7 @@ class FakeManagerDeviceList: Adapter = None def __init__(self) -> None: - self.values = {"initial_anim": True, "device": FakeDevice()} + self.values = {"initial_anim": True, "device": FakeDevice(), "objpush": False, "uuids": ()} self.set_calls: list[dict[str, object]] = [] self.monitor_calls = [] self.disabled = [] @@ -103,6 +103,7 @@ def test_row_setup_uses_get_all_properties(self): self.assertEqual(fake.values["paired"], False) self.assertEqual(fake.values["connected"], True) self.assertEqual(fake.values["blocked"], False) + self.assertEqual(fake.values["uuids"], ()) self.assertEqual(len(fake.monitor_calls), 1) self.assertEqual(fake.monitor_calls[0][2]["Address"], device.properties["Address"]) @@ -131,6 +132,36 @@ def test_row_update_connected_false_uses_batched_address(self): cinfo.deinit.assert_called_once_with() self.assertEqual(fake._monitored_devices, {}) + def test_row_update_uuids_updates_cached_state(self): + fake = FakeManagerDeviceList() + tree_iter = object() + uuids = ["00001105-0000-1000-8000-00805f9b34fb"] + + ManagerDeviceList.row_update_event(fake, tree_iter, "UUIDs", uuids) + + self.assertEqual(fake.values["uuids"], tuple(uuids)) + self.assertTrue(fake.values["objpush"]) + + def test_drag_motion_uses_cached_objpush_state(self): + fake = FakeManagerDeviceList() + device = fake.values["device"] + fake.values["objpush"] = True + fake.filter = Mock() + fake.filter.convert_path_to_child_path.return_value = Mock() + fake.selection = Mock() + fake.selection.path_is_selected.return_value = False + fake.get_path_at_pos = Mock(return_value=(Mock(), Mock(), Mock(), Mock())) + fake.get_iter = Mock(return_value=Mock()) + fake.set_cursor = Mock() + + with patch("blueman.gui.manager.ManagerDeviceList.Gdk.drag_status") as drag_status: + result = ManagerDeviceList.drag_motion(fake, Mock(), Mock(), 1, 2, 3) + + self.assertTrue(result) + self.assertEqual(device.item_reads, []) + drag_status.assert_called_once() + fake.set_cursor.assert_called_once() + class TestManagerDeviceListPowerTimer(TestCase): def test_monitor_power_levels_starts_one_timer(self): From af0d8fe39f2cda8d4abfd39db5e7ece17c473bfa Mon Sep 17 00:00:00 2001 From: Geraldo Netto Date: Mon, 15 Jun 2026 06:44:31 +0200 Subject: [PATCH 3/9] test: address manager perf review feedback --- blueman/gui/manager/ManagerDeviceList.py | 10 +++++++--- test/gui/manager/test_manager_device_list.py | 8 ++++---- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/blueman/gui/manager/ManagerDeviceList.py b/blueman/gui/manager/ManagerDeviceList.py index f05ab3920..1a5f329e0 100644 --- a/blueman/gui/manager/ManagerDeviceList.py +++ b/blueman/gui/manager/ManagerDeviceList.py @@ -428,11 +428,12 @@ def _monitor_power_levels( self._power_levels_timer = GLib.timeout_add(1000, self._check_power_levels) def _check_power_levels(self) -> bool: - for address, (row_ref, cinfo) in list(self._monitored_devices.items()): + remove_addresses = [] + for address, (row_ref, cinfo) in self._monitored_devices.items(): if not row_ref.valid(): logging.warning("stopping monitor (row does not exist)") cinfo.deinit() - del self._monitored_devices[address] + remove_addresses.append(address) continue tree_iter = self.get_iter(row_ref.get_path()) @@ -445,7 +446,10 @@ def _check_power_levels(self) -> bool: else: cinfo.deinit() self._disable_power_levels(tree_iter) - del self._monitored_devices[address] + remove_addresses.append(address) + + for address in remove_addresses: + del self._monitored_devices[address] if self._monitored_devices: return True diff --git a/test/gui/manager/test_manager_device_list.py b/test/gui/manager/test_manager_device_list.py index 1cc7f8311..bdf470cf9 100644 --- a/test/gui/manager/test_manager_device_list.py +++ b/test/gui/manager/test_manager_device_list.py @@ -99,10 +99,10 @@ def test_row_setup_uses_get_all_properties(self): ManagerDeviceList.row_setup_event(fake, tree_iter, device) self.assertEqual(device.item_reads, []) - self.assertEqual(fake.values["trusted"], False) - self.assertEqual(fake.values["paired"], False) - self.assertEqual(fake.values["connected"], True) - self.assertEqual(fake.values["blocked"], False) + self.assertFalse(fake.values["trusted"]) + self.assertFalse(fake.values["paired"]) + self.assertTrue(fake.values["connected"]) + self.assertFalse(fake.values["blocked"]) self.assertEqual(fake.values["uuids"], ()) self.assertEqual(len(fake.monitor_calls), 1) self.assertEqual(fake.monitor_calls[0][2]["Address"], device.properties["Address"]) From c100da1a55e19efa5e85fc9eaeec13138289505b Mon Sep 17 00:00:00 2001 From: Geraldo Netto Date: Fri, 19 Jun 2026 10:52:18 +0200 Subject: [PATCH 4/9] test: add ManagerDeviceList row-path D-Bus round-trip benchmark Benchmarks the row_setup_event/row_update_event hot path by modeling each device property access as one synchronous D-Bus round-trip. run_compare.sh diffs the current branch against a base ref (default main) and asserts a minimum gain. Why the gain: every device["Key"] resolves to one synchronous org.freedesktop.DBus.Properties.Get via Gio call_sync (blueman/bluez/Base.py get()). main reads each property individually and re-runs row_update_event four times inside row_setup_event, so a single row costs ~57 round-trips. The branch calls get_properties() once (one Properties.GetAll), caches uuids in the liststore, and batches the set() calls, cutting the path to ~5 round-trips. D-Bus round-trips dominate wall time, so collapsing them is the gain; pure-Python CPU is near-flat between versions. Measured (10000 iterations, DBUS_RTT=200us, /usr/bin/python3): metric main (fb418907) branch (af0d8fe3) gain ---------------------- --------------- ----------------- --------- D-Bus round-trips/iter 57 5 91.2% fewer round-trips total 570000 50000 91.2% fewer cpu_seconds 4.54 4.15 8.5% faster modeled wall time 118.54s 14.15s 88.1% faster speedup factor 1.0x 8.4x 8.4x Co-Authored-By: Claude Opus 4.8 (1M context) --- test/benchmarks/__init__.py | 0 test/benchmarks/bench_manager_device_list.py | 175 +++++++++++++++++++ test/benchmarks/run_compare.sh | 64 +++++++ 3 files changed, 239 insertions(+) create mode 100644 test/benchmarks/__init__.py create mode 100644 test/benchmarks/bench_manager_device_list.py create mode 100755 test/benchmarks/run_compare.sh diff --git a/test/benchmarks/__init__.py b/test/benchmarks/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/test/benchmarks/bench_manager_device_list.py b/test/benchmarks/bench_manager_device_list.py new file mode 100644 index 000000000..858375354 --- /dev/null +++ b/test/benchmarks/bench_manager_device_list.py @@ -0,0 +1,175 @@ +"""Micro-benchmark for ManagerDeviceList row setup/update hot paths. + +The optimization under test collapses many per-property D-Bus reads +(``device["Key"]`` -> ``org.freedesktop.DBus.Properties.Get``) into a single +``device.get_properties()`` call (``Properties.GetAll``) plus cached liststore +state. Each ``__getitem__`` and each ``get_properties()`` therefore models one +synchronous D-Bus round-trip -- the dominant real-world cost on this path. + +The benchmark is version-agnostic: it drives whichever ``ManagerDeviceList`` is +importable from the current tree, so the same file runs against both ``main`` +and the optimized branch (see ``run_compare.sh``). + +Output: a single JSON line with round-trip counts and a modeled wall time +(pure-Python CPU time + round-trips * DBUS_RTT). +""" +from __future__ import annotations + +import inspect +import json +import logging +import sys +import time +import types +from pathlib import Path +from unittest.mock import Mock + +import gi + +gi.require_version("Gtk", "3.0") # must precede ManagerDeviceList import + +# Stub blueman.Constants the same way the unit tests do, so importing the +# module under test never touches the real installation paths. +_constants = types.ModuleType("blueman.Constants") +_constants.BIN_DIR = Path("/tmp") +_constants.BLUETOOTHD_PATH = Path("/tmp/bluetoothd") +_constants.ICON_PATH = Path("/tmp") +_constants.PIXMAP_PATH = Path("/tmp") +_constants.UI_PATH = Path("/tmp") +sys.modules.setdefault("blueman.Constants", _constants) + +from blueman.gui.manager.ManagerDeviceList import ManagerDeviceList # noqa: E402 + +# Modeled cost of one synchronous D-Bus round-trip (Properties.Get / GetAll via +# Gio call_sync) on a local system bus. Conservative; the relative gain is +# insensitive to the exact value because it is dominated by round-trip count. +DBUS_RTT = 0.0002 # 200 microseconds + +# Keys updated after initial setup, mirroring real signal traffic. "Connected" +# is intentionally excluded so neither version enters the power-level timer +# machinery -- keeping the comparison focused on the property-read path. +UPDATE_KEYS: list[tuple[str, object]] = [ + ("Trusted", True), + ("Paired", True), + ("Blocked", False), + ("Alias", "Keyboard"), + ("UUIDs", ["00001105-0000-1000-8000-00805f9b34fb"]), +] + + +class CountingDevice: + """Fake bluez Device that counts simulated D-Bus round-trips.""" + + display_name = "Keyboard" + + def __init__(self, counters: dict[str, int]) -> None: + self._counters = counters + self.properties = { + "Address": "AA:BB:CC:DD:EE:FF", + "Alias": "Keyboard", + "Appearance": 0, + "Blocked": False, + "Class": 0, + "Connected": False, + "Icon": "input-keyboard", + "Paired": False, + "Trusted": False, + "UUIDs": ["00001105-0000-1000-8000-00805f9b34fb"], + } + + def __getitem__(self, key: str) -> object: + self._counters["roundtrips"] += 1 # Properties.Get + return self.properties[key] + + def get_properties(self) -> dict[str, object]: + self._counters["roundtrips"] += 1 # Properties.GetAll + return dict(self.properties) + + def get_object_path(self) -> str: + return "/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF" + + +class Harness: + """Minimal stand-in for ManagerDeviceList state. + + Only GTK/render primitives are faked; the methods under test + (``row_setup_event``, ``row_update_event``, ``_has_objpush``, + ``get_device_class``) are the real ones, bound below. + """ + + Adapter = None + filter = Mock() + + def __init__(self, counters: dict[str, int]) -> None: + self.values: dict[str, object] = { + "initial_anim": True, + "device": CountingDevice(counters), + "objpush": False, + "uuids": (), + } + self._monitored_devices: set[str] = set() + self._power_levels_timer = None + + def get(self, _tree_iter: object, *keys: str) -> dict[str, object]: + return {key: self.values[key] for key in keys} + + def set(self, _tree_iter: object, **kwargs: object) -> None: + self.values.update(kwargs) + + def _make_device_icon(self, *_args: object) -> object: + # Cheap stand-in for a cairo surface; SurfaceObject only stores it. + return None + + def _disable_power_levels(self, _tree_iter: object) -> None: + pass + + def make_display_name(self, display_name: str, _klass: object, _address: object) -> str: + return display_name + + def make_caption(self, display_name: str, description: str, address: object) -> str: + return f"{display_name} {description} {address}" + + # Real implementations under test, copied as the *original descriptors* + # (staticmethod vs plain function) so each version binds correctly: + # main's _has_objpush is an instance method, the branch's is static. + _has_objpush = inspect.getattr_static(ManagerDeviceList, "_has_objpush") + get_device_class = inspect.getattr_static(ManagerDeviceList, "get_device_class") + row_setup_event = inspect.getattr_static(ManagerDeviceList, "row_setup_event") + row_update_event = inspect.getattr_static(ManagerDeviceList, "row_update_event") + + +def _one_pass(counters: dict[str, int]) -> None: + harness = Harness(counters) + tree_iter = object() + harness.row_setup_event(tree_iter, harness.values["device"]) + for key, value in UPDATE_KEYS: + harness.row_update_event(tree_iter, key, value) + + +def run(iterations: int) -> dict[str, object]: + logging.disable(logging.CRITICAL) # silence per-update info logging + + # Warm up (import/JIT of attribute caches) without scoring it. + _one_pass({"roundtrips": 0}) + + counters = {"roundtrips": 0} + start = time.perf_counter() + for _ in range(iterations): + _one_pass(counters) + cpu = time.perf_counter() - start + + roundtrips = counters["roundtrips"] + modeled = cpu + roundtrips * DBUS_RTT + return { + "iterations": iterations, + "roundtrips_total": roundtrips, + "roundtrips_per_iter": roundtrips / iterations, + "cpu_seconds": cpu, + "dbus_rtt": DBUS_RTT, + "modeled_seconds": modeled, + } + + +if __name__ == "__main__": + iters = int(sys.argv[1]) if len(sys.argv) > 1 else 20000 + print(json.dumps(run(iters))) diff --git a/test/benchmarks/run_compare.sh b/test/benchmarks/run_compare.sh new file mode 100755 index 000000000..ec0b77134 --- /dev/null +++ b/test/benchmarks/run_compare.sh @@ -0,0 +1,64 @@ +#!/usr/bin/env bash +# Compare the ManagerDeviceList row-path benchmark between the current branch +# and main, then assert the branch is at least MIN_GAIN% faster. +# +# Usage: test/benchmarks/run_compare.sh [iterations] [base_ref] +# +# Models each device["Key"] / get_properties() as one synchronous D-Bus +# round-trip (see bench_manager_device_list.py). The headline metric is the +# modeled wall time (CPU + round-trips * DBUS_RTT); round-trip count is the +# deterministic underlying driver. +set -euo pipefail + +ITERS="${1:-10000}" +BASE_REF="${2:-main}" +MIN_GAIN=5 + +# blueman.Constants is generated at build time; the benchmark stubs it, but the +# package must be importable from the tree under test, hence PYTHONPATH. +PY="${PYTHON:-/usr/bin/python3}" + +REPO_ROOT="$(git -C "$(dirname "$0")" rev-parse --show-toplevel)" +BENCH="test/benchmarks/bench_manager_device_list.py" +WORKTREE="$(mktemp -d)/base" + +cleanup() { git -C "$REPO_ROOT" worktree remove --force "$WORKTREE" 2>/dev/null || true; } +trap cleanup EXIT + +echo "Benchmark: $BENCH (iterations=$ITERS, dbus model in-script)" + +BRANCH_REV="$(git -C "$REPO_ROOT" rev-parse --short HEAD)" +echo "=== current branch ($BRANCH_REV) ===" +BRANCH_JSON="$(cd "$REPO_ROOT" && PYTHONPATH="$REPO_ROOT" "$PY" "$BENCH" "$ITERS")" +echo "$BRANCH_JSON" + +git -C "$REPO_ROOT" worktree add -q "$WORKTREE" "$BASE_REF" +cp "$REPO_ROOT/$BENCH" "$WORKTREE/bench_base.py" +BASE_REV="$(git -C "$WORKTREE" rev-parse --short HEAD)" +echo "=== base $BASE_REF ($BASE_REV) ===" +BASE_JSON="$(cd "$WORKTREE" && PYTHONPATH="$WORKTREE" "$PY" bench_base.py "$ITERS")" +echo "$BASE_JSON" + +echo "=== result ===" +BRANCH_JSON="$BRANCH_JSON" BASE_JSON="$BASE_JSON" MIN_GAIN="$MIN_GAIN" "$PY" - <<'PYEOF' +import json, os, sys +br = json.loads(os.environ["BRANCH_JSON"]) +base = json.loads(os.environ["BASE_JSON"]) +min_gain = float(os.environ["MIN_GAIN"]) + +def pct(old, new): + return (old - new) / old * 100 + +rt_gain = pct(base["roundtrips_per_iter"], br["roundtrips_per_iter"]) +t_gain = pct(base["modeled_seconds"], br["modeled_seconds"]) +print(f"D-Bus round-trips/iter: base={base['roundtrips_per_iter']:.0f} " + f"branch={br['roundtrips_per_iter']:.0f} -> {rt_gain:.1f}% fewer") +print(f"Modeled wall time: base={base['modeled_seconds']:.2f}s " + f"branch={br['modeled_seconds']:.2f}s -> {t_gain:.1f}% faster " + f"({base['modeled_seconds']/br['modeled_seconds']:.1f}x)") +if t_gain >= min_gain: + print(f"PASS: {t_gain:.1f}% >= {min_gain}% target") + sys.exit(0) +print(f"FAIL: {t_gain:.1f}% < {min_gain}% target") +sys.exit(1) +PYEOF From 9ddd1b65ec7c268219346e68228c8921230e426c Mon Sep 17 00:00:00 2001 From: Geraldo Netto Date: Fri, 19 Jun 2026 11:31:22 +0200 Subject: [PATCH 5/9] refactor: call BtAddress NewType instead of casting Per review: NewType values can be constructed directly (BtAddress(value)); type checkers infer the correct type without typing.cast. Co-Authored-By: Claude Opus 4.8 (1M context) --- blueman/gui/manager/ManagerDeviceList.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/blueman/gui/manager/ManagerDeviceList.py b/blueman/gui/manager/ManagerDeviceList.py index 1a5f329e0..80d53fff1 100644 --- a/blueman/gui/manager/ManagerDeviceList.py +++ b/blueman/gui/manager/ManagerDeviceList.py @@ -391,7 +391,7 @@ def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: surface = self._make_device_icon(properties["Icon"], properties["Paired"], properties["Connected"], properties["Trusted"], properties["Blocked"]) surface_object = SurfaceObject(surface) - address = cast(BtAddress, properties["Address"]) + address = BtAddress(properties["Address"]) display_name = self.make_display_name(device.display_name, klass_id, address) caption = self.make_caption(display_name, description, address) @@ -406,7 +406,7 @@ def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: def _monitor_power_levels( self, tree_iter: Gtk.TreeIter, device: Device, properties: Mapping[str, Any] | None = None ) -> None: - address = cast(BtAddress, device["Address"] if properties is None else properties["Address"]) + address = BtAddress(device["Address"] if properties is None else properties["Address"]) if address in self._monitored_devices: return @@ -479,7 +479,7 @@ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> Non elif key == "Alias": properties = device.get_properties() properties[key] = value - address = cast(BtAddress, properties["Address"]) + address = BtAddress(properties["Address"]) c = self.make_caption(value, self.get_device_class(device, properties), address) name = self.make_display_name(device.display_name, properties["Class"], address) self.set(tree_iter, caption=c, alias=name) @@ -498,7 +498,7 @@ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> Non else: self._disable_power_levels(tree_iter) assert properties is not None - address = cast(BtAddress, properties["Address"]) + address = BtAddress(properties["Address"]) monitored = self._monitored_devices.pop(address, None) if monitored is not None: monitored[1].deinit() From ca490988cdb63a4a2cfc45d7c596cbbf20733ee7 Mon Sep 17 00:00:00 2001 From: Geraldo Netto Date: Wed, 8 Jul 2026 21:15:51 +0200 Subject: [PATCH 6/9] perf: batch liststore writes into one row-changed emission GenericList.set wrote columns one liststore.set call at a time, emitting one row-changed per column. Every emission re-runs the sort machinery and the visibility filter, so a multi-column row update paid that cost repeatedly. Passing all columns in a single call emits row-changed once. Co-Authored-By: Claude Fable 5 --- blueman/gui/GenericList.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/blueman/gui/GenericList.py b/blueman/gui/GenericList.py index 4125189db..c844baf79 100644 --- a/blueman/gui/GenericList.py +++ b/blueman/gui/GenericList.py @@ -93,8 +93,10 @@ def prepend(self, **list_columns: object) -> Gtk.TreeIter: return self.liststore.prepend(vals) def set(self, tree_iter: Gtk.TreeIter, **list_columns: object) -> None: - for col_id, col_value in list_columns.items(): - self.liststore.set(tree_iter, self.list_col_order[col_id], col_value) + # a single set call emits one row-changed instead of one per column, + # so the filter and sort machinery only run once per update + self.liststore.set(tree_iter, {self.list_col_order[col_id]: col_value + for col_id, col_value in list_columns.items()}) def get(self, tree_iter: Gtk.TreeIter, *items: str) -> dict[str, Any]: data = {} From 24ceaea8dcb8bc2f77419aaace4091cfcfddcb22 Mon Sep 17 00:00:00 2001 From: Geraldo Netto Date: Wed, 8 Jul 2026 21:16:04 +0200 Subject: [PATCH 7/9] perf: cache device properties to cut remaining D-Bus reads Every device["Key"] is a synchronous Properties.Get round trip and get_properties() is one GetAll. Cache the remaining hot properties in row state and share one snapshot across the add path: - Cache Class, Address and Icon in liststore columns (kept fresh from property-changed signals, mirroring the GetAll fallbacks for invalidated properties). Icon rebuilds and Alias updates in row_update_event now run without any D-Bus calls, as do filter_func (evaluated on every row-changed), drag-and-drop send, Ctrl+C address copy, double-click connect and menu generation. - The per-second power-level timer read Connected live from D-Bus for every monitored device; use the cached column instead. - add_device did three round trips per discovered device (Adapter get, GetAll for the Name check, GetAll in row_setup_event); do one GetAll and pass the snapshot through row_setup_event. Same for DeviceSelectorList, which did four separate property reads per row. - Cache signal bar pixbufs per (bar, level, scale) instead of decoding the PNG on every level change, and stop reading the battery percentage over D-Bus just to build a debug log line when debug logging is disabled. Benchmark (test/benchmarks): 5 -> 1 modeled round trips per row lifecycle vs the previous commit, 57 -> 1 vs main. Co-Authored-By: Claude Fable 5 --- blueman/gui/DeviceList.py | 12 +- blueman/gui/DeviceSelectorList.py | 14 +- blueman/gui/manager/ManagerDeviceList.py | 120 +++--- blueman/gui/manager/ManagerDeviceMenu.py | 12 +- test/gui/manager/test_manager_device_list.py | 380 ++++++++++++++++++- 5 files changed, 461 insertions(+), 77 deletions(-) diff --git a/blueman/gui/DeviceList.py b/blueman/gui/DeviceList.py index 282d2e401..815ea3c48 100644 --- a/blueman/gui/DeviceList.py +++ b/blueman/gui/DeviceList.py @@ -1,7 +1,7 @@ from datetime import datetime import logging from typing import Any -from collections.abc import Callable +from collections.abc import Callable, Mapping from blueman.Functions import adapter_path_to_name from blueman.gui.GenericList import GenericList, ListDataDict @@ -151,7 +151,8 @@ def on_icon_theme_changed(self, _icon_them: Gtk.IconTheme) -> None: # ##### virtual funcs ##### # called when row needs to be initialized - def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: + def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device, + properties: Mapping[str, Any] | None = None) -> None: pass # called when a property for a device changes @@ -222,8 +223,9 @@ def update_progress(self, time: float, totaltime: float) -> bool: def add_device(self, object_path: ObjectPath) -> None: device = Device(obj_path=object_path) + properties = device.get_properties() # device belongs to another adapter - if not self.Adapter or not device['Adapter'] == self.Adapter.get_object_path(): + if not self.Adapter or not properties["Adapter"] == self.Adapter.get_object_path(): return logging.info("adding new device") @@ -232,11 +234,11 @@ def add_device(self, object_path: ObjectPath) -> None: "device": device, "dbus_path": object_path, "timestamp": float(datetime.strftime(datetime.now(), '%Y%m%d%H%M%S%f')), - "no_name": "Name" not in device + "no_name": "Name" not in properties } tree_iter = self.append(**colls) - self.row_setup_event(tree_iter, device) + self.row_setup_event(tree_iter, device, properties) if self.get_selected_device() is None: self.selection.select_path(Gtk.TreePath.new_first()) diff --git a/blueman/gui/DeviceSelectorList.py b/blueman/gui/DeviceSelectorList.py index 36dc1b7aa..84166bdaa 100644 --- a/blueman/gui/DeviceSelectorList.py +++ b/blueman/gui/DeviceSelectorList.py @@ -1,3 +1,4 @@ +from collections.abc import Mapping from html import escape from typing import Any @@ -36,11 +37,14 @@ def on_icon_theme_changed(self, _icon_them: Gtk.IconTheme) -> None: device = self.get(row.iter, "device")["device"] self.row_setup_event(row.iter, device) - def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: - self.row_update_event(tree_iter, "Trusted", device['Trusted']) - self.row_update_event(tree_iter, "Paired", device['Paired']) - self.row_update_event(tree_iter, "Alias", device.display_name) - self.row_update_event(tree_iter, "Icon", device['Icon']) + def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device, + properties: Mapping[str, Any] | None = None) -> None: + if properties is None: + properties = device.get_properties() + self.row_update_event(tree_iter, "Trusted", properties["Trusted"]) + self.row_update_event(tree_iter, "Paired", properties["Paired"]) + self.row_update_event(tree_iter, "Alias", properties["Alias"].strip()) + self.row_update_event(tree_iter, "Icon", properties["Icon"]) def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> None: if key == "Trusted": diff --git a/blueman/gui/manager/ManagerDeviceList.py b/blueman/gui/manager/ManagerDeviceList.py index 80d53fff1..ad9498cb0 100644 --- a/blueman/gui/manager/ManagerDeviceList.py +++ b/blueman/gui/manager/ManagerDeviceList.py @@ -6,7 +6,6 @@ import cairo from blueman.bluemantyping import ObjectPath, BtAddress -from blueman.bluez.Adapter import Adapter from blueman.bluez.Battery import Battery from blueman.bluez.Device import Device from blueman.bluez.Manager import Manager @@ -68,6 +67,9 @@ def __init__(self, inst: "Blueman", adapter: str | None = None) -> None: {"id": "trusted", "type": bool}, # used for quick access instead of device.GetProperties {"id": "objpush", "type": bool}, # used to set Send File button {"id": "uuids", "type": object}, + {"id": "klass", "type": int}, # used for quick access instead of device.GetProperties + {"id": "address", "type": str}, # used for quick access instead of device.GetProperties + {"id": "icon_name", "type": str}, # used for quick access instead of device.GetProperties {"id": "battery", "type": float}, {"id": "rssi", "type": float}, {"id": "tpl", "type": float}, @@ -84,6 +86,7 @@ def __init__(self, inst: "Blueman", adapter: str | None = None) -> None: self._monitored_devices: dict[BtAddress, tuple[Gtk.TreeRowReference, conn_info]] = {} self._power_levels_timer: int | None = None + self._bar_pixbuf_cache: dict[tuple[str, int, int], GdkPixbuf.Pixbuf] = {} self.manager.connect_signal("battery-created", self.on_battery_created) self.manager.connect_signal("battery-removed", self.on_battery_removed) @@ -138,7 +141,8 @@ def on_battery_created(self, _manager: Manager, obj_path: ObjectPath) -> None: if obj_path not in self._batteries: battery_proxy = Battery(obj_path=obj_path) self._batteries[obj_path] = battery_proxy - logging.debug(f"{obj_path} {battery_proxy['Percentage']}") + if logging.getLogger().isEnabledFor(logging.DEBUG): + logging.debug(f"{obj_path} {battery_proxy['Percentage']}") def on_battery_removed(self, _manager: Manager, obj_path: str) -> None: if obj_path in self._batteries: @@ -149,15 +153,14 @@ def search_func(self, model: Gtk.TreeModel, column: int, key: str, tree_iter: Gt row = self.get(tree_iter, "caption") if key.lower() in row["caption"].lower(): return False - logging.info(f"{model} {column} {key} {tree_iter}") + logging.info("%s %s %s %s", model, column, key, tree_iter) return True def filter_func(self, _model: Gtk.TreeModel, tree_iter: Gtk.TreeIter, _data: Any) -> bool: - row = self.get(tree_iter, "no_name", "device") - device = row["device"] - klass = get_minor_class(device["Class"]) if device is not None else None + row = self.get(tree_iter, "no_name", "klass") - if row["no_name"] and self.Config["hide-unnamed"] and klass not in (_("Keyboard"), _("Combo")): + if row["no_name"] and self.Config["hide-unnamed"] \ + and get_minor_class(row["klass"]) not in (_("Keyboard"), _("Combo")): logging.info("Hiding unnamed device") return False else: @@ -174,8 +177,8 @@ def drag_recv(self, _widget: Gtk.Widget, context: Gdk.DragContext, x: int, y: in if path: tree_iter = self.get_iter(path[0]) assert tree_iter is not None - device = self.get(tree_iter, "device")["device"] - command = f"blueman-sendto --device={device['Address']}" + address = self.get(tree_iter, "address")["address"] + command = f"blueman-sendto --device={address}" launch(command, paths=uris, name=_("File Sender")) context.finish(True, False, time) @@ -244,9 +247,11 @@ def _on_event_clicked(self, _widget: Gtk.Widget, event: Gdk.Event) -> bool: if event.type == Gdk.EventType._2BUTTON_PRESS and cast(Gdk.EventButton, event).button == 1: if self.menu.show_generic_connect_calc(row["uuids"]): + # rows only exist for devices that belong to self.Adapter + assert self.Adapter is not None if row["connected"]: self.menu.disconnect_service(row["device"]) - elif Adapter(obj_path=row["device"]["Adapter"])["Powered"]: + elif self.Adapter["Powered"]: self.menu.connect_service(row["device"]) if event.type == Gdk.EventType.BUTTON_PRESS and cast(Gdk.EventButton, event).button == 3: @@ -262,11 +267,11 @@ def _on_key_pressed(self, _widget: Gtk.Widget, event: Gdk.EventKey) -> bool: if not selected: return False - row = self.get(selected, "device") + row = self.get(selected, "address") if not row: return False - Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD).set_text(row["device"]["Address"], -1) + Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD).set_text(row["address"], -1) return True def _load_surface(self, icon_name: str, size: int) -> cairo.ImageSurface: @@ -345,16 +350,22 @@ def make_display_name(alias: str, klass: int, address: BtAddress) -> str: return alias @staticmethod - def get_device_class(device: Device, properties: Mapping[str, Any] | None = None) -> str: - klass_id = device["Class"] if properties is None else cast(int, properties["Class"]) + def get_device_class(klass_id: int) -> str: klass = get_minor_class(klass_id) if klass != _("Uncategorized"): return klass else: return get_major_class(klass_id) - def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: - properties = device.get_properties() + def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device, + properties: Mapping[str, Any] | None = None) -> None: + if properties is None: + properties = device.get_properties() + + klass_id = cast(int, properties["Class"]) + address = BtAddress(properties["Address"]) + # cache the filter/sort inputs before the animation setup below queries the filtered model + self.set(tree_iter, klass=klass_id, address=address, icon_name=properties["Icon"]) if not self.get(tree_iter, "initial_anim")["initial_anim"]: assert self.liststore is not None @@ -377,7 +388,6 @@ def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: uuids = tuple(cast(Iterable[str], properties["UUIDs"])) has_objpush = self._has_objpush(uuids) - klass_id = cast(int, properties["Class"]) klass = get_minor_class(klass_id) # Bluetooth >= 4 devices use Appearance property appearance = properties["Appearance"] @@ -391,22 +401,17 @@ def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None: surface = self._make_device_icon(properties["Icon"], properties["Paired"], properties["Connected"], properties["Trusted"], properties["Blocked"]) surface_object = SurfaceObject(surface) - address = BtAddress(properties["Address"]) - display_name = self.make_display_name(device.display_name, klass_id, address) + display_name = self.make_display_name(cast(str, properties["Alias"]).strip(), klass_id, address) caption = self.make_caption(display_name, description, address) self.set(tree_iter, caption=caption, alias=display_name, objpush=has_objpush, uuids=uuids, - device_surface=surface_object) - self.set(tree_iter, trusted=properties["Trusted"], paired=properties["Paired"], + device_surface=surface_object, trusted=properties["Trusted"], paired=properties["Paired"], connected=properties["Connected"], blocked=properties["Blocked"]) if properties["Connected"]: - self._monitor_power_levels(tree_iter, device, properties) + self._monitor_power_levels(tree_iter, device, address) - def _monitor_power_levels( - self, tree_iter: Gtk.TreeIter, device: Device, properties: Mapping[str, Any] | None = None - ) -> None: - address = BtAddress(device["Address"] if properties is None else properties["Address"]) + def _monitor_power_levels(self, tree_iter: Gtk.TreeIter, device: Device, address: BtAddress) -> None: if address in self._monitored_devices: return @@ -439,10 +444,10 @@ def _check_power_levels(self) -> bool: tree_iter = self.get_iter(row_ref.get_path()) assert tree_iter is not None - device = self.get(tree_iter, "device")["device"] + row = self.get(tree_iter, "device", "connected") - if device["Connected"]: - self._update_power_levels(tree_iter, device, cinfo) + if row["connected"]: + self._update_power_levels(tree_iter, row["device"], cinfo) else: cinfo.deinit() self._disable_power_levels(tree_iter) @@ -458,16 +463,13 @@ def _check_power_levels(self) -> bool: return False def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> None: - logging.info(f"{key} {value}") - - device = self.get(tree_iter, "device")["device"] - properties = None + logging.info("%s %s", key, value) if key in ("Blocked", "Connected", "Paired", "Trusted"): - properties = device.get_properties() - properties[key] = value - surface = self._make_device_icon(properties["Icon"], properties["Paired"], properties["Connected"], - properties["Trusted"], properties["Blocked"]) + row = self.get(tree_iter, "icon_name", "paired", "connected", "trusted", "blocked") + row[key.lower()] = value + surface = self._make_device_icon(row["icon_name"], row["paired"], row["connected"], + row["trusted"], row["blocked"]) self.set(tree_iter, device_surface=SurfaceObject(surface)) if key == "Trusted": @@ -477,13 +479,19 @@ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> Non self.set(tree_iter, paired=value) elif key == "Alias": - properties = device.get_properties() - properties[key] = value - address = BtAddress(properties["Address"]) - c = self.make_caption(value, self.get_device_class(device, properties), address) - name = self.make_display_name(device.display_name, properties["Class"], address) + row = self.get(tree_iter, "klass", "address") + address = BtAddress(row["address"]) + c = self.make_caption(value, self.get_device_class(row["klass"]), address) + name = self.make_display_name(value.strip(), row["klass"], address) self.set(tree_iter, caption=c, alias=name) + elif key == "Icon": + # an invalidated property (value None) falls back the same way Base.get_properties does + self.set(tree_iter, icon_name=value if value is not None else "blueman") + + elif key == "Class": + self.set(tree_iter, klass=value if value is not None else 0) + elif key == "UUIDs": uuids = tuple(cast(Iterable[str], value)) has_objpush = self._has_objpush(uuids) @@ -492,13 +500,12 @@ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> Non elif key == "Connected": self.set(tree_iter, connected=value) + address = BtAddress(self.get(tree_iter, "address")["address"]) if value: - assert properties is not None - self._monitor_power_levels(tree_iter, device, properties) + device = self.get(tree_iter, "device")["device"] + self._monitor_power_levels(tree_iter, device, address) else: self._disable_power_levels(tree_iter) - assert properties is not None - address = BtAddress(properties["Address"]) monitored = self._monitored_devices.pop(address, None) if monitored is not None: monitored[1].deinit() @@ -510,7 +517,7 @@ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> Non self.set(tree_iter, blocked=value) def _update_power_levels(self, tree_iter: Gtk.TreeIter, device: Device, cinfo: conn_info) -> None: - row = self.get(tree_iter, "cell_fader", "battery", "rssi", "lq", "tpl") + row = self.get(tree_iter, "cell_fader", "battery", "rssi", "tpl") bars = {} @@ -540,8 +547,13 @@ def _update_power_levels(self, tree_iter: Gtk.TreeIter, device: Device, cinfo: c for (name, perc) in bars.items(): if round(row[name], -1) != round(perc, -1): - path = PIXMAP_PATH / f"blueman-{name}-{int(round(perc, -1))}.png" - icon = GdkPixbuf.Pixbuf.new_from_file_at_scale(path.as_posix(), w, h, True) + level = int(round(perc, -1)) + cache_key = (name, level, w) + icon = self._bar_pixbuf_cache.get(cache_key) + if icon is None: + path = PIXMAP_PATH / f"blueman-{name}-{level}.png" + icon = GdkPixbuf.Pixbuf.new_from_file_at_scale(path.as_posix(), w, h, True) + self._bar_pixbuf_cache[cache_key] = icon self.set(tree_iter, **{name: perc, f"{name}_pb": icon}) def _disable_power_levels(self, tree_iter: Gtk.TreeIter) -> None: @@ -605,15 +617,15 @@ def tooltip_query(self, _tw: Gtk.Widget, x: int, y: int, _kb: bool, tooltip: Gtk tree_iter = self.get_iter(path[0]) assert tree_iter is not None - dt = self.get(tree_iter, "connected")["connected"] - if not dt: + row = self.get(tree_iter, "connected", "battery", "rssi", "tpl") + if not row["connected"]: return False lines = [_("Connected")] - battery = self.get(tree_iter, "battery")["battery"] - rssi = self.get(tree_iter, "rssi")["rssi"] - tpl = self.get(tree_iter, "tpl")["tpl"] + battery = row["battery"] + rssi = row["rssi"] + tpl = row["tpl"] if battery != 0: if path[1] == self.view_columns["battery_pb"]: diff --git a/blueman/gui/manager/ManagerDeviceMenu.py b/blueman/gui/manager/ManagerDeviceMenu.py index b20c10848..7ea6e2f11 100644 --- a/blueman/gui/manager/ManagerDeviceMenu.py +++ b/blueman/gui/manager/ManagerDeviceMenu.py @@ -8,7 +8,6 @@ from blueman.bluemantyping import ObjectPath from blueman.Functions import create_menuitem, e_ -from blueman.bluez.Adapter import Adapter from blueman.bluez.Network import AnyNetwork from blueman.bluez.Device import AnyDevice, Device from blueman.config.AutoConnectConfig import AutoConnectConfig @@ -266,7 +265,7 @@ def generate(self) -> None: if not selected: return row = self.Blueman.List.get(selected, "alias", "paired", "connected", "trusted", "objpush", "uuids", - "device", "blocked") + "address", "device", "blocked") else: (x, y) = self.Blueman.List.get_pointer() posdata = self.Blueman.List.get_path_at_pos(x, y) @@ -283,7 +282,7 @@ def generate(self) -> None: assert child_iter is not None row = self.Blueman.List.get(child_iter, "alias", "paired", "connected", "trusted", "objpush", "uuids", - "device", "blocked") + "address", "device", "blocked") self.SelectedDevice = row["device"] @@ -298,7 +297,10 @@ def generate(self) -> None: show_generic_connect = self.show_generic_connect_calc(row["uuids"]) - powered = Adapter(obj_path=self.SelectedDevice["Adapter"])["Powered"] + # list rows only exist for devices that belong to the list's adapter + adapter = self.Blueman.List.Adapter + assert adapter is not None + powered = adapter["Powered"] if not row["connected"] and show_generic_connect and powered: connect_item = create_menuitem(_("_Connect"), "bluetooth-symbolic") @@ -336,7 +338,7 @@ def generate(self) -> None: config = AutoConnectConfig() generic_service = ServiceUUID("00000000-0000-0000-0000-000000000000") object_path = self.SelectedDevice.get_object_path() - btaddress: BtAddress = self.SelectedDevice["Address"] + btaddress = BtAddress(row["address"]) generic_autoconnect = (object_path, str(generic_service)) in set(config["services"]) if row["connected"] or generic_autoconnect or autoconnect_items: diff --git a/test/gui/manager/test_manager_device_list.py b/test/gui/manager/test_manager_device_list.py index bdf470cf9..eb58437c8 100644 --- a/test/gui/manager/test_manager_device_list.py +++ b/test/gui/manager/test_manager_device_list.py @@ -1,13 +1,15 @@ from pathlib import Path +import random import sys import types from unittest import TestCase -from unittest.mock import Mock, patch +from unittest.mock import MagicMock, Mock, patch import gi gi.require_version("Gtk", "3.0") -from gi.repository import Gtk +gi.require_version("Gdk", "3.0") +from gi.repository import Gdk, Gtk constants = types.ModuleType("blueman.Constants") constants.BIN_DIR = Path("/tmp") @@ -25,6 +27,7 @@ class FakeDevice: def __init__(self) -> None: self.item_reads: list[str] = [] + self.getall_reads = 0 self.properties = { "Address": "AA:BB:CC:DD:EE:FF", "Alias": "Keyboard", @@ -39,18 +42,35 @@ def __init__(self) -> None: } def get_properties(self): + self.getall_reads += 1 return dict(self.properties) def __getitem__(self, key): self.item_reads.append(key) return self.properties[key] + def get_object_path(self): + return "/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF" + class FakeManagerDeviceList: Adapter = None def __init__(self) -> None: - self.values = {"initial_anim": True, "device": FakeDevice(), "objpush": False, "uuids": ()} + # column state as row_setup_event would have cached it + self.values = { + "initial_anim": True, + "device": FakeDevice(), + "objpush": False, + "uuids": (), + "klass": 0, + "address": "AA:BB:CC:DD:EE:FF", + "icon_name": "input-keyboard", + "paired": False, + "connected": False, + "trusted": False, + "blocked": False, + } self.set_calls: list[dict[str, object]] = [] self.monitor_calls = [] self.disabled = [] @@ -67,8 +87,8 @@ def _make_device_icon(self, *args): self.icon_args = args return Mock() - def _monitor_power_levels(self, tree_iter, device, properties=None): - self.monitor_calls.append((tree_iter, device, properties)) + def _monitor_power_levels(self, tree_iter, device, address): + self.monitor_calls.append((tree_iter, device, address)) def _check_power_levels(self): return False @@ -83,6 +103,7 @@ def make_caption(self, display_name, description, address): return f"{display_name} {description} {address}" _has_objpush = staticmethod(ManagerDeviceList._has_objpush) + get_device_class = staticmethod(ManagerDeviceList.get_device_class) class TestManagerDeviceListProperties(TestCase): @@ -94,18 +115,34 @@ def test_row_setup_uses_get_all_properties(self): fake = FakeManagerDeviceList() device = FakeDevice() device.properties["Connected"] = True + device.properties["Class"] = 0x5A020C + device.properties["Icon"] = "phone" tree_iter = object() ManagerDeviceList.row_setup_event(fake, tree_iter, device) self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 1) self.assertFalse(fake.values["trusted"]) self.assertFalse(fake.values["paired"]) self.assertTrue(fake.values["connected"]) self.assertFalse(fake.values["blocked"]) self.assertEqual(fake.values["uuids"], ()) + self.assertEqual(fake.values["klass"], 0x5A020C) + self.assertEqual(fake.values["address"], device.properties["Address"]) + self.assertEqual(fake.values["icon_name"], "phone") self.assertEqual(len(fake.monitor_calls), 1) - self.assertEqual(fake.monitor_calls[0][2]["Address"], device.properties["Address"]) + self.assertEqual(fake.monitor_calls[0][2], device.properties["Address"]) + + def test_row_setup_reuses_provided_properties(self): + fake = FakeManagerDeviceList() + device = FakeDevice() + tree_iter = object() + + ManagerDeviceList.row_setup_event(fake, tree_iter, device, device.get_properties()) + + self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 1) def test_row_update_batches_icon_properties(self): fake = FakeManagerDeviceList() @@ -115,6 +152,7 @@ def test_row_update_batches_icon_properties(self): ManagerDeviceList.row_update_event(fake, tree_iter, "Trusted", True) self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 0) self.assertEqual(fake.icon_args, ("input-keyboard", False, False, True, False)) self.assertTrue(fake.values["trusted"]) @@ -128,10 +166,33 @@ def test_row_update_connected_false_uses_batched_address(self): ManagerDeviceList.row_update_event(fake, tree_iter, "Connected", False) self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 0) self.assertEqual(fake.disabled, [tree_iter]) cinfo.deinit.assert_called_once_with() self.assertEqual(fake._monitored_devices, {}) + def test_row_update_alias_uses_cached_state(self): + fake = FakeManagerDeviceList() + device = fake.values["device"] + tree_iter = object() + + ManagerDeviceList.row_update_event(fake, tree_iter, "Alias", " New Name ") + + self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 0) + self.assertEqual(fake.values["alias"], "New Name") + self.assertIn("AA:BB:CC:DD:EE:FF", fake.values["caption"]) + + def test_row_update_icon_and_class_update_cache(self): + fake = FakeManagerDeviceList() + tree_iter = object() + + ManagerDeviceList.row_update_event(fake, tree_iter, "Icon", "phone") + ManagerDeviceList.row_update_event(fake, tree_iter, "Class", 42) + + self.assertEqual(fake.values["icon_name"], "phone") + self.assertEqual(fake.values["klass"], 42) + def test_row_update_uuids_updates_cached_state(self): fake = FakeManagerDeviceList() tree_iter = object() @@ -179,8 +240,8 @@ def test_monitor_power_levels_starts_one_timer(self): patch("blueman.gui.manager.ManagerDeviceList.conn_info") as conn_info_cls, \ patch("blueman.gui.manager.ManagerDeviceList.Gtk.TreeRowReference.new", return_value=Mock()), \ patch("blueman.gui.manager.ManagerDeviceList.GLib.timeout_add", return_value=12) as timeout_add: - ManagerDeviceList._monitor_power_levels(fake, tree_iter, device, device.get_properties()) - ManagerDeviceList._monitor_power_levels(fake, tree_iter, device, device.get_properties()) + ManagerDeviceList._monitor_power_levels(fake, tree_iter, device, "AA:BB:CC:DD:EE:FF") + ManagerDeviceList._monitor_power_levels(fake, tree_iter, device, "AA:BB:CC:DD:EE:FF") conn_info_cls.assert_called_once_with("AA:BB:CC:DD:EE:FF", "hci0") timeout_add.assert_called_once_with(1000, fake._check_power_levels) @@ -201,3 +262,306 @@ def test_check_power_levels_stops_when_no_devices_remain(self): cinfo.deinit.assert_called_once_with() self.assertEqual(fake._monitored_devices, {}) self.assertIsNone(fake._power_levels_timer) + + def test_check_power_levels_uses_cached_connected_state(self): + row_ref = Mock() + row_ref.valid.return_value = True + cinfo = Mock() + fake = FakeManagerDeviceList() + device = fake.values["device"] + fake.values["connected"] = True + fake.get_iter = Mock(return_value=object()) + fake._update_power_levels = Mock() + fake._monitored_devices = {"AA:BB:CC:DD:EE:FF": (row_ref, cinfo)} + fake._power_levels_timer = 12 + + keep = ManagerDeviceList._check_power_levels(fake) + + self.assertTrue(keep) + self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 0) + fake._update_power_levels.assert_called_once() + self.assertIs(fake._update_power_levels.call_args.args[1], device) + self.assertEqual(len(fake._monitored_devices), 1) + + def test_check_power_levels_drops_disconnected_device(self): + row_ref = Mock() + row_ref.valid.return_value = True + cinfo = Mock() + fake = FakeManagerDeviceList() + device = fake.values["device"] + fake.values["connected"] = False + fake.get_iter = Mock(return_value=object()) + fake._update_power_levels = Mock() + fake._monitored_devices = {"AA:BB:CC:DD:EE:FF": (row_ref, cinfo)} + fake._power_levels_timer = 12 + + keep = ManagerDeviceList._check_power_levels(fake) + + self.assertFalse(keep) + self.assertEqual(device.item_reads, []) + cinfo.deinit.assert_called_once_with() + self.assertEqual(len(fake.disabled), 1) + self.assertEqual(fake._monitored_devices, {}) + self.assertIsNone(fake._power_levels_timer) + + +class TestManagerDeviceListEventPaths(TestCase): + def test_get_device_class_uncategorized_falls_back_to_major(self): + self.assertEqual(ManagerDeviceList.get_device_class(0x0540), "Keyboard") + self.assertEqual(ManagerDeviceList.get_device_class(0x0500), "Peripheral") + + def test_search_func_matches_caption(self): + fake = FakeManagerDeviceList() + fake.values["caption"] = "My Keyboard AA:BB" + # search_func returns False on match (Gtk convention) + self.assertFalse(ManagerDeviceList.search_func(fake, Mock(), 0, "keyboard", object())) + self.assertTrue(ManagerDeviceList.search_func(fake, Mock(), 0, "mouse", object())) + + def test_on_battery_created_skips_dbus_read_without_debug_logging(self): + fake = FakeManagerDeviceList() + fake._batteries = {} + battery = MagicMock() + + with patch("blueman.gui.manager.ManagerDeviceList.Battery", return_value=battery): + ManagerDeviceList.on_battery_created(fake, Mock(), "/org/bluez/hci0/dev_X") + + self.assertIn("/org/bluez/hci0/dev_X", fake._batteries) + battery.__getitem__.assert_not_called() + + def test_on_battery_created_still_reads_percentage_with_debug_logging(self): + import logging + fake = FakeManagerDeviceList() + fake._batteries = {} + battery = MagicMock() + root = logging.getLogger() + old_level = root.level + root.setLevel(logging.DEBUG) + try: + with patch("blueman.gui.manager.ManagerDeviceList.Battery", return_value=battery), \ + self.assertLogs(level=logging.DEBUG): + ManagerDeviceList.on_battery_created(fake, Mock(), "/org/bluez/hci0/dev_Y") + finally: + root.setLevel(old_level) + + battery.__getitem__.assert_called_once_with("Percentage") + + def test_drag_recv_uses_cached_address(self): + fake = FakeManagerDeviceList() + device = fake.values["device"] + fake.get_path_at_pos = Mock(return_value=(Mock(), Mock())) + fake.get_iter = Mock(return_value=object()) + selection = Mock() + selection.get_uris.return_value = ["file:///tmp/f"] + context = Mock() + + with patch("blueman.gui.manager.ManagerDeviceList.launch") as launch: + ManagerDeviceList.drag_recv(fake, Mock(), context, 1, 2, selection, 0, 123) + + self.assertEqual(device.item_reads, []) + launch.assert_called_once() + self.assertIn("--device=AA:BB:CC:DD:EE:FF", launch.call_args.args[0]) + + def _make_click_fake(self, connected, powered): + fake = FakeManagerDeviceList() + fake.values["connected"] = connected + fake.values["uuids"] = ("00001124-0000-1000-8000-00805f9b34fb",) # HID + fake.get_path_at_pos = Mock(return_value=(Mock(),)) + fake.filter = Mock() + fake.filter.get_iter.return_value = object() + fake.filter.convert_iter_to_child_iter.return_value = object() + fake.menu = Mock() + fake.menu.show_generic_connect_calc = lambda uuids: True + fake.Adapter = MagicMock() + fake.Adapter.__getitem__ = Mock(return_value=powered) + fake.Blueman = Mock() + return fake + + def test_double_click_disconnects_connected_device(self): + fake = self._make_click_fake(connected=True, powered=True) + device = fake.values["device"] + event = Mock(type=Gdk.EventType._2BUTTON_PRESS, button=1, x=1.0, y=2.0) + + ManagerDeviceList._on_event_clicked(fake, Mock(), event) + + self.assertEqual(device.item_reads, []) + fake.menu.disconnect_service.assert_called_once_with(device) + fake.Adapter.__getitem__.assert_not_called() + + def test_double_click_connects_when_adapter_powered(self): + fake = self._make_click_fake(connected=False, powered=True) + device = fake.values["device"] + event = Mock(type=Gdk.EventType._2BUTTON_PRESS, button=1, x=1.0, y=2.0) + + ManagerDeviceList._on_event_clicked(fake, Mock(), event) + + self.assertEqual(device.item_reads, []) + fake.Adapter.__getitem__.assert_called_once_with("Powered") + fake.menu.connect_service.assert_called_once_with(device) + + def test_ctrl_c_copies_cached_address(self): + fake = FakeManagerDeviceList() + device = fake.values["device"] + fake.selected = Mock(return_value=object()) + event = Mock(state=Gdk.ModifierType.CONTROL_MASK, keyval=Gdk.KEY_c) + + with patch("blueman.gui.manager.ManagerDeviceList.Gtk") as gtk: + handled = ManagerDeviceList._on_key_pressed(fake, Mock(), event) + + self.assertTrue(handled) + self.assertEqual(device.item_reads, []) + gtk.Clipboard.get.return_value.set_text.assert_called_once_with("AA:BB:CC:DD:EE:FF", -1) + + def test_tooltip_uses_batched_row_read(self): + fake = FakeManagerDeviceList() + device = fake.values["device"] + fake.values.update(connected=True, battery=80.0, rssi=50.0, tpl=50.0) + path, col = Mock(), Mock() + fake.get_path_at_pos = Mock(return_value=(path, col)) + fake.get_iter = Mock(return_value=object()) + fake.view_columns = {"device_surface": Mock(), "battery_pb": col, "rssi_pb": Mock(), "tpl_pb": Mock()} + fake.tooltip_row = path + fake.tooltip_col = col + tooltip = Mock() + + shown = ManagerDeviceList.tooltip_query(fake, Mock(), 1, 2, False, tooltip) + + self.assertTrue(shown) + self.assertEqual(device.item_reads, []) + markup = tooltip.set_markup.call_args.args[0] + self.assertIn("Battery: 80%", markup) + + +class TestManagerDeviceListFilter(TestCase): + def _make_fake(self, no_name, klass, hide_unnamed): + fake = FakeManagerDeviceList() + fake.values["no_name"] = no_name + fake.values["klass"] = klass + fake.Config = {"hide-unnamed": hide_unnamed} + return fake + + def test_filter_hides_unnamed_non_input_device(self): + fake = self._make_fake(no_name=True, klass=0x5A020C, hide_unnamed=True) # Smartphone + self.assertFalse(ManagerDeviceList.filter_func(fake, Mock(), object(), None)) + self.assertEqual(fake.values["device"].item_reads, []) + + def test_filter_keeps_unnamed_keyboard_and_combo(self): + for klass in (0x0540, 0x05C0): # Keyboard, Combo + fake = self._make_fake(no_name=True, klass=klass, hide_unnamed=True) + self.assertTrue(ManagerDeviceList.filter_func(fake, Mock(), object(), None)) + + def test_filter_keeps_named_devices_and_respects_config(self): + fake = self._make_fake(no_name=False, klass=0x5A020C, hide_unnamed=True) + self.assertTrue(ManagerDeviceList.filter_func(fake, Mock(), object(), None)) + fake = self._make_fake(no_name=True, klass=0x5A020C, hide_unnamed=False) + self.assertTrue(ManagerDeviceList.filter_func(fake, Mock(), object(), None)) + + +class TestUpdatePowerLevels(TestCase): + def test_bar_pixbufs_are_cached_per_level(self): + fake = FakeManagerDeviceList() + device = fake.values["device"] + fake.values.update(cell_fader=Mock(), battery=0.0, rssi=0.0, tpl=0.0) + fake.get_scale_factor = Mock(return_value=1) + fake._prepare_fader = Mock() + fake._bar_pixbuf_cache = {} + fake._batteries = {} + cinfo = Mock() + cinfo.failed = True # rssi/tpl fall back to 100.0 + tree_iter = object() + + with patch("blueman.gui.manager.ManagerDeviceList.GdkPixbuf") as gdkpixbuf: + ManagerDeviceList._update_power_levels(fake, tree_iter, device, cinfo) + first_loads = gdkpixbuf.Pixbuf.new_from_file_at_scale.call_count + # simulate a second device reaching the same levels + fake.values.update(battery=0.0, rssi=0.0, tpl=0.0) + ManagerDeviceList._update_power_levels(fake, tree_iter, device, cinfo) + second_loads = gdkpixbuf.Pixbuf.new_from_file_at_scale.call_count + + self.assertEqual(first_loads, 2) # rssi + tpl + self.assertEqual(second_loads, 2) # served from cache + self.assertEqual(fake.values["rssi"], 100.0) + self.assertEqual(fake.values["tpl"], 100.0) + self.assertIsNotNone(fake.values["rssi_pb"]) + self.assertIsNotNone(fake.values["tpl_pb"]) + + +class TestRowUpdateFuzz(TestCase): + """Deterministic randomized state-machine run over row_update_event. + + Feeds random property-changed sequences and checks two invariants after + every event: the cached liststore state matches a reference model, and the + device proxy is never read (no D-Bus round-trips on the update path). + """ + + KEYBOARD_UUIDS = ("00001105-0000-1000-8000-00805f9b34fb",) + + @staticmethod + def _uuid16(short): + return f"0000{short:04x}-0000-1000-8000-00805f9b34fb" + + def test_randomized_updates_keep_cached_state_consistent(self): + rng = random.Random(20260708) + fake = FakeManagerDeviceList() + fake.filter = Mock() + device = fake.values["device"] + tree_iter = object() + + model = { + "trusted": False, "paired": False, "connected": False, "blocked": False, + "icon_name": "input-keyboard", "klass": 0, "objpush": False, + } + icon_keys = ("Blocked", "Connected", "Paired", "Trusted") + + for _ in range(2000): + key = rng.choice(icon_keys + ("Alias", "UUIDs", "Icon", "Class", "Name")) + + if key in icon_keys: + value = rng.random() < 0.5 + model[key.lower()] = value + elif key == "Alias": + value = rng.choice(["", " ", "Dev ", " &x ", "名前 ", "plain"]) + elif key == "UUIDs": + shorts = rng.sample(range(0x1101, 0x1120), rng.randint(0, 5)) + value = [self._uuid16(s) for s in shorts] + model["objpush"] = 0x1105 in shorts + elif key == "Icon": + value = rng.choice(["phone", "audio-headset", None]) + model["icon_name"] = value if value is not None else "blueman" + elif key == "Class": + value = rng.choice([rng.randrange(0, 0x1000000), None]) + model["klass"] = value if value is not None else 0 + else: # Name + value = "Some Name" + + ManagerDeviceList.row_update_event(fake, tree_iter, key, value) + + self.assertEqual(device.item_reads, [], f"D-Bus read after {key}") + self.assertEqual(device.getall_reads, 0, f"GetAll after {key}") + for column in ("trusted", "paired", "connected", "blocked", "icon_name", "klass", "objpush"): + self.assertEqual(fake.values[column], model[column], f"{column} after {key}") + if key in icon_keys: + self.assertEqual( + fake.icon_args, + (model["icon_name"], model["paired"], model["connected"], + model["trusted"], model["blocked"])) + if key == "Alias": + self.assertEqual(fake.values["alias"], value.strip()) + + def test_has_objpush_fuzz(self): + rng = random.Random(20260708) + for _ in range(500): + shorts = rng.sample(range(0x1000, 0x2000), rng.randint(0, 8)) + uuids = [self._uuid16(s) for s in shorts] + self.assertEqual(ManagerDeviceList._has_objpush(uuids), 0x1105 in shorts) + + def test_make_display_name_fuzz(self): + rng = random.Random(20260708) + for _ in range(500): + address = ":".join(f"{rng.randrange(256):02X}" for _ in range(6)) + alias = rng.choice([address, address.replace(":", "-"), "Headphones", f"dev-{rng.randrange(100)}"]) + result = ManagerDeviceList.make_display_name(alias, 0, address) + if alias.replace("-", ":") == address: + self.assertEqual(result, "Unnamed device") + else: + self.assertEqual(result, alias) From 24cdca00c6055d207822f8addbd4fb4300f1bf20 Mon Sep 17 00:00:00 2001 From: Geraldo Netto Date: Wed, 8 Jul 2026 21:16:15 +0200 Subject: [PATCH 8/9] test: add device list coverage and fuzz-style tests Extend the headless unit tests to cover the paths the perf work touches: GenericList.set batching (row-changed emission count against a real ListStore), single-GetAll add_device including the foreign-adapter and named-device branches, and DeviceSelectorList row setup. Add deterministic fuzz-style tests: a 2000-event randomized property state machine over row_update_event asserting the cached row state always matches a reference model and the device proxy is never read, plus randomized sweeps over _has_objpush and make_display_name. Changed-line coverage vs main: ManagerDeviceList 94%, GenericList, DeviceList and DeviceSelectorList 100% (ManagerDeviceMenu.generate still needs a real Gtk.Menu and GSettings schemas, so it stays untested, as before). Also list the test modules in EXTRA_DIST so release tarballs include them. Co-Authored-By: Claude Fable 5 --- test/gui/Makefile.am | 3 +- test/gui/manager/Makefile.am | 3 +- test/gui/test_device_lists.py | 155 ++++++++++++++++++++++++++++++++++ 3 files changed, 159 insertions(+), 2 deletions(-) create mode 100644 test/gui/test_device_lists.py diff --git a/test/gui/Makefile.am b/test/gui/Makefile.am index 3911e592d..c026c8482 100644 --- a/test/gui/Makefile.am +++ b/test/gui/Makefile.am @@ -4,4 +4,5 @@ SUBDIRS = \ EXTRA_DIST = \ __init__.py \ - test_imports.py + test_imports.py \ + test_device_lists.py diff --git a/test/gui/manager/Makefile.am b/test/gui/manager/Makefile.am index 3e92b4be6..cc64ba492 100644 --- a/test/gui/manager/Makefile.am +++ b/test/gui/manager/Makefile.am @@ -1,3 +1,4 @@ EXTRA_DIST = \ __init__.py \ - test_imports.py + test_imports.py \ + test_manager_device_list.py diff --git a/test/gui/test_device_lists.py b/test/gui/test_device_lists.py new file mode 100644 index 000000000..139475038 --- /dev/null +++ b/test/gui/test_device_lists.py @@ -0,0 +1,155 @@ +from pathlib import Path +import sys +import types +from unittest import TestCase +from unittest.mock import Mock, patch + +import gi + +gi.require_version("Gtk", "3.0") +from gi.repository import Gtk + +constants = types.ModuleType("blueman.Constants") +constants.BIN_DIR = Path("/tmp") +constants.BLUETOOTHD_PATH = Path("/tmp/bluetoothd") +constants.ICON_PATH = Path("/tmp") +constants.PIXMAP_PATH = Path("/tmp") +constants.UI_PATH = Path("/tmp") +sys.modules.setdefault("blueman.Constants", constants) + +from blueman.gui.GenericList import GenericList +from blueman.gui.DeviceList import DeviceList +from blueman.gui.DeviceSelectorList import DeviceSelectorList + + +class FakeDevice: + def __init__(self) -> None: + self.item_reads: list[str] = [] + self.getall_reads = 0 + self.properties = { + "Adapter": "/org/bluez/hci0", + "Address": "AA:BB:CC:DD:EE:FF", + "Alias": " Keyboard ", + "Icon": "input-keyboard", + "Paired": False, + "Trusted": False, + } + + def get_properties(self): + self.getall_reads += 1 + return dict(self.properties) + + def __getitem__(self, key): + self.item_reads.append(key) + return self.properties[key] + + def get_object_path(self): + return "/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF" + + +class TestGenericListSet(TestCase): + def _make_fake(self): + fake = types.SimpleNamespace() + fake.liststore = Gtk.ListStore(str, int, bool) + fake.list_col_order = {"name": 0, "count": 1, "flag": 2} + return fake + + def test_set_batches_columns_into_one_row_changed(self): + fake = self._make_fake() + tree_iter = fake.liststore.append(["a", 1, False]) + emissions = [] + fake.liststore.connect("row-changed", lambda *_args: emissions.append(1)) + + GenericList.set(fake, tree_iter, name="b", count=2, flag=True) + + self.assertEqual(len(emissions), 1) + self.assertEqual(fake.liststore.get_value(tree_iter, 0), "b") + self.assertEqual(fake.liststore.get_value(tree_iter, 1), 2) + self.assertTrue(fake.liststore.get_value(tree_iter, 2)) + + def test_set_unknown_column_raises_key_error(self): + fake = self._make_fake() + tree_iter = fake.liststore.append(["a", 1, False]) + + with self.assertRaises(KeyError): + GenericList.set(fake, tree_iter, bogus=1) + + +class TestDeviceListAddDevice(TestCase): + def _make_fake(self): + fake = types.SimpleNamespace() + fake.Adapter = Mock() + fake.Adapter.get_object_path.return_value = "/org/bluez/hci0" + fake.append = Mock(return_value=object()) + fake.row_setup_event = Mock() + fake.get_selected_device = Mock(return_value=object()) + fake.selection = Mock() + return fake + + def test_add_device_does_a_single_get_all(self): + fake = self._make_fake() + device = FakeDevice() + + with patch("blueman.gui.DeviceList.Device", return_value=device): + DeviceList.add_device(fake, "/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF") + + self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 1) + self.assertTrue(fake.append.call_args.kwargs["no_name"]) + fake.row_setup_event.assert_called_once() + tree_iter, dev_arg, properties = fake.row_setup_event.call_args.args + self.assertIs(dev_arg, device) + self.assertEqual(properties["Address"], "AA:BB:CC:DD:EE:FF") + + def test_add_device_detects_named_device(self): + fake = self._make_fake() + device = FakeDevice() + device.properties["Name"] = "Keyboard" + + with patch("blueman.gui.DeviceList.Device", return_value=device): + DeviceList.add_device(fake, "/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF") + + self.assertFalse(fake.append.call_args.kwargs["no_name"]) + + def test_add_device_skips_foreign_adapter_without_extra_reads(self): + fake = self._make_fake() + device = FakeDevice() + device.properties["Adapter"] = "/org/bluez/hci1" + + with patch("blueman.gui.DeviceList.Device", return_value=device): + DeviceList.add_device(fake, "/org/bluez/hci1/dev_AA_BB_CC_DD_EE_FF") + + self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 1) + fake.append.assert_not_called() + fake.row_setup_event.assert_not_called() + + +class TestDeviceSelectorListRowSetup(TestCase): + def test_row_setup_uses_get_all_properties(self): + fake = types.SimpleNamespace() + updates = [] + fake.row_update_event = lambda tree_iter, key, value: updates.append((key, value)) + device = FakeDevice() + tree_iter = object() + + DeviceSelectorList.row_setup_event(fake, tree_iter, device) + + self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 1) + self.assertEqual(updates, [ + ("Trusted", False), + ("Paired", False), + ("Alias", "Keyboard"), + ("Icon", "input-keyboard"), + ]) + + def test_row_setup_reuses_provided_properties(self): + fake = types.SimpleNamespace() + fake.row_update_event = lambda tree_iter, key, value: None + device = FakeDevice() + + DeviceSelectorList.row_setup_event(fake, object(), device, device.get_properties()) + + self.assertEqual(device.item_reads, []) + self.assertEqual(device.getall_reads, 1) From 78cfefcdd226bbe39ee068b33a08b5f6a1f52510 Mon Sep 17 00:00:00 2001 From: Geraldo Netto Date: Wed, 8 Jul 2026 21:33:02 +0200 Subject: [PATCH 9/9] refactor: use != instead of negated equality in add_device Per review. Co-Authored-By: Claude Fable 5 --- blueman/gui/DeviceList.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/blueman/gui/DeviceList.py b/blueman/gui/DeviceList.py index 815ea3c48..1bd0fad91 100644 --- a/blueman/gui/DeviceList.py +++ b/blueman/gui/DeviceList.py @@ -225,7 +225,7 @@ def add_device(self, object_path: ObjectPath) -> None: device = Device(obj_path=object_path) properties = device.get_properties() # device belongs to another adapter - if not self.Adapter or not properties["Adapter"] == self.Adapter.get_object_path(): + if not self.Adapter or properties["Adapter"] != self.Adapter.get_object_path(): return logging.info("adding new device")