141 lines
5.6 KiB
Python
141 lines
5.6 KiB
Python
"""Exercise the installed Bleak manager against BlueZ's single-session contract.
|
|
|
|
Only the D-Bus wire is synthetic. No Bluetooth adapter or device is accessed.
|
|
"""
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from bleak import BleakScanner
|
|
from bleak.backends.bluezdbus import defs
|
|
from bleak.backends.bluezdbus import manager as manager_module
|
|
from bleak.backends.bluezdbus import scanner as backend_module
|
|
from bleak.backends.bluezdbus.manager import BlueZManager
|
|
from bleak.backends.bluezdbus.scanner import BleakScannerBlueZDBus
|
|
from bleak.backends.device import BLEDevice
|
|
from bleak.exc import BleakDBusError
|
|
from dbus_fast import Message, MessageType
|
|
|
|
from k1link.device_plugins.xgrids_k1.ble import scanner
|
|
from k1link.device_plugins.xgrids_k1.ble.runtime_arbiter import (
|
|
bind_ble_runtime_owner_loop,
|
|
configure_ble_runtime_process_lease,
|
|
reset_ble_runtime_arbiter_for_tests,
|
|
run_ble_operation,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["status-read", "wifi-provision", "ap-enable"])
|
|
@pytest.mark.parametrize(
|
|
"mode", ["present", "missing", "busy", "absent", "cancelled", "connect-failed"]
|
|
)
|
|
def test_one_discovery_session_retains_exact_path_until_connect(monkeypatch, tmp_path, kind, mode):
|
|
async def scenario():
|
|
reset_ble_runtime_arbiter_for_tests()
|
|
configure_ble_runtime_process_lease(tmp_path)
|
|
bind_ble_runtime_owner_loop()
|
|
manager = BlueZManager()
|
|
adapter = "/org/bluez/hci7"
|
|
address = "AA:BB:CC:DD:EE:FF"
|
|
path = adapter + "/dev_AA_BB_CC_DD_EE_FF"
|
|
device = BLEDevice(address, "synthetic", {"path": path})
|
|
manager._properties[adapter] = {defs.ADAPTER_INTERFACE: {"Powered": True}}
|
|
calls = []
|
|
discovery_active = False
|
|
|
|
def advertise():
|
|
props = {
|
|
"Address": address,
|
|
"Alias": "synthetic",
|
|
"Name": "synthetic",
|
|
"Adapter": adapter,
|
|
"RSSI": -45,
|
|
}
|
|
manager._properties[path] = {defs.DEVICE_INTERFACE: props}
|
|
for callback in tuple(manager._advertisement_callbacks[adapter]):
|
|
callback(path, props)
|
|
|
|
class Bus:
|
|
async def call(self, message):
|
|
nonlocal discovery_active
|
|
calls.append(message.member)
|
|
if message.member == "StartDiscovery":
|
|
if discovery_active or mode == "busy":
|
|
return Message(
|
|
message_type=MessageType.ERROR,
|
|
reply_serial=1,
|
|
error_name="org.bluez.Error.InProgress",
|
|
signature="s",
|
|
body=["Operation already in progress"],
|
|
)
|
|
discovery_active = True
|
|
if mode != "absent":
|
|
asyncio.get_running_loop().call_later(0.01, advertise)
|
|
elif message.member == "StopDiscovery":
|
|
discovery_active = False
|
|
manager._properties.pop(path, None)
|
|
return Message(message_type=MessageType.METHOD_RETURN, reply_serial=1)
|
|
|
|
manager._bus = Bus()
|
|
if mode == "present":
|
|
advertise()
|
|
|
|
async def get_manager():
|
|
return manager
|
|
|
|
monkeypatch.setattr(manager_module, "get_global_bluez_manager", get_manager)
|
|
monkeypatch.setattr(backend_module, "get_global_bluez_manager", get_manager)
|
|
monkeypatch.setattr(scanner, "sys", SimpleNamespace(platform="linux"))
|
|
|
|
class LinuxScanner(BleakScanner):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, backend=BleakScannerBlueZDBus, **kwargs)
|
|
|
|
monkeypatch.setattr(scanner, "BleakScanner", LinuxScanner)
|
|
|
|
async def connect():
|
|
async with scanner.selected_device_discovery(device, timeout_seconds=1):
|
|
await asyncio.sleep(0)
|
|
assert manager.get_device_address(path) == address
|
|
assert discovery_active
|
|
calls.append("Connect")
|
|
if mode == "cancelled":
|
|
raise asyncio.CancelledError()
|
|
if mode == "connect-failed":
|
|
raise ConnectionError("synthetic connection failure")
|
|
|
|
try:
|
|
action = run_ble_operation(
|
|
kind, operation=lambda _progress: connect(), hard_timeout_seconds=2
|
|
)
|
|
expected_error = {
|
|
"busy": BleakDBusError,
|
|
"absent": TimeoutError,
|
|
"cancelled": asyncio.CancelledError,
|
|
"connect-failed": ConnectionError,
|
|
}.get(mode)
|
|
if expected_error:
|
|
with pytest.raises(expected_error) as raised:
|
|
await action
|
|
if mode in {"busy", "absent"}:
|
|
assert "Connect" not in calls
|
|
assert raised.value.reason_code == (
|
|
"ble-discovery-busy"
|
|
if mode == "busy"
|
|
else "ble-selected-device-unavailable"
|
|
)
|
|
else:
|
|
await action
|
|
assert calls.count("StartDiscovery") == 1
|
|
assert calls.count("StopDiscovery") == (mode != "busy")
|
|
if "Connect" in calls:
|
|
assert calls.index("Connect") < calls.index("StopDiscovery")
|
|
assert not discovery_active
|
|
assert not manager._advertisement_callbacks[adapter]
|
|
assert not manager._device_removed_callbacks
|
|
finally:
|
|
reset_ble_runtime_arbiter_for_tests()
|
|
|
|
asyncio.run(scenario())
|