"""Native cache loss between owner selection and one admitted GATT operation.""" import asyncio from types import SimpleNamespace import pytest from bleak.backends.device import BLEDevice from bleak.exc import BleakDeviceNotFoundError, BleakError from k1link.device_plugins.xgrids_k1.ble import scanner from k1link.device_plugins.xgrids_k1.ble import wifi_provisioning as wifi 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, ) @pytest.fixture(autouse=True) def isolated_owner(tmp_path): scanner.reset_runtime_handles_for_tests() reset_ble_runtime_arbiter_for_tests() configure_ble_runtime_process_lease(tmp_path) yield scanner.reset_runtime_handles_for_tests() reset_ble_runtime_arbiter_for_tests() @pytest.mark.parametrize("operation", ["read", "provision"]) @pytest.mark.parametrize("native", [ "macos", "present", "cache-cleanup-race", "vanished", "absent", "wrong-address", "wrong-adapter", "vanished-again", "owner-changed", "invalidated", "connect-failed", "connect-cancelled", "write-failed", ]) def test_selected_transport_survives_native_cache_loss(monkeypatch, operation, native): address = "AA:BB:CC:DD:EE:FF" path = "/org/bluez/hci7/dev_AA_BB_CC_DD_EE_FF" device = BLEDevice(address, "synthetic", {"path": path}) events = [] capture = None validations = [] async def retrieve(selected_address, details): assert selected_address == address and details["path"] == path events.append("cache") present = native in {"present", "cache-cleanup-race"} or ( events.count("cache") == 2 and native != "vanished-again" ) return device if present else None async def find(selected_address, *, timeout, bluez): assert selected_address == address assert 0 < timeout <= 8 assert bluez == {"adapter": "hci7"} events.append("scan") assert events.count("scan") == 1 if native == "absent": return None if native == "owner-changed": monkeypatch.setattr(scanner, "ble_runtime_owner_epoch_for_current_loop", lambda: -1) if native == "invalidated": scanner.demote_connected_device_handle_after_gatt_failure(capture) return BLEDevice( "AA:BB:CC:DD:EE:00" if native == "wrong-address" else address, "synthetic", {"path": path.replace("hci7", "hci8") if native == "wrong-adapter" else path}, ) service = SimpleNamespace(uuid=wifi.SERVICE_UUID) write = SimpleNamespace( uuid=wifi.WRITE_CHARACTERISTIC_UUID, service_uuid=wifi.SERVICE_UUID, properties=["write"], ) status = SimpleNamespace( uuid=wifi.STATUS_CHARACTERISTIC_UUID, service_uuid=wifi.SERVICE_UUID, properties=["read"], ) services = SimpleNamespace( get_service=lambda uuid: service if uuid == service.uuid else None, get_characteristic=lambda uuid: {write.uuid: write, status.uuid: status}.get(uuid), ) class Client: name = "synthetic" is_connected = True def __init__(self, selected, **_kwargs): # Restore the native path without replacing the owner capture. assert selected is device self.services = services async def __aenter__(self): events.append("connect") if native == "cache-cleanup-race": await asyncio.sleep(0) # The D-Bus connection yields to BlueZ cleanup. if events.count("hold-discovery") <= events.count("release-discovery"): raise BleakError("selected device not found before watcher registration") if native == "connect-cancelled": raise asyncio.CancelledError() if native == "connect-failed": raise BleakError("synthetic native connection failure") return self async def __aexit__(self, *_args): events.append("disconnect") async def read_gatt_char(self, characteristic): assert characteristic is status if native != "macos": assert "release-discovery" in events events.append("read") value = bytearray(52) value[0] = 11 value[1:12] = b"WIFI_CLIENT" value[33] = 4 value[34:38] = bytes((10, 255, 254, 77)) return value async def write_gatt_char(self, characteristic, _frame, *, response): assert characteristic is write and response is True assert events[-2:] == ["read", "dispatch"] events.append("write") if native == "write-failed": raise BleakError("synthetic unconfirmed write") monkeypatch.setattr(scanner, "sys", SimpleNamespace( platform="darwin" if native == "macos" else "linux", )) monkeypatch.setattr(scanner, "_retrieve_bluez_device", retrieve) class Scanner: find_device_by_address = staticmethod(find) def __init__(self, *, bluez): assert bluez == {"adapter": "hci7"} async def __aenter__(self): events.append("hold-discovery") return self async def __aexit__(self, *_args): events.append("release-discovery") monkeypatch.setattr(scanner, "BleakScanner", Scanner) monkeypatch.setattr(wifi, "BleakClient", Client) async def scenario(): nonlocal capture owner_epoch = bind_ble_runtime_owner_loop() capture = scanner.CapturedDiscoveredDevice( device=device, macos_uuid=address, owner_epoch=owner_epoch, ) scanner.pin_connected_device_handle(capture, device_session_id="synthetic-session") generation = scanner._runtime_handle_generation # noqa: SLF001 kwargs = dict(captured_device=capture, timeout_seconds=1) action = ( wifi.read_wifi_status_once(address, on_gatt_validated=validations.append, **kwargs) if operation == "read" else wifi.provision_wifi_once( address, "SyntheticNet", "x" * 13, write_mode="with_response", on_write_dispatch=lambda *_args: events.append("dispatch"), **kwargs, ) ) fails_before_connect = native in { "absent", "wrong-address", "wrong-adapter", "vanished-again", "owner-changed", "invalidated", } if native == "connect-cancelled": with pytest.raises(asyncio.CancelledError): await action assert "read" not in events and "write" not in events elif fails_before_connect or native == "connect-failed" or ( native == "write-failed" and operation == "provision" ): with pytest.raises(BleakError) as raised: await action if fails_before_connect: assert isinstance(raised.value, BleakDeviceNotFoundError) assert "connect" not in events if operation == "provision": assert raised.value.device_write_attempted is (native == "write-failed") assert events.count("write") == (native == "write-failed") else: result = await action assert result.get("outcome", "lan_address_observed") == "lan_address_observed" assert scanner.connected_device_capture( address, device_session_id="synthetic-session", ).device is device if operation == "read": assert validations == [capture] assert events.count("write") == (operation == "provision") assert scanner._runtime_handle_generation == generation # noqa: SLF001 asyncio.run(scenario()) assert events.count("scan") == (native not in {"macos", "present", "cache-cleanup-race"}) assert events.count("hold-discovery") == events.count("release-discovery") assert events.count("hold-discovery") == (native != "macos") assert events.count("connect") <= 1 assert events.count("write") <= 1