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)