diff --git a/blueman/Functions.py b/blueman/Functions.py
index 2e48e8aab..7a13a4e28 100644
--- a/blueman/Functions.py
+++ b/blueman/Functions.py
@@ -23,6 +23,7 @@
import re
import os
import pathlib
+import shutil
import sys
import errno
from gettext import gettext as _
@@ -49,6 +50,7 @@
from gi.repository import Gdk
from gi.repository import GdkPixbuf
from gi.repository import Gio
+from gi.repository import GLib
__all__ = ["check_bluetooth_status", "launch", "setup_icon_path", "adapter_path_to_name", "e_", "bmexit",
"format_bytes", "create_menuitem", "have", "set_proc_title", "create_logger", "create_parser", "open_rfcomm",
@@ -59,9 +61,8 @@ def check_bluetooth_status(message: str, exitfunc: Callable[[], Any]) -> None:
try:
applet = AppletService()
powermanager = AppletPowerManagerService()
- except DBusProxyFailed as e:
- logging.exception(e)
- print("Blueman applet needs to be running")
+ except DBusProxyFailed:
+ logging.exception("Blueman applet needs to be running")
exitfunc()
return
@@ -84,7 +85,7 @@ def check_bluetooth_status(message: str, exitfunc: Callable[[], Any]) -> None:
powermanager.set_bluetooth_status(True)
if not powermanager.get_bluetooth_status():
- print('Failed to enable bluetooth')
+ logging.error("Failed to enable bluetooth")
exitfunc()
@@ -95,8 +96,17 @@ def launch(
icon_name: str | None = None,
name: str = "blueman",
sn: bool = True,
+ args: Iterable[str] | None = None,
) -> bool:
- """Launch a gui app with startup notification"""
+ """Launch a gui app with startup notification.
+
+ ``cmd`` is the program to run. Pass options as an ``args`` iterable rather
+ than embedding them in ``cmd``: each program token and argument is then
+ shell-quoted individually so argument boundaries follow an argv contract
+ instead of GLib command-line parsing. The legacy form, where ``cmd`` itself
+ is a full command line, is still accepted when ``args`` is omitted but is
+ deprecated.
+ """
context = None
gtktimestamp = Gtk.get_current_event_time()
if gtktimestamp == 0:
@@ -122,6 +132,13 @@ def launch(
else:
command = pathlib.Path(cmd).expanduser()
+ if args is not None:
+ # argv contract: quote each token so spaces/quotes/separators in
+ # arguments cannot cross argument boundaries.
+ command_line = " ".join(GLib.shell_quote(arg) for arg in (command.as_posix(), *args))
+ else:
+ command_line = command.as_posix()
+
if paths:
files: list[Gio.File] | None = [Gio.File.new_for_commandline_arg(p) for p in paths]
else:
@@ -130,7 +147,7 @@ def launch(
if icon_name and context is not None:
context.set_icon_name(icon_name)
- appinfo = Gio.AppInfo.create_from_commandline(command.as_posix(), name, flags)
+ appinfo = Gio.AppInfo.create_from_commandline(command_line, name, flags)
launched: bool = appinfo.launch(files, context)
if not launched:
@@ -168,10 +185,10 @@ def format_bytes(size: float) -> tuple[float, str]:
if size < 1024:
ret = size
suffix = _("B")
- elif 1024 < size < (1024 * 1024):
+ elif size < (1024 * 1024):
ret = size / 1024
suffix = _("KB")
- elif (1024 * 1024) < size < (1024 * 1024 * 1024):
+ elif size < (1024 * 1024 * 1024):
ret = size / (1024 * 1024)
suffix = _("MB")
else:
@@ -207,23 +224,39 @@ def create_menuitem(
def have(t: str) -> pathlib.Path | None:
- pathstr = os.environ['PATH'] + ':/sbin:/usr/sbin'
- for path in [pathlib.Path(p, t) for p in pathstr.split(":")]:
- if path.exists() and os.access(path, os.EX_OK):
- return path
- return None
+ search_path = os.environ.get("PATH", os.defpath)
+ # System binaries such as dhcp clients commonly live in sbin dirs that are
+ # not on a desktop session's PATH; append them if missing.
+ for sbin in ("/sbin", "/usr/sbin"):
+ if sbin not in search_path.split(os.pathsep):
+ search_path += os.pathsep + sbin
+
+ found = shutil.which(t, path=search_path)
+ return pathlib.Path(found) if found else None
def set_proc_title(name: str | None = None) -> int:
- """Set the process title"""
+ """Set the process title via ``prctl(PR_SET_NAME)``.
+
+ Only Linux exposes ``prctl`` through glibc; on other platforms this is a
+ no-op returning 0. Returns -1 if libc/prctl cannot be reached.
+ """
if not name:
name = pathlib.Path(sys.argv[0]).name
- libc = cdll.LoadLibrary('libc.so.6')
- buff = create_string_buffer(len(name) + 1)
- buff.value = name.encode("UTF-8")
- ret: int = libc.prctl(15, byref(buff), 0, 0, 0)
+ if not sys.platform.startswith("linux"):
+ logging.debug("set_proc_title is only supported on Linux")
+ return 0
+
+ try:
+ libc = cdll.LoadLibrary('libc.so.6')
+ buff = create_string_buffer(len(name) + 1)
+ buff.value = name.encode("UTF-8")
+ ret: int = libc.prctl(15, byref(buff), 0, 0, 0)
+ except (OSError, AttributeError):
+ logging.exception("Failed to set process title")
+ return -1
if ret != 0:
logging.error("Failed to set process title")
@@ -243,6 +276,12 @@ def create_logger(
date_fmt: str | None = None,
syslog: bool = False,
) -> logging.Logger:
+ """Configure and return the root logger for an entry point.
+
+ Used by every blueman binary (see ``apps/*.in``) to set the process-wide
+ log level, name, and format. With ``syslog`` enabled a SysLogHandler is
+ added when ``/dev/log`` is available, otherwise logging stays on stderr.
+ """
if log_format is None:
log_format = logger_format
if date_fmt is None:
@@ -253,10 +292,16 @@ def create_logger(
logger.name = name
if syslog:
- syslog_handler = logging.handlers.SysLogHandler(address="/dev/log")
- syslog_formatter = logging.Formatter(syslog_logger_format)
- syslog_handler.setFormatter(syslog_formatter)
- logger.addHandler(syslog_handler)
+ try:
+ syslog_handler = logging.handlers.SysLogHandler(address="/dev/log")
+ except OSError:
+ # /dev/log is absent on non-Linux platforms and minimal containers;
+ # the basicConfig stderr handler still provides logging.
+ logging.warning("Syslog socket /dev/log unavailable, logging to stderr only")
+ else:
+ syslog_formatter = logging.Formatter(syslog_logger_format)
+ syslog_handler.setFormatter(syslog_formatter)
+ logger.addHandler(syslog_handler)
return logger
@@ -266,11 +311,19 @@ def create_parser(
syslog: bool = True,
loglevel: bool = True,
) -> argparse.ArgumentParser:
+ """Build the shared argument parser used by every blueman entry point.
+
+ Adds the common ``--loglevel`` and ``--syslog`` options (each toggleable)
+ so all binaries in ``apps/*.in`` expose a consistent CLI surface.
+ """
if parser is None:
parser = argparse.ArgumentParser()
if loglevel:
- parser.add_argument("--loglevel", dest="LEVEL", default="warning")
+ parser.add_argument(
+ "--loglevel", dest="LEVEL", default="warning", type=str.lower,
+ choices=["debug", "info", "warning", "error", "critical"],
+ help="Logging verbosity (case-insensitive); defaults to warning.")
if syslog:
parser.add_argument("--syslog", dest="syslog", action="store_true")
@@ -327,7 +380,7 @@ def get_local_interfaces() -> dict[str, tuple[str, str | None]]:
mask = _netmask_for_ifacename(name, sock)
ip_dict[name] = (ipaddr, mask)
except OSError:
- logging.error('Socket creation failed', exc_info=True)
+ logging.exception('Socket creation failed')
return {}
return ip_dict
@@ -337,24 +390,25 @@ def bmexit(msg: str | int | None = None) -> None:
raise SystemExit(msg)
-def log_system_info() -> None:
- def parse_os_release(path: Path) -> dict[str, str]:
- release_dict = {}
- try:
- with path.open() as f:
- for line in f:
- line = line.strip()
- if line.startswith("#"):
- continue
- try:
- key, val = line.split("=")
- release_dict[key] = val.strip("\"")
- except ValueError:
- logging.error(f"Unable to parse line: {line}")
- except OSError:
- logging.error(f"Could not read {path.as_uri()}")
- return release_dict
+def parse_os_release(path: Path) -> dict[str, str]:
+ release_dict = {}
+ try:
+ with path.open() as f:
+ for line in f:
+ line = line.strip()
+ if not line or line.startswith("#"):
+ continue
+ key, sep, val = line.partition("=")
+ if not sep:
+ logging.error(f"Unable to parse line: {line}")
+ continue
+ release_dict[key] = val.strip("\"")
+ except OSError:
+ logging.error(f"Could not read {path.as_uri()}")
+ return release_dict
+
+def log_system_info() -> None:
try:
complete = subprocess.run(
[BLUETOOTHD_PATH, "-v"],
diff --git a/blueman/plugins/applet/TransferService.py b/blueman/plugins/applet/TransferService.py
index 29a47769b..bf797eb36 100644
--- a/blueman/plugins/applet/TransferService.py
+++ b/blueman/plugins/applet/TransferService.py
@@ -1,11 +1,12 @@
from datetime import datetime
from gettext import gettext as _, ngettext
from pathlib import Path
+import os
import shutil
import logging
from html import escape
-from typing import Optional, TypedDict, Union
-from collections.abc import Callable
+from typing import TypedDict, Union
+from collections.abc import Callable, Iterator
from blueman.bluemantyping import ObjectPath, BtAddress
from blueman.bluez.obex.AgentManager import AgentManager
@@ -14,7 +15,6 @@
from blueman.bluez.obex.Session import Session
from blueman.Functions import launch
from blueman.gui.Notification import Notification, _NotificationBubble, _NotificationDialog
-from blueman.main.Applet import BluemanApplet
from blueman.main.DbusService import DbusService, DbusError
from blueman.plugins.AppletPlugin import AppletPlugin
@@ -38,6 +38,37 @@ class PendingTransferDict(TypedDict):
NotificationType = Union[_NotificationBubble, _NotificationDialog]
+# Resolve (display name, trusted) for a device by adapter source path + address.
+# May raise; callers treat a raised resolver as an untrusted/unknown device.
+DeviceResolver = Callable[[str, BtAddress], tuple[str, bool]]
+
+_MAX_DESTINATION_ATTEMPTS = 10000
+
+
+def _destination_candidates(filename: str, stamp: str) -> "Iterator[str]":
+ yield filename
+ yield f"{stamp}_{filename}"
+ for index in range(1, _MAX_DESTINATION_ATTEMPTS):
+ yield f"{stamp}_{index}_{filename}"
+
+
+def reserve_destination(dest_dir: Path, filename: str, now: datetime) -> Path:
+ """Atomically reserve a unique destination, returning the reserved (empty) path.
+
+ Uses O_EXCL so two transfers completing in the same second cannot pick the
+ same name and overwrite each other; the caller moves the source onto it.
+ """
+ stamp = now.strftime('%Y%m%d%H%M%S')
+ for candidate in _destination_candidates(filename, stamp):
+ dest = dest_dir / candidate
+ try:
+ fd = os.open(dest, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+ except FileExistsError:
+ continue
+ os.close(fd)
+ return dest
+ raise FileExistsError(f"No free destination for {filename} in {dest_dir}")
+
class ObexErrorRejected(DbusError):
_name = "org.bluez.obex.Error.Rejected"
@@ -50,7 +81,7 @@ class ObexErrorCanceled(DbusError):
class Agent(DbusService):
__agent_path = ObjectPath('/org/bluez/obex/agent/blueman')
- def __init__(self, applet: BluemanApplet):
+ def __init__(self, resolve_device: "DeviceResolver"):
super().__init__(None, "org.bluez.obex.Agent1", self.__agent_path, Gio.BusType.SESSION)
self.add_method("Release", (), "", self._release)
@@ -58,12 +89,12 @@ def __init__(self, applet: BluemanApplet):
self.add_method("AuthorizePush", ("o",), "s", self._authorize_push, is_async=True)
self.register()
- self._applet = applet
+ self._resolve_device = resolve_device
self._config = Gio.Settings(schema_id="org.blueman.transfer")
- self._allowed_devices: list[str] = []
+ self._allowed_devices: set[BtAddress] = set()
self._notification: NotificationType | None = None
- self._pending_transfer: Optional[PendingTransferDict] = None
+ self._pending_transfers: dict[ObjectPath, PendingTransferDict] = {}
self.transfers: dict[ObjectPath, TransferDict] = {}
def register_at_manager(self) -> None:
@@ -77,30 +108,6 @@ def _release(self) -> None:
def _authorize_push(self, transfer_path: ObjectPath, ok: Callable[[str], None],
err: Callable[[ObexErrorRejected], None]) -> None:
- def on_action(action: str) -> None:
- logging.info(f"Action {action}")
-
- if action == "accept":
- assert self._pending_transfer
- self.transfers[self._pending_transfer['transfer_path']] = {
- 'path': self._pending_transfer['root'] / self._pending_transfer['filename'],
- 'size': self._pending_transfer['size'],
- 'name': self._pending_transfer['name']
- }
-
- ok(self.transfers[self._pending_transfer['transfer_path']]['path'].as_posix())
-
- self._allowed_devices.append(self._pending_transfer['address'])
-
- def _remove() -> bool:
- assert self._pending_transfer is not None # https://github.com/python/mypy/issues/2608
- self._allowed_devices.remove(self._pending_transfer['address'])
- return False
-
- GLib.timeout_add(60000, _remove)
- else:
- err(ObexErrorRejected("Rejected"))
-
transfer = Transfer(obj_path=transfer_path)
session = Session(obj_path=transfer.session)
root = Path(session.root)
@@ -109,18 +116,39 @@ def _remove() -> bool:
size = transfer.size
try:
- adapter = self._applet.Manager.get_adapter(session.source)
- device = self._applet.Manager.find_device(address, adapter.get_object_path())
- assert device is not None
- name = device.display_name
- trusted = device["Trusted"]
+ name, trusted = self._resolve_device(session.source, address)
except Exception as e:
logging.exception(e)
name = address
trusted = False
- self._pending_transfer = {'transfer_path': transfer_path, 'address': address, 'root': root,
- 'filename': filename, 'size': size, 'name': name}
+ pending: PendingTransferDict = {'transfer_path': transfer_path, 'address': address, 'root': root,
+ 'filename': filename, 'size': size, 'name': name}
+ self._pending_transfers[transfer_path] = pending
+
+ def on_action(action: str, pending: PendingTransferDict = pending) -> None:
+ logging.info(f"Action {action}")
+ self._pending_transfers.pop(pending['transfer_path'], None)
+
+ if action == "accept":
+ self.transfers[pending['transfer_path']] = {
+ 'path': pending['root'] / pending['filename'],
+ 'size': pending['size'],
+ 'name': pending['name']
+ }
+
+ ok(self.transfers[pending['transfer_path']]['path'].as_posix())
+
+ allowed_address = pending['address']
+ self._allowed_devices.add(allowed_address)
+
+ def _remove(address: BtAddress = allowed_address) -> bool:
+ self._allowed_devices.discard(address)
+ return False
+
+ GLib.timeout_add(60000, _remove)
+ else:
+ err(ObexErrorRejected("Rejected"))
# This device was neither allowed nor is it trusted -> ask for confirmation
if address not in self._allowed_devices and not (self._config['opp-accept'] and trusted):
@@ -174,6 +202,7 @@ def on_reset(_action: str) -> None:
logging.info('Reset share path')
self._config = Gio.Settings(schema_id="org.blueman.transfer")
+ self._handlerids = [] # per-instance; avoid sharing the class-level list across instances
share_path, invalid_share_path = self._make_share_path()
@@ -181,9 +210,10 @@ def on_reset(_action: str) -> None:
text = _('Configured directory for incoming files does not exist')
secondary_text = _('Please make sure that directory "%s" exists or '
'configure it with blueman-services. Until then the default "%s" will be used')
- self._notification = Notification(text, secondary_text % (self._config["shared-path"], share_path),
+ self._notification = Notification(text, secondary_text % (escape(self._config["shared-path"]),
+ escape(share_path.as_posix())),
icon_name='blueman', timeout=30000,
- actions=[('reset', 'Reset to default')], actions_cb=on_reset)
+ actions=[('reset', _('Reset to default'))], actions_cb=on_reset)
self._notification.show()
self._watch = Manager.watch_name_owner(self._on_dbus_name_appeared, self._on_dbus_name_vanished)
@@ -196,7 +226,8 @@ def on_unload(self) -> None:
def _make_share_path(self) -> tuple[Path, bool]:
config_path = Path(self._config["shared-path"])
- default_path = Path(GLib.get_user_special_dir(GLib.UserDirectory.DIRECTORY_DOWNLOAD))
+ download_dir = GLib.get_user_special_dir(GLib.UserDirectory.DIRECTORY_DOWNLOAD)
+ default_path = Path(download_dir) if download_dir else None
path = None
error = False
@@ -220,9 +251,15 @@ def _make_share_path(self) -> tuple[Path, bool]:
return path, error
+ def _resolve_device(self, source: str, address: BtAddress) -> tuple[str, bool]:
+ adapter = self.parent.Manager.get_adapter(source)
+ device = self.parent.Manager.find_device(address, adapter.get_object_path())
+ assert device is not None
+ return device.display_name, device["Trusted"]
+
def _register_agent(self) -> None:
if not self._agent:
- self._agent = Agent(self.parent)
+ self._agent = Agent(self._resolve_device)
self._agent.register_at_manager()
def _unregister_agent(self) -> None:
@@ -291,18 +328,17 @@ def _on_transfer_completed(self, _manager: Manager, transfer_path: ObjectPath, s
src = attributes['path']
dest_dir, ignored = self._make_share_path()
- filename = src.name
- if dest_dir.joinpath(filename).exists():
- now = datetime.now()
- filename = f"{now.strftime('%Y%m%d%H%M%S')}_{filename}"
+ dest = reserve_destination(dest_dir, src.name, datetime.now())
+ filename = dest.name
+ if filename != src.name:
logging.info(f"Destination file exists, renaming to: {filename}")
- dest = dest_dir.joinpath(filename)
try:
shutil.move(src, dest)
except (OSError, PermissionError):
logging.error("Failed to move files", exc_info=True)
+ dest.unlink(missing_ok=True)
success = False
if success:
@@ -313,7 +349,7 @@ def _on_transfer_completed(self, _manager: Manager, transfer_path: ObjectPath, s
icon_name="blueman")
self._add_open(self._notification, _("Open"), dest)
self._notification.show()
- elif not success:
+ else:
n = Notification(
_("Transfer failed"),
_("Transfer of file %(0)s failed") % {
@@ -352,3 +388,7 @@ def _on_session_removed(self, _manager: Manager, _session_path: str) -> None:
icon_name="blueman")
self._add_open(self._notification, _("Open Location"), share_path)
self._notification.show()
+
+ # The summary consumes the counts; reset so the next session starts fresh.
+ self._silent_transfers = 0
+ self._normal_transfers = 0
diff --git a/blueman/plugins/manager/Notes.py b/blueman/plugins/manager/Notes.py
index e37149c69..3deb3fc9d 100644
--- a/blueman/plugins/manager/Notes.py
+++ b/blueman/plugins/manager/Notes.py
@@ -32,7 +32,7 @@ def send_note_cb(dialog: Gtk.Dialog, response_id: int, device_address: str, text
tempfile = NamedTemporaryFile(suffix='.vnt', prefix='note', delete=False)
tempfile.write(data.encode('utf-8'))
tempfile.close()
- launch(f"blueman-sendto --delete --device={device_address}", paths=[tempfile.name])
+ launch("blueman-sendto", args=["--delete", f"--device={device_address}"], paths=[tempfile.name])
def send_note(device: Device, parent: Gtk.ApplicationWindow) -> None:
diff --git a/test/Makefile.am b/test/Makefile.am
index 0333b925e..7bea7ffc0 100644
--- a/test/Makefile.am
+++ b/test/Makefile.am
@@ -11,4 +11,5 @@ SUBDIRS = \
EXTRA_DIST = \
__init__.py \
test_imports.py \
- test_gobject.py
+ test_gobject.py \
+ test_functions.py
diff --git a/test/main/test_dns_server_provider.py b/test/main/test_dns_server_provider.py
index 1de3dd0c3..2224988a4 100644
--- a/test/main/test_dns_server_provider.py
+++ b/test/main/test_dns_server_provider.py
@@ -78,6 +78,14 @@ def _test_changed(action: Callable[[], None]) -> None:
action()
context = GLib.MainContext.default()
- while context.pending():
- context.iteration()
+ timed_out = False
+
+ def on_timeout() -> bool:
+ nonlocal timed_out
+ timed_out = True
+ return GLib.SOURCE_REMOVE
+
+ GLib.timeout_add_seconds(5, on_timeout)
+ while not mock.called and not timed_out:
+ context.iteration(may_block=True)
mock.assert_called_with(provider)
diff --git a/test/plugins/applet/Makefile.am b/test/plugins/applet/Makefile.am
index 3e92b4be6..0530f72df 100644
--- a/test/plugins/applet/Makefile.am
+++ b/test/plugins/applet/Makefile.am
@@ -1,3 +1,4 @@
EXTRA_DIST = \
__init__.py \
- test_imports.py
+ test_imports.py \
+ test_transfer_service.py
diff --git a/test/plugins/applet/test_transfer_service.py b/test/plugins/applet/test_transfer_service.py
new file mode 100644
index 000000000..1a8254258
--- /dev/null
+++ b/test/plugins/applet/test_transfer_service.py
@@ -0,0 +1,683 @@
+import tempfile
+from datetime import datetime
+from pathlib import Path
+from unittest import TestCase
+from unittest.mock import MagicMock, patch
+
+from blueman.plugins.applet.TransferService import Agent, TransferService, reserve_destination
+
+
+def _make_agent(resolve_device: object = None) -> Agent:
+ agent = Agent.__new__(Agent)
+ agent._allowed_devices = set()
+ agent._notification = None
+ agent._pending_transfers = {}
+ agent.transfers = {}
+ config = MagicMock()
+ config.__getitem__.side_effect = lambda key: True if key == "opp-accept" else ""
+ agent._config = config
+ agent._resolve_device = resolve_device or (lambda source, address: ("Phone", True))
+ return agent
+
+
+def _configure_transfer(transfer_mock: MagicMock, session_mock: MagicMock, *, address: str,
+ name: str = "file.bin", size: int = 10) -> None:
+ transfer = transfer_mock.return_value
+ transfer.session = "/sess"
+ transfer.name = name
+ transfer.size = size
+ session = session_mock.return_value
+ session.root = "/root"
+ session.address = address
+ session.source = "/org/bluez/hci0"
+
+
+def _make_plugin(configured_share_path: str) -> TransferService:
+ plugin = TransferService.__new__(TransferService)
+ config = MagicMock()
+ config.__getitem__.side_effect = lambda key: {"shared-path": configured_share_path}[key]
+ plugin._config = config
+ return plugin
+
+
+@patch("blueman.plugins.applet.TransferService.Manager")
+@patch("blueman.plugins.applet.TransferService.Notification")
+@patch("blueman.plugins.applet.TransferService.Gio.Settings")
+class TestSharePathEscaping(TestCase):
+ def _body(self, settings_mock: MagicMock, notification_mock: MagicMock, configured: str) -> str:
+ plugin = _make_plugin(configured)
+ settings_mock.return_value = plugin._config
+ with patch.object(TransferService, "_make_share_path", return_value=(Path("/srv/Downloads"), True)):
+ plugin.on_load()
+ notification_mock.assert_called_once()
+ return notification_mock.call_args.args[1]
+
+ def test_escapes_angle_brackets(self, settings_mock: MagicMock, notification_mock: MagicMock,
+ _manager_mock: MagicMock) -> None:
+ body = self._body(settings_mock, notification_mock, "/home/evil")
+ self.assertNotIn("evil", body)
+ self.assertIn("<b>evil</b>", body)
+
+ def test_escapes_ampersand(self, settings_mock: MagicMock, notification_mock: MagicMock,
+ _manager_mock: MagicMock) -> None:
+ body = self._body(settings_mock, notification_mock, "/home/a & b")
+ self.assertIn("&", body)
+
+ def test_escapes_quotes(self, settings_mock: MagicMock, notification_mock: MagicMock,
+ _manager_mock: MagicMock) -> None:
+ body = self._body(settings_mock, notification_mock, '/home/"quoted"')
+ self.assertNotIn('"quoted"', body)
+ self.assertIn(""quoted"", body)
+
+
+@patch("blueman.plugins.applet.TransferService.Manager")
+@patch("blueman.plugins.applet.TransferService.Notification")
+@patch("blueman.plugins.applet.TransferService.Gio.Settings")
+class TestResetActionTranslated(TestCase):
+ def test_reset_label_routed_through_gettext(self, settings_mock: MagicMock, notification_mock: MagicMock,
+ _manager_mock: MagicMock) -> None:
+ plugin = _make_plugin("/does/not/matter")
+ settings_mock.return_value = plugin._config
+ with patch.object(TransferService, "_make_share_path", return_value=(Path("/srv/Downloads"), True)), \
+ patch("blueman.plugins.applet.TransferService._", lambda s: f"{s}"):
+ plugin.on_load()
+ actions = notification_mock.call_args.kwargs["actions"]
+ self.assertEqual(actions, [("reset", "
Reset to default")])
+
+
+@patch("blueman.plugins.applet.TransferService.GLib")
+@patch("blueman.plugins.applet.TransferService.Notification")
+@patch("blueman.plugins.applet.TransferService.Session")
+@patch("blueman.plugins.applet.TransferService.Transfer")
+class TestAllowedDeviceRemoval(TestCase):
+ def _authorize(self, transfer_mock: MagicMock, session_mock: MagicMock, glib_mock: MagicMock,
+ address: str) -> tuple[Agent, object]:
+ agent = _make_agent()
+ _configure_transfer(transfer_mock, session_mock, address=address)
+ agent._authorize_push("/transfer", MagicMock(), MagicMock())
+ remove_cb = glib_mock.timeout_add.call_args.args[1]
+ return agent, remove_cb
+
+ def test_address_allowed_then_removed(self, transfer_mock: MagicMock, session_mock: MagicMock,
+ _notification_mock: MagicMock, glib_mock: MagicMock) -> None:
+ agent, remove_cb = self._authorize(transfer_mock, session_mock, glib_mock, "AA:BB:CC:DD:EE:FF")
+ self.assertIn("AA:BB:CC:DD:EE:FF", agent._allowed_devices)
+ remove_cb()
+ self.assertNotIn("AA:BB:CC:DD:EE:FF", agent._allowed_devices)
+
+ def test_removal_is_idempotent(self, transfer_mock: MagicMock, session_mock: MagicMock,
+ _notification_mock: MagicMock, glib_mock: MagicMock) -> None:
+ agent, remove_cb = self._authorize(transfer_mock, session_mock, glib_mock, "AA:BB:CC:DD:EE:FF")
+ remove_cb()
+ remove_cb() # must not raise even though the address is already gone
+
+ def test_removes_captured_address_not_current_pending(self, transfer_mock: MagicMock, session_mock: MagicMock,
+ _notification_mock: MagicMock, glib_mock: MagicMock) -> None:
+ agent, remove_cb = self._authorize(transfer_mock, session_mock, glib_mock, "AA:BB:CC:DD:EE:FF")
+ agent._allowed_devices.add("11:22:33:44:55:66") # a later, overlapping request's address
+ remove_cb()
+ self.assertNotIn("AA:BB:CC:DD:EE:FF", agent._allowed_devices)
+ self.assertIn("11:22:33:44:55:66", agent._allowed_devices)
+
+
+@patch("blueman.plugins.applet.TransferService.GLib")
+@patch("blueman.plugins.applet.TransferService.Notification")
+@patch("blueman.plugins.applet.TransferService.Session")
+@patch("blueman.plugins.applet.TransferService.Transfer")
+class TestOverlappingPending(TestCase):
+ def _setup_two_untrusted(self, agent: Agent, transfer_mock: MagicMock, session_mock: MagicMock) -> None:
+ agent._resolve_device = lambda source, address: ("Phone", False)
+
+ def make_transfer(obj_path: str) -> MagicMock:
+ transfer = MagicMock()
+ transfer.session = "/s" + obj_path
+ transfer.name = "first.bin" if obj_path == "/t1" else "second.bin"
+ transfer.size = 10
+ return transfer
+
+ def make_session(obj_path: str) -> MagicMock:
+ session = MagicMock()
+ session.root = "/root"
+ session.address = "AA:AA" if obj_path == "/s/t1" else "BB:BB"
+ session.source = "/org/bluez/hci0"
+ return session
+
+ transfer_mock.side_effect = make_transfer
+ session_mock.side_effect = make_session
+
+ def test_overlapping_requests_keep_independent_records(self, transfer_mock: MagicMock, session_mock: MagicMock,
+ notification_mock: MagicMock, _glib_mock: MagicMock) -> None:
+ agent = _make_agent()
+ self._setup_two_untrusted(agent, transfer_mock, session_mock)
+ ok_a, err_a, ok_b, err_b = MagicMock(), MagicMock(), MagicMock(), MagicMock()
+
+ agent._authorize_push("/t1", ok_a, err_a)
+ agent._authorize_push("/t2", ok_b, err_b)
+ self.assertEqual(set(agent._pending_transfers), {"/t1", "/t2"})
+
+ on_action_a = notification_mock.call_args_list[0].kwargs["actions_cb"]
+ on_action_b = notification_mock.call_args_list[1].kwargs["actions_cb"]
+
+ on_action_a("accept")
+ self.assertEqual(agent.transfers["/t1"]["path"].name, "first.bin")
+ ok_a.assert_called_once()
+ self.assertNotIn("/t1", agent._pending_transfers)
+ self.assertIn("/t2", agent._pending_transfers) # the second request is untouched by the first's action
+
+ on_action_b("reject")
+ err_b.assert_called_once()
+ self.assertNotIn("/t2", agent._pending_transfers)
+ self.assertNotIn("/t2", agent.transfers)
+
+
+class TestReserveDestination(TestCase):
+ _NOW = datetime(2020, 1, 2, 3, 4, 5)
+ _STAMP = "20200102030405"
+
+ def setUp(self) -> None:
+ self._tmp = tempfile.TemporaryDirectory()
+ self.dir = Path(self._tmp.name)
+ self.addCleanup(self._tmp.cleanup)
+
+ def test_no_collision_uses_plain_name(self) -> None:
+ dest = reserve_destination(self.dir, "file.bin", self._NOW)
+ self.assertEqual(dest, self.dir / "file.bin")
+ self.assertTrue(dest.exists())
+
+ def test_first_collision_uses_timestamp(self) -> None:
+ (self.dir / "file.bin").write_text("existing")
+ dest = reserve_destination(self.dir, "file.bin", self._NOW)
+ self.assertEqual(dest, self.dir / f"{self._STAMP}_file.bin")
+
+ def test_second_collision_uses_indexed_timestamp(self) -> None:
+ (self.dir / "file.bin").write_text("existing")
+ (self.dir / f"{self._STAMP}_file.bin").write_text("existing")
+ dest = reserve_destination(self.dir, "file.bin", self._NOW)
+ self.assertEqual(dest, self.dir / f"{self._STAMP}_1_file.bin")
+
+ def test_same_second_calls_never_collide(self) -> None:
+ # Two transfers of the same name completing in the same second must each
+ # reserve a distinct, freshly-created destination — no overwrite.
+ reserved = [reserve_destination(self.dir, "photo.jpg", self._NOW) for _ in range(5)]
+ self.assertEqual(len(set(reserved)), 5)
+ for dest in reserved:
+ self.assertTrue(dest.exists())
+
+ def test_reserved_file_is_exclusive(self) -> None:
+ dest = reserve_destination(self.dir, "x", self._NOW)
+ # A second reservation must not hand back the same path it just created.
+ other = reserve_destination(self.dir, "x", self._NOW)
+ self.assertNotEqual(dest, other)
+
+ def test_fuzz_weird_names_stay_unique_and_safe(self) -> None:
+ names = ["a b.bin", "résumé.pdf", ".hidden", "name.with.dots.tar.gz",
+ "UPPER.TXT", " spaces ", "emoji-😀.png", "a" * 200 + ".bin"]
+ for raw in names:
+ with self.subTest(name=raw):
+ first = reserve_destination(self.dir, raw, self._NOW)
+ second = reserve_destination(self.dir, raw, self._NOW)
+ self.assertNotEqual(first, second)
+ self.assertTrue(first.exists() and second.exists())
+ # Reserved name must stay within the destination directory.
+ self.assertEqual(first.parent, self.dir)
+ self.assertEqual(second.parent, self.dir)
+
+
+class TestDeviceResolverInjection(TestCase):
+ def test_plugin_resolver_delegates_to_manager(self) -> None:
+ plugin = TransferService.__new__(TransferService)
+ plugin.parent = MagicMock()
+ device = plugin.parent.Manager.find_device.return_value
+ device.display_name = "Watch"
+ device.__getitem__.side_effect = lambda key: True if key == "Trusted" else None
+
+ name, trusted = plugin._resolve_device("/org/bluez/hci0", "AA:BB:CC:DD:EE:FF")
+
+ self.assertEqual((name, trusted), ("Watch", True))
+ plugin.parent.Manager.get_adapter.assert_called_once_with("/org/bluez/hci0")
+ plugin.parent.Manager.find_device.assert_called_once()
+
+ @patch("blueman.plugins.applet.TransferService.GLib")
+ @patch("blueman.plugins.applet.TransferService.Notification")
+ @patch("blueman.plugins.applet.TransferService.Session")
+ @patch("blueman.plugins.applet.TransferService.Transfer")
+ def test_agent_uses_injected_resolver(self, transfer_mock: MagicMock, session_mock: MagicMock,
+ _notification_mock: MagicMock, _glib_mock: MagicMock) -> None:
+ resolver = MagicMock(return_value=("Laptop", True))
+ agent = _make_agent(resolver)
+ _configure_transfer(transfer_mock, session_mock, address="AA:BB:CC:DD:EE:FF")
+
+ agent._authorize_push("/transfer", MagicMock(), MagicMock())
+
+ resolver.assert_called_once_with("/org/bluez/hci0", "AA:BB:CC:DD:EE:FF")
+ self.assertFalse(hasattr(agent, "_applet"))
+
+ @patch("blueman.plugins.applet.TransferService.GLib")
+ @patch("blueman.plugins.applet.TransferService.Notification")
+ @patch("blueman.plugins.applet.TransferService.Session")
+ @patch("blueman.plugins.applet.TransferService.Transfer")
+ def test_resolver_failure_falls_back_to_address(self, transfer_mock: MagicMock, session_mock: MagicMock,
+ notification_mock: MagicMock, _glib_mock: MagicMock) -> None:
+ resolver = MagicMock(side_effect=RuntimeError("no device"))
+ agent = _make_agent(resolver)
+ agent._config.__getitem__.side_effect = lambda key: False # force the confirmation path
+ _configure_transfer(transfer_mock, session_mock, address="AA:BB:CC:DD:EE:FF", size=10)
+
+ agent._authorize_push("/transfer", MagicMock(), MagicMock())
+
+ body = notification_mock.call_args.args[1]
+ self.assertIn("AA:BB:CC:DD:EE:FF", body) # falls back to the raw address as the display name
+
+
+@patch("blueman.plugins.applet.TransferService.Notification")
+class TestTransferCompleted(TestCase):
+ def setUp(self) -> None:
+ self._tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(self._tmp.cleanup)
+ root = Path(self._tmp.name)
+ self.dest_dir = root / "Downloads"
+ self.dest_dir.mkdir()
+ self.src = root / "incoming.bin"
+ self.src.write_text("payload")
+
+ def _plugin(self, size: int = 10) -> TransferService:
+ plugin = TransferService.__new__(TransferService)
+ plugin._agent = MagicMock()
+ plugin._agent.transfers = {"/t": {"path": self.src, "size": size, "name": "Phone"}}
+ plugin._normal_transfers = 0
+ plugin._silent_transfers = 1
+ plugin._notification = None
+ return plugin
+
+ def test_unauthorized_transfer_ignored(self, notification_mock: MagicMock) -> None:
+ plugin = self._plugin()
+ plugin._agent.transfers = {}
+ plugin._on_transfer_completed(MagicMock(), "/t", True)
+ notification_mock.assert_not_called()
+
+ def test_success_moves_file_and_notifies(self, notification_mock: MagicMock) -> None:
+ plugin = self._plugin()
+ with patch.object(TransferService, "_make_share_path", return_value=(self.dest_dir, False)):
+ plugin._on_transfer_completed(MagicMock(), "/t", True)
+ self.assertTrue((self.dest_dir / "incoming.bin").exists())
+ self.assertFalse(self.src.exists())
+ notification_mock.assert_called_once()
+ self.assertNotIn("/t", plugin._agent.transfers)
+
+ def test_failed_move_cleans_placeholder_and_decrements(self, notification_mock: MagicMock) -> None:
+ plugin = self._plugin(size=10)
+ with patch.object(TransferService, "_make_share_path", return_value=(self.dest_dir, False)), \
+ patch("blueman.plugins.applet.TransferService.shutil.move", side_effect=OSError("boom")):
+ plugin._on_transfer_completed(MagicMock(), "/t", True)
+ self.assertEqual(list(self.dest_dir.iterdir()), []) # reserved placeholder cleaned up
+ self.assertEqual(plugin._silent_transfers, 0)
+ self.assertNotIn("/t", plugin._agent.transfers)
+
+ def test_failed_move_decrements_normal_for_large_file(self, notification_mock: MagicMock) -> None:
+ plugin = self._plugin(size=400000)
+ plugin._normal_transfers = 1
+ plugin._silent_transfers = 0
+ with patch.object(TransferService, "_make_share_path", return_value=(self.dest_dir, False)), \
+ patch("blueman.plugins.applet.TransferService.shutil.move", side_effect=PermissionError):
+ plugin._on_transfer_completed(MagicMock(), "/t", False)
+ self.assertEqual(plugin._normal_transfers, 0)
+
+
+class TestTransferStarted(TestCase):
+ def _plugin(self, size: int) -> TransferService:
+ plugin = TransferService.__new__(TransferService)
+ plugin._agent = MagicMock()
+ plugin._agent.transfers = {"/t": {"path": Path("/x"), "size": size, "name": "Phone"}}
+ plugin._normal_transfers = 0
+ plugin._silent_transfers = 0
+ return plugin
+
+ def test_large_file_counts_as_normal(self) -> None:
+ plugin = self._plugin(400000)
+ plugin._on_transfer_started(MagicMock(), "/t")
+ self.assertEqual((plugin._normal_transfers, plugin._silent_transfers), (1, 0))
+
+ def test_small_file_counts_as_silent(self) -> None:
+ plugin = self._plugin(10)
+ plugin._on_transfer_started(MagicMock(), "/t")
+ self.assertEqual((plugin._normal_transfers, plugin._silent_transfers), (0, 1))
+
+ def test_unauthorized_transfer_ignored(self) -> None:
+ plugin = self._plugin(10)
+ plugin._agent = None
+ plugin._on_transfer_started(MagicMock(), "/t") # must not raise
+
+
+@patch("blueman.plugins.applet.TransferService.Notification")
+class TestSessionRemoved(TestCase):
+ def _plugin(self, silent: int, normal: int) -> TransferService:
+ plugin = TransferService.__new__(TransferService)
+ plugin._silent_transfers = silent
+ plugin._normal_transfers = normal
+ plugin._notification = None
+ return plugin
+
+ def test_no_silent_transfers_does_nothing(self, notification_mock: MagicMock) -> None:
+ plugin = self._plugin(silent=0, normal=0)
+ plugin._on_session_removed(MagicMock(), "/sess")
+ notification_mock.assert_not_called()
+
+ def test_only_silent_transfers_notifies(self, notification_mock: MagicMock) -> None:
+ plugin = self._plugin(silent=2, normal=0)
+ with patch.object(TransferService, "_make_share_path", return_value=(Path("/dl"), False)):
+ plugin._on_session_removed(MagicMock(), "/sess")
+ notification_mock.assert_called_once()
+
+ def test_mixed_transfers_notifies_more_variant(self, notification_mock: MagicMock) -> None:
+ plugin = self._plugin(silent=1, normal=1)
+ with patch.object(TransferService, "_make_share_path", return_value=(Path("/dl"), False)):
+ plugin._on_session_removed(MagicMock(), "/sess")
+ notification_mock.assert_called_once()
+
+
+class TestAgentControl(TestCase):
+ def test_cancel_closes_notification_and_raises(self) -> None:
+ from blueman.plugins.applet.TransferService import ObexErrorCanceled
+ agent = _make_agent()
+ agent._notification = MagicMock()
+ with self.assertRaises(ObexErrorCanceled):
+ agent._cancel()
+ agent._notification.close.assert_called_once()
+
+ def test_release_raises(self) -> None:
+ agent = _make_agent()
+ with self.assertRaises(Exception):
+ agent._release()
+
+
+@patch("blueman.plugins.applet.TransferService.GLib")
+class TestMakeSharePath(TestCase):
+ def _plugin(self, configured: str) -> TransferService:
+ plugin = TransferService.__new__(TransferService)
+ config = MagicMock()
+ config.__getitem__.side_effect = lambda key: configured
+ config.__setitem__ = MagicMock()
+ plugin._config = config
+ return plugin
+
+ def test_empty_config_uses_download_dir(self, glib_mock: MagicMock) -> None:
+ glib_mock.get_user_special_dir.return_value = "/dl"
+ plugin = self._plugin("")
+ path, error = plugin._make_share_path()
+ self.assertEqual(path, Path("/dl"))
+ self.assertFalse(error)
+
+ def test_invalid_dir_flags_error(self, glib_mock: MagicMock) -> None:
+ glib_mock.get_user_special_dir.return_value = "/dl"
+ plugin = self._plugin("/does/not/exist/here")
+ path, error = plugin._make_share_path()
+ self.assertEqual(path, Path("/dl"))
+ self.assertTrue(error)
+
+ def test_valid_dir_used(self, glib_mock: MagicMock) -> None:
+ glib_mock.get_user_special_dir.return_value = "/dl"
+ with tempfile.TemporaryDirectory() as tmp:
+ plugin = self._plugin(tmp)
+ path, error = plugin._make_share_path()
+ self.assertEqual(path, Path(tmp))
+ self.assertFalse(error)
+
+
+@patch("blueman.plugins.applet.TransferService.GLib")
+@patch("blueman.plugins.applet.TransferService.Notification")
+@patch("blueman.plugins.applet.TransferService.Session")
+@patch("blueman.plugins.applet.TransferService.Transfer")
+class TestAutoAccept(TestCase):
+ def test_trusted_large_file_auto_accepts_with_notification(self, transfer_mock: MagicMock, session_mock: MagicMock,
+ notification_mock: MagicMock,
+ _glib_mock: MagicMock) -> None:
+ agent = _make_agent() # default resolver -> trusted
+ _configure_transfer(transfer_mock, session_mock, address="AA:BB:CC:DD:EE:FF", size=400001)
+ ok = MagicMock()
+ agent._authorize_push("/t", ok, MagicMock())
+ ok.assert_called_once()
+ self.assertIn("/t", agent.transfers)
+ self.assertIn("AA:BB:CC:DD:EE:FF", agent._allowed_devices)
+ notification_mock.assert_called_once()
+
+
+class TestRegisterAgent(TestCase):
+ @patch("blueman.plugins.applet.TransferService.Agent")
+ def test_register_then_unregister(self, agent_cls: MagicMock) -> None:
+ plugin = TransferService.__new__(TransferService)
+ plugin._agent = None
+ plugin._register_agent()
+ agent_cls.assert_called_once_with(plugin._resolve_device)
+ agent_cls.return_value.register_at_manager.assert_called_once()
+ agent = plugin._agent
+ plugin._unregister_agent()
+ agent.unregister_from_manager.assert_called_once()
+ agent.unregister.assert_called_once()
+ self.assertIsNone(plugin._agent)
+
+
+@patch("blueman.plugins.applet.TransferService.launch")
+@patch("blueman.plugins.applet.TransferService.Notification")
+class TestOpenAction(TestCase):
+ def test_open_action_launches_xdg_open(self, notification_mock: MagicMock, launch_mock: MagicMock) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ dest_dir = Path(tmp)
+ src = dest_dir / "incoming.bin"
+ src.write_text("x")
+ plugin = TransferService.__new__(TransferService)
+ plugin._agent = MagicMock()
+ plugin._agent.transfers = {"/t": {"path": src, "size": 10, "name": "Phone"}}
+ plugin._normal_transfers = 0
+ plugin._silent_transfers = 1
+ plugin._notification = None
+ notification_mock.return_value.actions_supported = True
+ dst = dest_dir / "out"
+ dst.mkdir()
+ with patch.object(TransferService, "_make_share_path", return_value=(dst, False)):
+ plugin._on_transfer_completed(MagicMock(), "/t", True)
+ on_open = notification_mock.return_value.add_action.call_args.args[2]
+ on_open("open")
+ launch_mock.assert_called_once()
+
+
+@patch("blueman.plugins.applet.TransferService.AgentManager")
+@patch("blueman.plugins.applet.TransferService.Gio")
+@patch("blueman.plugins.applet.TransferService.DbusService.__init__", return_value=None)
+class TestAgentLifecycle(TestCase):
+ def _make(self) -> tuple:
+ with patch.object(Agent, "add_method") as add, patch.object(Agent, "register") as reg:
+ agent = Agent(lambda source, address: ("X", True))
+ return agent, add, reg
+
+ def test_init_registers_three_methods_and_state(self, _dbus: MagicMock, _gio: MagicMock,
+ _am: MagicMock) -> None:
+ agent, add, reg = self._make()
+ self.assertEqual(add.call_count, 3)
+ reg.assert_called_once()
+ self.assertEqual(agent._allowed_devices, set())
+ self.assertEqual(agent._pending_transfers, {})
+ self.assertEqual(agent.transfers, {})
+
+ def test_register_and_unregister_at_manager(self, _dbus: MagicMock, _gio: MagicMock,
+ am_mock: MagicMock) -> None:
+ agent, _add, _reg = self._make()
+ agent.register_at_manager()
+ am_mock.return_value.register_agent.assert_called_once()
+ agent.unregister_from_manager()
+ am_mock.return_value.unregister_agent.assert_called_once()
+
+
+@patch("blueman.plugins.applet.TransferService.Manager")
+@patch("blueman.plugins.applet.TransferService.Notification")
+@patch("blueman.plugins.applet.TransferService.Gio.Settings")
+class TestOnLoadClosures(TestCase):
+ def _load(self, settings_mock: MagicMock, notification_mock: MagicMock, invalid: bool) -> TransferService:
+ plugin = _make_plugin("/configured")
+ settings_mock.return_value = plugin._config
+ with patch.object(TransferService, "_make_share_path", return_value=(Path("/srv/Downloads"), invalid)):
+ plugin.on_load()
+ return plugin
+
+ def test_handlerids_are_per_instance(self, settings_mock: MagicMock, notification_mock: MagicMock,
+ _manager_mock: MagicMock) -> None:
+ plugin = self._load(settings_mock, notification_mock, invalid=False)
+ self.assertEqual(plugin._handlerids, [])
+ self.assertIsNot(plugin._handlerids, TransferService._handlerids)
+
+ def test_on_reset_clears_config_and_notification(self, settings_mock: MagicMock, notification_mock: MagicMock,
+ _manager_mock: MagicMock) -> None:
+ plugin = self._load(settings_mock, notification_mock, invalid=True)
+ on_reset = notification_mock.call_args.kwargs["actions_cb"]
+ on_reset("reset")
+ plugin._config.reset.assert_called_once_with("shared-path")
+ self.assertIsNone(plugin._notification)
+
+
+@patch("blueman.plugins.applet.TransferService.Gio")
+class TestOnUnload(TestCase):
+ def test_unwatch_and_unregister(self, gio_mock: MagicMock) -> None:
+ plugin = TransferService.__new__(TransferService)
+ plugin._watch = 42
+ agent = MagicMock()
+ plugin._agent = agent
+ plugin.on_unload()
+ gio_mock.bus_unwatch_name.assert_called_once_with(42)
+ agent.unregister_from_manager.assert_called_once()
+ self.assertIsNone(plugin._agent)
+
+ def test_noop_without_watch_or_agent(self, gio_mock: MagicMock) -> None:
+ plugin = TransferService.__new__(TransferService)
+ plugin._watch = None
+ plugin._agent = None
+ plugin.on_unload()
+ gio_mock.bus_unwatch_name.assert_not_called()
+
+
+@patch("blueman.plugins.applet.TransferService.Agent")
+@patch("blueman.plugins.applet.TransferService.Manager")
+class TestDbusNameLifecycle(TestCase):
+ def _plugin(self) -> TransferService:
+ plugin = TransferService.__new__(TransferService)
+ plugin._agent = None
+ plugin._manager = None
+ plugin._handlerids = []
+ return plugin
+
+ def test_appeared_connects_signals_and_registers(self, manager_mock: MagicMock, agent_mock: MagicMock) -> None:
+ plugin = self._plugin()
+ plugin._on_dbus_name_appeared(MagicMock(), "name", "owner")
+ self.assertEqual(len(plugin._handlerids), 3)
+ agent_mock.assert_called_once_with(plugin._resolve_device)
+
+ def test_appeared_handles_manager_failure(self, manager_mock: MagicMock, agent_mock: MagicMock) -> None:
+ from gi.repository import GLib
+ manager_mock.side_effect = GLib.Error("obex down")
+ plugin = self._plugin()
+ plugin._on_dbus_name_appeared(MagicMock(), "name", "owner")
+ self.assertEqual(plugin._handlerids, [])
+ agent_mock.assert_not_called()
+
+ def test_vanished_disconnects_and_clears(self, manager_mock: MagicMock, _agent_mock: MagicMock) -> None:
+ plugin = self._plugin()
+ manager = MagicMock()
+ plugin._manager = manager
+ plugin._handlerids = [1, 2, 3]
+ agent = MagicMock()
+ plugin._agent = agent
+ plugin._on_dbus_name_vanished(MagicMock(), "name")
+ self.assertEqual(manager.disconnect.call_count, 3)
+ self.assertIsNone(plugin._manager)
+ self.assertEqual(plugin._handlerids, [])
+ agent.unregister.assert_called_once()
+ self.assertIsNone(plugin._agent)
+
+ def test_vanished_noop_when_idle(self, manager_mock: MagicMock, _agent_mock: MagicMock) -> None:
+ plugin = self._plugin()
+ plugin._on_dbus_name_vanished(MagicMock(), "name") # must not raise
+
+
+class TestReserveDestinationExhaustion(TestCase):
+ def test_raises_when_no_free_candidate(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ d = Path(tmp)
+ now = datetime(2020, 1, 2, 3, 4, 5)
+ stamp = "20200102030405"
+ with patch("blueman.plugins.applet.TransferService._MAX_DESTINATION_ATTEMPTS", 2):
+ (d / "f").write_text("")
+ (d / f"{stamp}_f").write_text("")
+ (d / f"{stamp}_1_f").write_text("")
+ with self.assertRaises(FileExistsError):
+ reserve_destination(d, "f", now)
+
+
+@patch("blueman.plugins.applet.TransferService.GLib")
+class TestMakeSharePathReset(TestCase):
+ def test_config_equal_to_default_is_reset(self, glib_mock: MagicMock) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ glib_mock.get_user_special_dir.return_value = tmp
+ plugin = TransferService.__new__(TransferService)
+ config = MagicMock()
+ config.__getitem__.side_effect = lambda key: tmp
+ config.__setitem__ = MagicMock()
+ plugin._config = config
+ path, error = plugin._make_share_path()
+ self.assertEqual(path, Path(tmp))
+ self.assertFalse(error)
+ config.__setitem__.assert_called_with("shared-path", "")
+
+
+@patch("blueman.plugins.applet.TransferService.Notification")
+class TestCompletionRenameAndReset(TestCase):
+ def setUp(self) -> None:
+ self._tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(self._tmp.cleanup)
+ root = Path(self._tmp.name)
+ self.dest_dir = root / "Downloads"
+ self.dest_dir.mkdir()
+ self.src = root / "incoming.bin"
+ self.src.write_text("payload")
+
+ def test_collision_renames_with_timestamp(self, _notification_mock: MagicMock) -> None:
+ (self.dest_dir / "incoming.bin").write_text("pre-existing")
+ plugin = TransferService.__new__(TransferService)
+ plugin._agent = MagicMock()
+ plugin._agent.transfers = {"/t": {"path": self.src, "size": 10, "name": "Phone"}}
+ plugin._normal_transfers = 0
+ plugin._silent_transfers = 1
+ plugin._notification = None
+ with patch.object(TransferService, "_make_share_path", return_value=(self.dest_dir, False)):
+ plugin._on_transfer_completed(MagicMock(), "/t", True)
+ moved = [p.name for p in self.dest_dir.iterdir()]
+ self.assertIn("incoming.bin", moved) # the pre-existing file
+ self.assertTrue(any(n.endswith("_incoming.bin") for n in moved)) # the renamed arrival
+
+
+@patch("blueman.plugins.applet.TransferService.Notification")
+class TestSessionCounterReset(TestCase):
+ def _plugin(self, silent: int, normal: int) -> TransferService:
+ plugin = TransferService.__new__(TransferService)
+ plugin._silent_transfers = silent
+ plugin._normal_transfers = normal
+ plugin._notification = None
+ return plugin
+
+ def test_counters_reset_after_summary(self, _notification_mock: MagicMock) -> None:
+ plugin = self._plugin(silent=2, normal=1)
+ with patch.object(TransferService, "_make_share_path", return_value=(Path("/dl"), False)):
+ plugin._on_session_removed(MagicMock(), "/sess")
+ self.assertEqual(plugin._silent_transfers, 0)
+ self.assertEqual(plugin._normal_transfers, 0)
+
+
+@patch("blueman.plugins.applet.TransferService.GLib")
+class TestMakeSharePathXdgMissing(TestCase):
+ def test_falls_back_to_home_when_xdg_unavailable(self, glib_mock: MagicMock) -> None:
+ glib_mock.get_user_special_dir.return_value = None
+ plugin = TransferService.__new__(TransferService)
+ config = MagicMock()
+ config.__getitem__.side_effect = lambda key: "" # no configured path
+ config.__setitem__ = MagicMock()
+ plugin._config = config
+ path, error = plugin._make_share_path()
+ self.assertEqual(path, Path("~").expanduser())
+ self.assertFalse(error)
diff --git a/test/test_functions.py b/test/test_functions.py
new file mode 100644
index 000000000..a5e47c8c5
--- /dev/null
+++ b/test/test_functions.py
@@ -0,0 +1,311 @@
+import logging
+import os
+import tempfile
+from pathlib import Path
+from unittest import TestCase
+from unittest.mock import MagicMock, patch
+
+from gi.repository import GLib
+
+from blueman.Constants import BIN_DIR
+from blueman.Functions import (
+ adapter_path_to_name,
+ check_bluetooth_status,
+ create_logger,
+ create_parser,
+ format_bytes,
+ have,
+ launch,
+ parse_os_release,
+ set_proc_title,
+)
+from blueman.main.DBusProxies import DBusProxyFailed
+
+
+class TestFormatBytes(TestCase):
+ def test_zero(self) -> None:
+ self.assertEqual(format_bytes(0), (0.0, "B"))
+
+ def test_sub_kilobyte(self) -> None:
+ self.assertEqual(format_bytes(512), (512.0, "B"))
+ self.assertEqual(format_bytes(1023), (1023.0, "B"))
+
+ def test_kilobyte_boundary(self) -> None:
+ # Regression: exact 1024 must be 1.0 KB, not a fraction of a GB.
+ self.assertEqual(format_bytes(1024), (1.0, "KB"))
+
+ def test_megabyte_boundary(self) -> None:
+ self.assertEqual(format_bytes(1024 * 1024), (1.0, "MB"))
+
+ def test_gigabyte_boundary(self) -> None:
+ self.assertEqual(format_bytes(1024 * 1024 * 1024), (1.0, "GB"))
+
+ def test_mid_band_values(self) -> None:
+ self.assertEqual(format_bytes(1536), (1.5, "KB"))
+ self.assertEqual(format_bytes(1024 * 1024 * 3), (3.0, "MB"))
+
+ def test_huge_value(self) -> None:
+ ret, suffix = format_bytes(5 * 1024 ** 4)
+ self.assertEqual(suffix, "GB")
+ self.assertEqual(ret, 5 * 1024)
+
+ def test_accepts_float_and_int(self) -> None:
+ self.assertEqual(format_bytes(2048.0), (2.0, "KB"))
+ self.assertEqual(format_bytes(2048), (2.0, "KB"))
+
+
+class TestAdapterPathToName(TestCase):
+ def test_normal_path(self) -> None:
+ self.assertEqual(adapter_path_to_name("/org/bluez/hci0"), "hci0")
+
+ def test_none_and_empty(self) -> None:
+ self.assertIsNone(adapter_path_to_name(None))
+ self.assertIsNone(adapter_path_to_name(""))
+
+ def test_no_hci(self) -> None:
+ self.assertIsNone(adapter_path_to_name("/org/bluez"))
+ self.assertIsNone(adapter_path_to_name("no-match"))
+
+ def test_case_sensitive(self) -> None:
+ # The pattern matches lowercase "hci" only.
+ self.assertIsNone(adapter_path_to_name("HCI0"))
+
+ def test_trailing_segments(self) -> None:
+ # Device sub-paths still resolve to the adapter name.
+ self.assertEqual(adapter_path_to_name("/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF"), "hci0")
+
+ def test_zero_digits_allowed(self) -> None:
+ # The `[0-9]*` quantifier permits an "hci" with no index.
+ self.assertEqual(adapter_path_to_name("/org/bluez/hci"), "hci")
+ self.assertEqual(adapter_path_to_name("hci"), "hci")
+
+ def test_greedy_picks_last_occurrence(self) -> None:
+ # Greedy `.*` consumes up to the final "hci" match.
+ self.assertEqual(adapter_path_to_name("/org/bluez/hci0hci1"), "hci1")
+
+ def test_embedded_in_other_text(self) -> None:
+ self.assertEqual(adapter_path_to_name("prefix-hci99-suffix"), "hci99")
+
+
+class TestParseOsRelease(TestCase):
+ def _parse(self, content: str) -> dict[str, str]:
+ with tempfile.NamedTemporaryFile("w", suffix="os-release", delete=False) as f:
+ f.write(content)
+ path = Path(f.name)
+ try:
+ return parse_os_release(path)
+ finally:
+ path.unlink()
+
+ def test_basic_keys(self) -> None:
+ result = self._parse('NAME="Foo"\nVERSION="1.0"\n')
+ self.assertEqual(result, {"NAME": "Foo", "VERSION": "1.0"})
+
+ def test_value_with_equals_sign(self) -> None:
+ # Regression (data-3): a quoted value containing "=" must survive.
+ result = self._parse('PRETTY_NAME="Name=Variant"\n')
+ self.assertEqual(result["PRETTY_NAME"], "Name=Variant")
+
+ def test_unquoted_value(self) -> None:
+ self.assertEqual(self._parse("ID=arch\n"), {"ID": "arch"})
+
+ def test_comments_and_blank_lines_skipped(self) -> None:
+ result = self._parse("# comment\n\n \nID=foo\n")
+ self.assertEqual(result, {"ID": "foo"})
+
+ def test_line_without_equals_skipped(self) -> None:
+ result = self._parse("garbage line\nID=foo\n")
+ self.assertEqual(result, {"ID": "foo"})
+
+ def test_missing_file_returns_empty(self) -> None:
+ self.assertEqual(parse_os_release(Path("/nonexistent/os-release")), {})
+
+
+class TestHave(TestCase):
+ @patch("blueman.Functions.shutil.which", return_value="/usr/bin/dhclient")
+ def test_found_returns_path(self, which: object) -> None:
+ result = have("dhclient")
+ self.assertEqual(result, Path("/usr/bin/dhclient"))
+
+ @patch("blueman.Functions.shutil.which", return_value=None)
+ def test_not_found_returns_none(self, which: object) -> None:
+ self.assertIsNone(have("nonexistent-binary"))
+
+ @patch.dict(os.environ, {"PATH": "/usr/bin"}, clear=True)
+ @patch("blueman.Functions.shutil.which", return_value=None)
+ def test_appends_sbin_dirs(self, which: object) -> None:
+ have("dhcpcd")
+ used_path = which.call_args.kwargs["path"]
+ parts = used_path.split(os.pathsep)
+ self.assertIn("/usr/bin", parts)
+ self.assertIn("/sbin", parts)
+ self.assertIn("/usr/sbin", parts)
+
+ @patch.dict(os.environ, {"PATH": "/sbin:/usr/sbin:/usr/bin"}, clear=True)
+ @patch("blueman.Functions.shutil.which", return_value=None)
+ def test_does_not_duplicate_existing_sbin(self, which: object) -> None:
+ have("udhcpc")
+ used_path = which.call_args.kwargs["path"]
+ self.assertEqual(used_path.split(os.pathsep).count("/sbin"), 1)
+ self.assertEqual(used_path.split(os.pathsep).count("/usr/sbin"), 1)
+
+
+class TestCreateLogger(TestCase):
+ def setUp(self) -> None:
+ root = logging.getLogger(None)
+ self._saved_handlers = root.handlers[:]
+ self._saved_name = root.name
+ self._saved_level = root.level
+
+ def tearDown(self) -> None:
+ root = logging.getLogger(None)
+ root.handlers[:] = self._saved_handlers
+ root.name = self._saved_name
+ root.level = self._saved_level
+
+ @patch("blueman.Functions.logging.handlers.SysLogHandler")
+ def test_syslog_handler_added_when_available(self, handler_cls: MagicMock) -> None:
+ handler_cls.return_value = logging.NullHandler()
+ create_logger(logging.INFO, "blueman-test", syslog=True)
+ handler_cls.assert_called_once_with(address="/dev/log")
+
+ @patch("blueman.Functions.logging.handlers.SysLogHandler", side_effect=OSError)
+ def test_falls_back_to_stderr_when_dev_log_missing(self, handler_cls: MagicMock) -> None:
+ # Must not raise when /dev/log is unavailable, and add no syslog handler.
+ before = len(logging.getLogger(None).handlers)
+ logger = create_logger(logging.INFO, "blueman-test", syslog=True)
+ handler_cls.assert_called_once_with(address="/dev/log")
+ self.assertLessEqual(len(logger.handlers), before + 1) # only basicConfig's stderr handler, if any
+
+ @patch("blueman.Functions.logging.handlers.SysLogHandler")
+ def test_no_syslog_handler_when_disabled(self, handler_cls: MagicMock) -> None:
+ create_logger(logging.INFO, "blueman-test", syslog=False)
+ handler_cls.assert_not_called()
+
+
+class TestCreateParser(TestCase):
+ def test_default_loglevel(self) -> None:
+ args = create_parser().parse_args([])
+ self.assertEqual(args.LEVEL, "warning")
+
+ def test_valid_loglevel_lowercased(self) -> None:
+ args = create_parser().parse_args(["--loglevel", "DEBUG"])
+ self.assertEqual(args.LEVEL, "debug")
+
+ def test_invalid_loglevel_rejected(self) -> None:
+ # argparse exits with status 2 on an out-of-choices value.
+ with self.assertRaises(SystemExit):
+ create_parser().parse_args(["--loglevel", "verbose"])
+
+ def test_loglevel_has_help(self) -> None:
+ for action in create_parser()._actions:
+ if action.dest == "LEVEL":
+ self.assertTrue(action.help)
+ self.assertEqual(
+ set(action.choices), {"debug", "info", "warning", "error", "critical"})
+ break
+ else:
+ self.fail("--loglevel argument not registered")
+
+ def test_syslog_flag(self) -> None:
+ self.assertTrue(create_parser().parse_args(["--syslog"]).syslog)
+ self.assertFalse(create_parser().parse_args([]).syslog)
+
+ def test_loglevel_can_be_disabled(self) -> None:
+ parser = create_parser(loglevel=False)
+ self.assertFalse(any(a.dest == "LEVEL" for a in parser._actions))
+
+
+class TestCheckBluetoothStatus(TestCase):
+ @patch("blueman.Functions.AppletService", side_effect=DBusProxyFailed)
+ def test_logs_and_exits_when_applet_missing(self, applet: MagicMock) -> None:
+ exitfunc = MagicMock()
+ with self.assertLogs(level=logging.ERROR) as logs:
+ check_bluetooth_status("msg", exitfunc)
+ exitfunc.assert_called_once_with()
+ self.assertTrue(any("applet needs to be running" in m for m in logs.output))
+
+ @patch("blueman.Functions.AppletPowerManagerService")
+ @patch("blueman.Functions.AppletService")
+ def test_returns_when_powermanager_plugin_absent(
+ self, applet: MagicMock, power: MagicMock
+ ) -> None:
+ applet.return_value.QueryPlugins.return_value = []
+ exitfunc = MagicMock()
+ check_bluetooth_status("msg", exitfunc)
+ exitfunc.assert_not_called()
+
+
+class TestSetProcTitle(TestCase):
+ @patch("blueman.Functions.sys.platform", "darwin")
+ @patch("blueman.Functions.cdll")
+ def test_noop_on_non_linux(self, cdll: MagicMock) -> None:
+ self.assertEqual(set_proc_title("blueman"), 0)
+ cdll.LoadLibrary.assert_not_called()
+
+ @patch("blueman.Functions.sys.platform", "linux")
+ @patch("blueman.Functions.cdll")
+ def test_calls_prctl_on_linux(self, cdll: MagicMock) -> None:
+ cdll.LoadLibrary.return_value.prctl.return_value = 0
+ self.assertEqual(set_proc_title("blueman"), 0)
+ cdll.LoadLibrary.return_value.prctl.assert_called_once()
+
+ @patch("blueman.Functions.sys.platform", "linux")
+ @patch("blueman.Functions.cdll")
+ def test_returns_minus_one_when_libc_unavailable(self, cdll: MagicMock) -> None:
+ cdll.LoadLibrary.side_effect = OSError
+ with self.assertLogs(level=logging.ERROR):
+ self.assertEqual(set_proc_title("blueman"), -1)
+
+
+@patch("blueman.Functions.Gtk.get_current_event_time", return_value=0)
+@patch("blueman.Functions.Gio.File.new_for_commandline_arg")
+@patch("blueman.Functions.Gio.AppInfo.create_from_commandline")
+class TestLaunch(TestCase):
+ @staticmethod
+ def _command_line(create: MagicMock) -> str:
+ return create.call_args.args[0]
+
+ def test_legacy_string_form_unquoted(
+ self, create: MagicMock, _new_file: MagicMock, _evt: MagicMock
+ ) -> None:
+ create.return_value.launch.return_value = True
+ self.assertTrue(launch("blueman-services"))
+ self.assertEqual(self._command_line(create), (BIN_DIR / "blueman-services").as_posix())
+
+ def test_argv_form_quotes_each_token(
+ self, create: MagicMock, _new_file: MagicMock, _evt: MagicMock
+ ) -> None:
+ create.return_value.launch.return_value = True
+ launch("blueman-sendto", args=["--delete", "--device=AA:BB:CC:DD:EE:FF"])
+ expected = " ".join(
+ GLib.shell_quote(t) for t in (
+ (BIN_DIR / "blueman-sendto").as_posix(),
+ "--delete",
+ "--device=AA:BB:CC:DD:EE:FF",
+ )
+ )
+ self.assertEqual(self._command_line(create), expected)
+
+ def test_argv_form_neutralizes_shell_metacharacters(
+ self, create: MagicMock, _new_file: MagicMock, _evt: MagicMock
+ ) -> None:
+ create.return_value.launch.return_value = True
+ hostile = "foo; rm -rf ~"
+ launch("blueman-sendto", args=[hostile])
+ command_line = self._command_line(create)
+ # The hostile argument must survive as a single shell-quoted token.
+ self.assertIn(GLib.shell_quote(hostile), command_line)
+
+ def test_returns_launch_result(
+ self, create: MagicMock, _new_file: MagicMock, _evt: MagicMock
+ ) -> None:
+ create.return_value.launch.return_value = False
+ self.assertFalse(launch("blueman-services"))
+
+ def test_paths_become_gio_files(
+ self, create: MagicMock, new_file: MagicMock, _evt: MagicMock
+ ) -> None:
+ create.return_value.launch.return_value = True
+ launch("xdg-open", paths=["/tmp/a", "/tmp/b"], system=True)
+ self.assertEqual(new_file.call_count, 2)