diff --git a/blueman/gui/DeviceList.py b/blueman/gui/DeviceList.py
index 282d2e401..1bd0fad91 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 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/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 = {}
diff --git a/blueman/gui/manager/ManagerDeviceList.py b/blueman/gui/manager/ManagerDeviceList.py
index 2514a6ec6..ad9498cb0 100644
--- a/blueman/gui/manager/ManagerDeviceList.py
+++ b/blueman/gui/manager/ManagerDeviceList.py
@@ -1,12 +1,11 @@
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
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
@@ -67,6 +66,10 @@ 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": "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},
@@ -81,7 +84,9 @@ 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._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)
@@ -136,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:
@@ -147,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:
@@ -172,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)
@@ -192,7 +197,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)
@@ -233,7 +238,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
@@ -241,10 +246,12 @@ 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"]):
+ # 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:
@@ -260,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:
@@ -343,14 +350,23 @@ 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(klass_id: int) -> str:
+ 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,
+ 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"])
- def row_setup_event(self, tree_iter: Gtk.TreeIter, device: Device) -> None:
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 +386,39 @@ 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'])
+ uuids = tuple(cast(Iterable[str], properties["UUIDs"]))
+ has_objpush = self._has_objpush(uuids)
+ 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'])
+ 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, device_surface=surface_object)
-
- 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)
+ self.set(tree_iter, caption=caption, alias=display_name, objpush=has_objpush, uuids=uuids,
+ device_surface=surface_object, trusted=properties["Trusted"], paired=properties["Paired"],
+ connected=properties["Connected"], blocked=properties["Blocked"])
- if device["Connected"]:
- self._monitor_power_levels(tree_iter, device)
+ if properties["Connected"]:
+ self._monitor_power_levels(tree_iter, device, address)
- 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, address: BtAddress) -> None:
+ 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 +428,87 @@ 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:
+ 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()
+ remove_addresses.append(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
+ row = self.get(tree_iter, "device", "connected")
- device = self.get(tree_iter, "device")["device"]
+ if row["connected"]:
+ self._update_power_levels(tree_iter, row["device"], cinfo)
+ else:
+ cinfo.deinit()
+ self._disable_power_levels(tree_iter)
+ remove_addresses.append(address)
+
+ for address in remove_addresses:
+ 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
- def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> None:
- logging.info(f"{key} {value}")
+ self._power_levels_timer = None
+ return False
- device = self.get(tree_iter, "device")["device"]
+ def row_update_event(self, tree_iter: Gtk.TreeIter, key: str, value: Any) -> None:
+ logging.info("%s %s", key, value)
if key in ("Blocked", "Connected", "Paired", "Trusted"):
- surface = self._make_device_icon(device["Icon"], device["Paired"], device["Connected"], device["Trusted"],
- device["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":
- 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"])
+ 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":
- has_objpush = self._has_objpush(device)
- 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)
+ address = BtAddress(self.get(tree_iter, "address")["address"])
if value:
- self._monitor_power_levels(tree_iter, device)
+ device = self.get(tree_iter, "device")["device"]
+ self._monitor_power_levels(tree_iter, device, address)
else:
self._disable_power_levels(tree_iter)
+ 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()
@@ -496,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 = {}
@@ -526,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:
@@ -591,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"]:
@@ -651,11 +677,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/blueman/gui/manager/ManagerDeviceMenu.py b/blueman/gui/manager/ManagerDeviceMenu.py
index eb0645548..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
@@ -265,8 +264,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",
+ "address", "device", "blocked")
else:
(x, y) = self.Blueman.List.get_pointer()
posdata = self.Blueman.List.get_path_at_pos(x, y)
@@ -282,8 +281,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",
+ "address", "device", "blocked")
self.SelectedDevice = row["device"]
@@ -296,9 +295,12 @@ 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"]
+ # 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/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
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/manager/test_manager_device_list.py b/test/gui/manager/test_manager_device_list.py
new file mode 100644
index 000000000..eb58437c8
--- /dev/null
+++ b/test/gui/manager/test_manager_device_list.py
@@ -0,0 +1,567 @@
+from pathlib import Path
+import random
+import sys
+import types
+from unittest import TestCase
+from unittest.mock import MagicMock, Mock, patch
+
+import gi
+
+gi.require_version("Gtk", "3.0")
+gi.require_version("Gdk", "3.0")
+from gi.repository import Gdk, 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.getall_reads = 0
+ 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):
+ 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:
+ # 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 = []
+ 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, address):
+ self.monitor_calls.append((tree_iter, device, address))
+
+ 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)
+ get_device_class = staticmethod(ManagerDeviceList.get_device_class)
+
+
+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
+ 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], 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()
+ device = fake.values["device"]
+ tree_iter = object()
+
+ 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"])
+
+ 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(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()
+ 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):
+ 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, "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)
+ 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)
+
+ 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)
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)