1408 lines
49 KiB
Python
1408 lines
49 KiB
Python
import asyncio
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from bleak.backends.device import BLEDevice
|
|
from bleak.exc import BleakDeviceNotFoundError, BleakError, BleakGATTProtocolError
|
|
|
|
import k1link.device_plugins.xgrids_k1.ble.scanner as scanner_module
|
|
import k1link.device_plugins.xgrids_k1.ble.wifi_provisioning as wifi_module
|
|
from k1link.device_plugins.xgrids_k1.ble.runtime_arbiter import (
|
|
BleOperationHardTimeout,
|
|
bind_ble_runtime_owner_loop,
|
|
ble_runtime_snapshot,
|
|
configure_ble_runtime_process_lease,
|
|
reset_ble_runtime_arbiter_for_tests,
|
|
wait_for_ble_runtime_idle,
|
|
)
|
|
from k1link.device_plugins.xgrids_k1.ble.wifi_provisioning import (
|
|
FRAME_LENGTH,
|
|
build_wifi_provisioning_frame,
|
|
parse_wifi_status,
|
|
provision_wifi_once,
|
|
read_wifi_status_once,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_runtime_handle_lease(tmp_path: Path) -> Iterator[None]:
|
|
scanner_module.reset_runtime_handles_for_tests()
|
|
reset_ble_runtime_arbiter_for_tests()
|
|
configure_ble_runtime_process_lease(tmp_path)
|
|
yield
|
|
scanner_module.reset_runtime_handles_for_tests()
|
|
reset_ble_runtime_arbiter_for_tests()
|
|
|
|
|
|
def _seed_scan_lease(handles: dict[str, object], *, observed_at: float) -> None:
|
|
with scanner_module._runtime_handle_lock: # noqa: SLF001
|
|
scanner_module._runtime_handles.clear() # noqa: SLF001
|
|
scanner_module._runtime_handles.update(handles) # type: ignore[arg-type] # noqa: SLF001
|
|
scanner_module._runtime_handle_observed_at_monotonic = observed_at # noqa: SLF001
|
|
scanner_module._runtime_handle_generation += 1 # noqa: SLF001
|
|
|
|
|
|
def test_build_wifi_provisioning_frame_layout() -> None:
|
|
credential = "x" * 13
|
|
frame = build_wifi_provisioning_frame("LabNet", credential)
|
|
|
|
assert len(frame) == FRAME_LENGTH
|
|
assert frame[0] == 6
|
|
assert frame[1:7] == b"LabNet"
|
|
assert frame[7:33] == bytes(26)
|
|
assert frame[33] == 13
|
|
assert frame[34:47] == b"x" * 13
|
|
assert frame[47:98] == bytes(51)
|
|
assert frame[98] == 0
|
|
|
|
|
|
def test_build_wifi_provisioning_frame_uses_utf8_byte_lengths() -> None:
|
|
ssid = "Ж" * 4
|
|
credential = "я" * 6
|
|
frame = build_wifi_provisioning_frame(ssid, credential)
|
|
|
|
assert frame[0] == len(ssid.encode())
|
|
assert frame[33] == len(credential.encode())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("ssid", "password", "message"),
|
|
[
|
|
("", "x" * 8, "SSID must not be empty"),
|
|
("network", "", "password must not be empty"),
|
|
("x" * 33, "y" * 8, "at most 32 UTF-8 bytes"),
|
|
("network", "x" * 65, "at most 64 UTF-8 bytes"),
|
|
],
|
|
)
|
|
def test_build_wifi_provisioning_frame_rejects_invalid_lengths(
|
|
ssid: str,
|
|
password: str,
|
|
message: str,
|
|
) -> None:
|
|
with pytest.raises(ValueError, match=message):
|
|
build_wifi_provisioning_frame(ssid, password)
|
|
|
|
|
|
def test_parse_wifi_status_ap_baseline() -> None:
|
|
value = bytearray(54)
|
|
value[0] = 7
|
|
value[1:8] = b"WIFI_AP"
|
|
value[33] = 4
|
|
value[34:38] = bytes((192, 168, 56, 1))
|
|
value[50] = 1
|
|
value[52:54] = b"XX"
|
|
|
|
assert parse_wifi_status(bytes(value)) == {
|
|
"value_length": 54,
|
|
"mode": "WIFI_AP",
|
|
"network_name": None,
|
|
"ipv4": "192.168.56.1",
|
|
"status_code": 1,
|
|
"reserved": 0,
|
|
"trailer_hex": "5858",
|
|
}
|
|
|
|
|
|
def test_parse_wifi_status_classifies_fw302_station_network_name() -> None:
|
|
"""Regression for the redacted shape observed from the physical K1."""
|
|
|
|
network_name = b"LAB_NETWORK"
|
|
address = bytes((192, 168, 68, 51))
|
|
value = bytearray(52)
|
|
value[0] = len(network_name)
|
|
value[1 : 1 + len(network_name)] = network_name
|
|
value[33] = len(address)
|
|
value[34 : 34 + len(address)] = address
|
|
value[50] = 1
|
|
|
|
assert parse_wifi_status(bytes(value)) == {
|
|
"value_length": 52,
|
|
"mode": "WIFI_CLIENT",
|
|
"network_name": "LAB_NETWORK",
|
|
"ipv4": "192.168.68.51",
|
|
"status_code": 1,
|
|
"reserved": 0,
|
|
"trailer_hex": "",
|
|
}
|
|
|
|
|
|
def test_parse_wifi_status_rejects_short_frame() -> None:
|
|
with pytest.raises(ValueError, match="at least 51 bytes"):
|
|
parse_wifi_status(bytes(50))
|
|
|
|
|
|
def test_read_wifi_status_once_reads_only_and_returns_current_dhcp_address(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
value = bytearray(54)
|
|
value[0] = 11
|
|
value[1:12] = b"WIFI_CLIENT"
|
|
value[33] = 4
|
|
value[34:38] = bytes((10, 255, 254, 77))
|
|
value[50] = 1
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write-without-response", "write"],
|
|
max_write_without_response_size=244,
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
connected = []
|
|
|
|
class FakeClient:
|
|
def __init__(self, _device: object, **_kwargs: object) -> None:
|
|
connected.append(_device)
|
|
self.services = FakeServices()
|
|
self.name = "XGR-K1"
|
|
self.mtu_size = 256
|
|
self.write_calls = 0
|
|
|
|
async def __aenter__(self) -> Any:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
async def read_gatt_char(self, _characteristic: object) -> bytes:
|
|
return bytes(value)
|
|
|
|
async def write_gatt_char(self, *_args: object, **_kwargs: object) -> None:
|
|
self.write_calls += 1
|
|
raise AssertionError("status refresh must not write a BLE characteristic")
|
|
|
|
retained_handle = object()
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"discovered_device_selection",
|
|
lambda _uuid: SimpleNamespace(
|
|
device=retained_handle,
|
|
from_fresh_scan=True,
|
|
),
|
|
)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", FakeClient)
|
|
|
|
result = asyncio.run(read_wifi_status_once("synthetic-corebluetooth-uuid"))
|
|
|
|
assert result["operation"] == "single_reviewed_wifi_status_read"
|
|
assert result["write_performed"] is False
|
|
assert result["write_characteristic_uuid"] == wifi_module.WRITE_CHARACTERISTIC_UUID
|
|
assert result["write_characteristic_properties"] == ["write", "write-without-response"]
|
|
assert result["max_write_without_response_size"] == 244
|
|
assert result["mtu_size"] == 256
|
|
assert result["status"]["ipv4"] == "10.255.254.77"
|
|
assert connected == [retained_handle]
|
|
|
|
|
|
def test_read_wifi_status_recovery_keeps_fresh_retained_handle(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
value = bytearray(54)
|
|
value[0] = 11
|
|
value[1:12] = b"WIFI_CLIENT"
|
|
value[33] = 4
|
|
value[34:38] = bytes((10, 255, 254, 77))
|
|
value[50] = 1
|
|
retained_handle = object()
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write"],
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
class FakeClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
assert device is retained_handle
|
|
self.services = FakeServices()
|
|
self.name = "XGR-K1"
|
|
|
|
async def __aenter__(self) -> Any:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
async def read_gatt_char(self, _characteristic: object) -> bytes:
|
|
return bytes(value)
|
|
|
|
async def rediscover(*_args: object, **_kwargs: object) -> object:
|
|
raise AssertionError("a fresh explicit scan handle must not be discarded")
|
|
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"discovered_device_selection",
|
|
lambda _uuid: SimpleNamespace(
|
|
device=retained_handle,
|
|
from_fresh_scan=True,
|
|
),
|
|
)
|
|
monkeypatch.setattr(wifi_module.BleakScanner, "find_device_by_address", rediscover)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", FakeClient)
|
|
|
|
result = asyncio.run(
|
|
read_wifi_status_once(
|
|
"synthetic-corebluetooth-uuid",
|
|
rediscover=True,
|
|
)
|
|
)
|
|
|
|
assert result["status"]["ipv4"] == "10.255.254.77"
|
|
assert result["max_write_without_response_size"] is None
|
|
assert result["mtu_size"] is None
|
|
|
|
|
|
def test_retrieved_capture_status_read_connects_and_validates_7f02_without_write(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device = BLEDevice("RETRIEVED-UUID", "XGR-RETRIEVED", details=object())
|
|
value = bytearray(54)
|
|
value[0] = 11
|
|
value[1:12] = b"WIFI_CLIENT"
|
|
value[33] = 4
|
|
value[34:38] = bytes((192, 168, 68, 52))
|
|
value[50] = 1
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write"],
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
events: list[str] = []
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
class ReadOnlyClient:
|
|
def __init__(self, selected: object, **_kwargs: object) -> None:
|
|
assert selected is device
|
|
events.append("new-connect")
|
|
self.services = FakeServices()
|
|
self.name = "XGR-RETRIEVED"
|
|
|
|
async def __aenter__(self) -> Any:
|
|
events.append("connected")
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
events.append("disconnected")
|
|
|
|
async def read_gatt_char(self, characteristic: object) -> bytes:
|
|
assert characteristic is status_characteristic
|
|
events.append("read-7f02")
|
|
return bytes(value)
|
|
|
|
async def write_gatt_char(self, *_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("retrieved status validation must remain read-only")
|
|
|
|
def forbidden_selection(_uuid: str) -> object:
|
|
raise AssertionError("an exact retrieved capture must not use scan selection")
|
|
|
|
async def forbidden_lookup(*_args: object, **_kwargs: object) -> object:
|
|
raise AssertionError("an exact retrieved capture must not use UUID lookup")
|
|
|
|
monkeypatch.setattr(wifi_module, "BleakClient", ReadOnlyClient)
|
|
monkeypatch.setattr(wifi_module, "discovered_device_selection", forbidden_selection)
|
|
monkeypatch.setattr(
|
|
wifi_module.BleakScanner,
|
|
"find_device_by_address",
|
|
forbidden_lookup,
|
|
)
|
|
|
|
async def scenario() -> None:
|
|
owner_epoch = bind_ble_runtime_owner_loop()
|
|
now = scanner_module._freshness_now() # noqa: SLF001
|
|
initial_capture = scanner_module.CapturedDiscoveredDevice(
|
|
device=device,
|
|
macos_uuid=device.address,
|
|
owner_epoch=owner_epoch,
|
|
scan_generation=9,
|
|
captured_at_monotonic=now.monotonic,
|
|
captured_at_suspend_aware=now.suspend_aware,
|
|
source="fresh-scan",
|
|
)
|
|
scanner_module.pin_connected_device_handle(
|
|
initial_capture,
|
|
device_session_id="device-session-a",
|
|
)
|
|
captured = scanner_module.CapturedDiscoveredDevice(
|
|
device=device,
|
|
macos_uuid=device.address,
|
|
owner_epoch=owner_epoch,
|
|
scan_generation=9,
|
|
captured_at_monotonic=now.monotonic,
|
|
captured_at_suspend_aware=now.suspend_aware,
|
|
source="retrieved-session",
|
|
)
|
|
|
|
result = await read_wifi_status_once(
|
|
device.address,
|
|
timeout_seconds=1.0,
|
|
captured_device=captured,
|
|
)
|
|
assert result["write_performed"] is False
|
|
assert result["status"]["ipv4"] == "192.168.68.52"
|
|
assert scanner_module.connected_device_recovery_snapshot(
|
|
device.address,
|
|
device_session_id="device-session-a",
|
|
) == {
|
|
"status": "retained",
|
|
"scope": "owner-epoch-device-session",
|
|
"advertised_now": False,
|
|
"gatt_validated_recently": True,
|
|
}
|
|
|
|
asyncio.run(scenario())
|
|
assert events == ["new-connect", "connected", "read-7f02", "disconnected"]
|
|
|
|
|
|
def test_durable_uuid_status_read_retrieves_then_connects_and_returns_capture(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
macos_uuid = "11111111-2222-4333-8444-555555555555"
|
|
device = BLEDevice(macos_uuid, "XGR-DURABLE", details=object())
|
|
value = bytearray(54)
|
|
value[0] = 11
|
|
value[1:12] = b"WIFI_CLIENT"
|
|
value[33] = 4
|
|
value[34:38] = bytes((192, 168, 68, 52))
|
|
value[50] = 1
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write"],
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
events: list[str] = []
|
|
returned_captures: list[scanner_module.CapturedDiscoveredDevice] = []
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
class ReadOnlyClient:
|
|
def __init__(self, selected: object, **_kwargs: object) -> None:
|
|
assert selected is device
|
|
self.services = FakeServices()
|
|
self.name = "XGR-DURABLE"
|
|
events.append("connect")
|
|
|
|
async def __aenter__(self) -> Any:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
events.append("disconnect")
|
|
|
|
async def read_gatt_char(self, characteristic: object) -> bytes:
|
|
assert characteristic is status_characteristic
|
|
events.append("read-7f02")
|
|
return bytes(value)
|
|
|
|
async def write_gatt_char(self, *_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("durable UUID reconciliation must not write")
|
|
|
|
async def retrieve(requested_uuid: str) -> scanner_module.CapturedDiscoveredDevice:
|
|
assert requested_uuid == macos_uuid
|
|
owner_epoch = scanner_module.ble_runtime_owner_epoch_for_current_loop()
|
|
assert owner_epoch is not None
|
|
now = scanner_module._freshness_now() # noqa: SLF001
|
|
events.append("retrieve-known-uuid")
|
|
return scanner_module.CapturedDiscoveredDevice(
|
|
device=device,
|
|
macos_uuid=macos_uuid,
|
|
owner_epoch=owner_epoch,
|
|
captured_at_monotonic=now.monotonic,
|
|
captured_at_suspend_aware=now.suspend_aware,
|
|
source="retrieved-durable",
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"discovered_device_selection",
|
|
lambda _uuid: SimpleNamespace(device=None, from_fresh_scan=False),
|
|
)
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"retrieve_known_device_capture_for_status_read",
|
|
retrieve,
|
|
)
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"discover_known_device_capture_for_status_read",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(
|
|
AssertionError("cached retrieval must not start an advertisement scan")
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
wifi_module.BleakScanner,
|
|
"find_device_by_address",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(
|
|
AssertionError("durable retrieval must not fall through to a scan")
|
|
),
|
|
)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", ReadOnlyClient)
|
|
|
|
result = asyncio.run(
|
|
read_wifi_status_once(
|
|
macos_uuid,
|
|
timeout_seconds=1.0,
|
|
allow_known_device_retrieval=True,
|
|
on_gatt_validated=returned_captures.append,
|
|
)
|
|
)
|
|
|
|
assert result["write_performed"] is False
|
|
assert result["status"]["ipv4"] == "192.168.68.52"
|
|
assert len(returned_captures) == 1
|
|
assert returned_captures[0].source == "retrieved-durable"
|
|
assert events == ["retrieve-known-uuid", "connect", "read-7f02", "disconnect"]
|
|
|
|
|
|
def test_durable_uuid_rediscover_scans_first_then_reads_once_without_write(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
macos_uuid = "AAAAAAAA-BBBB-4CCC-8DDD-EEEEEEEEEEEE"
|
|
device = BLEDevice(macos_uuid, "XGR-ADVERTISEMENT", details=object())
|
|
value = bytearray(54)
|
|
value[0] = 11
|
|
value[1:12] = b"WIFI_CLIENT"
|
|
value[33] = 4
|
|
value[34:38] = bytes((192, 168, 68, 99))
|
|
value[50] = 1
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write"],
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
events: list[str] = []
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
class ReadOnlyClient:
|
|
def __init__(self, selected: object, **_kwargs: object) -> None:
|
|
assert selected is device
|
|
events.append("connect")
|
|
self.services = FakeServices()
|
|
self.name = "XGR-ADVERTISEMENT"
|
|
|
|
async def __aenter__(self) -> Any:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
events.append("disconnect")
|
|
|
|
async def read_gatt_char(self, characteristic: object) -> bytes:
|
|
assert characteristic is status_characteristic
|
|
events.append("read-7f02")
|
|
return bytes(value)
|
|
|
|
async def write_gatt_char(self, *_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("advertisement recovery must remain read-only")
|
|
|
|
async def exact_scan(
|
|
requested_uuid: str,
|
|
*,
|
|
timeout_seconds: float,
|
|
) -> scanner_module.CapturedDiscoveredDevice:
|
|
assert requested_uuid == macos_uuid
|
|
assert timeout_seconds == 30.0
|
|
owner_epoch = scanner_module.ble_runtime_owner_epoch_for_current_loop()
|
|
assert owner_epoch is not None
|
|
now = scanner_module._freshness_now() # noqa: SLF001
|
|
events.append("exact-uuid-scan")
|
|
return scanner_module.CapturedDiscoveredDevice(
|
|
device=device,
|
|
macos_uuid=macos_uuid,
|
|
owner_epoch=owner_epoch,
|
|
captured_at_monotonic=now.monotonic,
|
|
captured_at_suspend_aware=now.suspend_aware,
|
|
source="retrieved-durable",
|
|
)
|
|
|
|
async def forbidden_retrieval(
|
|
_requested_uuid: str,
|
|
) -> scanner_module.CapturedDiscoveredDevice | None:
|
|
raise AssertionError("rediscover must not connect a cached peripheral first")
|
|
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"discover_known_device_capture_for_status_read",
|
|
exact_scan,
|
|
)
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"retrieve_known_device_capture_for_status_read",
|
|
forbidden_retrieval,
|
|
)
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"discovered_device_selection",
|
|
lambda _uuid: (_ for _ in ()).throw(
|
|
AssertionError("durable rediscovery must not consume public scan state")
|
|
),
|
|
)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", ReadOnlyClient)
|
|
|
|
result = asyncio.run(
|
|
read_wifi_status_once(
|
|
macos_uuid,
|
|
timeout_seconds=1.0,
|
|
exact_scan_timeout_seconds=30.0,
|
|
rediscover=True,
|
|
allow_known_device_retrieval=True,
|
|
)
|
|
)
|
|
|
|
assert result["write_performed"] is False
|
|
assert result["status"]["ipv4"] == "192.168.68.99"
|
|
assert events == ["exact-uuid-scan", "connect", "read-7f02", "disconnect"]
|
|
|
|
|
|
def test_durable_uuid_rediscover_timeout_fails_without_cache_or_connect(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
macos_uuid = "AAAAAAAA-BBBB-4CCC-8DDD-EEEEEEEEEEEE"
|
|
scan_calls: list[tuple[str, float]] = []
|
|
|
|
async def exact_scan(
|
|
requested_uuid: str,
|
|
*,
|
|
timeout_seconds: float,
|
|
) -> scanner_module.CapturedDiscoveredDevice | None:
|
|
scan_calls.append((requested_uuid, timeout_seconds))
|
|
return None
|
|
|
|
async def forbidden_retrieval(
|
|
_requested_uuid: str,
|
|
) -> scanner_module.CapturedDiscoveredDevice | None:
|
|
raise AssertionError("scan timeout must not fall back to cached retrieval")
|
|
|
|
class ForbiddenClient:
|
|
def __init__(self, *_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("scan timeout must fail before GATT connect")
|
|
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"discover_known_device_capture_for_status_read",
|
|
exact_scan,
|
|
)
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"retrieve_known_device_capture_for_status_read",
|
|
forbidden_retrieval,
|
|
)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", ForbiddenClient)
|
|
|
|
with pytest.raises(BleakDeviceNotFoundError) as raised:
|
|
asyncio.run(
|
|
read_wifi_status_once(
|
|
macos_uuid,
|
|
timeout_seconds=1.0,
|
|
exact_scan_timeout_seconds=30.0,
|
|
rediscover=True,
|
|
allow_known_device_retrieval=True,
|
|
)
|
|
)
|
|
|
|
assert scan_calls == [(macos_uuid, 30.0)]
|
|
assert raised.value.operation_stage == "exact-uuid-scan" # type: ignore[attr-defined]
|
|
|
|
|
|
def test_durable_uuid_retrieval_exception_preserves_sanitized_resolution_stage(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def failing_retrieval(
|
|
_requested_uuid: str,
|
|
) -> scanner_module.CapturedDiscoveredDevice | None:
|
|
raise BleakError("synthetic native retrieval detail")
|
|
|
|
class ForbiddenClient:
|
|
def __init__(self, *_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("retrieval failure must occur before GATT connect")
|
|
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"retrieve_known_device_capture_for_status_read",
|
|
failing_retrieval,
|
|
)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", ForbiddenClient)
|
|
|
|
with pytest.raises(BleakError) as raised:
|
|
asyncio.run(
|
|
read_wifi_status_once(
|
|
"AAAAAAAA-BBBB-4CCC-8DDD-EEEEEEEEEEEE",
|
|
timeout_seconds=1.0,
|
|
allow_known_device_retrieval=True,
|
|
)
|
|
)
|
|
|
|
assert raised.value.operation_stage == "resolution" # type: ignore[attr-defined]
|
|
|
|
|
|
def test_durable_uuid_rediscover_hard_timeout_includes_scan_and_gatt_budgets(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
hard_timeouts: list[float] = []
|
|
|
|
async def capture_hard_timeout(
|
|
_operation_kind: str,
|
|
*,
|
|
hard_timeout_seconds: float,
|
|
operation: object,
|
|
progress: object,
|
|
) -> dict[str, object]:
|
|
del operation, progress
|
|
hard_timeouts.append(hard_timeout_seconds)
|
|
return {}
|
|
|
|
monkeypatch.setattr(wifi_module, "run_ble_operation", capture_hard_timeout)
|
|
|
|
asyncio.run(
|
|
read_wifi_status_once(
|
|
"AAAAAAAA-BBBB-4CCC-8DDD-EEEEEEEEEEEE",
|
|
timeout_seconds=20.0,
|
|
exact_scan_timeout_seconds=30.0,
|
|
rediscover=True,
|
|
allow_known_device_retrieval=True,
|
|
)
|
|
)
|
|
|
|
assert hard_timeouts == [55.0]
|
|
|
|
|
|
def test_failed_retrieved_status_read_invalidates_session_token(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device = BLEDevice("OFFLINE-RETRIEVED-UUID", "XGR-OFFLINE", details=object())
|
|
|
|
class OfflineClient:
|
|
def __init__(self, selected: object, **_kwargs: object) -> None:
|
|
assert selected is device
|
|
|
|
async def __aenter__(self) -> Any:
|
|
raise BleakDeviceNotFoundError(device.address, "powered off")
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(wifi_module, "BleakClient", OfflineClient)
|
|
|
|
async def scenario() -> None:
|
|
owner_epoch = bind_ble_runtime_owner_loop()
|
|
now = scanner_module._freshness_now() # noqa: SLF001
|
|
initial_capture = scanner_module.CapturedDiscoveredDevice(
|
|
device=device,
|
|
macos_uuid=device.address,
|
|
owner_epoch=owner_epoch,
|
|
captured_at_monotonic=now.monotonic,
|
|
captured_at_suspend_aware=now.suspend_aware,
|
|
source="fresh-scan",
|
|
)
|
|
scanner_module.pin_connected_device_handle(
|
|
initial_capture,
|
|
device_session_id="device-session-a",
|
|
)
|
|
captured = scanner_module.CapturedDiscoveredDevice(
|
|
device=device,
|
|
macos_uuid=device.address,
|
|
owner_epoch=owner_epoch,
|
|
captured_at_monotonic=now.monotonic,
|
|
captured_at_suspend_aware=now.suspend_aware,
|
|
source="retrieved-session",
|
|
)
|
|
|
|
with pytest.raises(BleakDeviceNotFoundError):
|
|
await read_wifi_status_once(
|
|
device.address,
|
|
timeout_seconds=1.0,
|
|
captured_device=captured,
|
|
)
|
|
assert (
|
|
scanner_module.connected_device_capture(
|
|
device.address,
|
|
device_session_id="device-session-a",
|
|
)
|
|
is None
|
|
)
|
|
assert scanner_module.connected_device_recovery_snapshot(
|
|
device.address,
|
|
device_session_id="device-session-a",
|
|
) == {
|
|
"status": "unavailable",
|
|
"scope": "owner-epoch-device-session",
|
|
"advertised_now": False,
|
|
"gatt_validated_recently": False,
|
|
}
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_provisioning_write_uses_retained_handle_without_rediscovery(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device_id = "synthetic-corebluetooth-uuid"
|
|
retained_handle = object()
|
|
rediscovery_calls: list[tuple[str, float]] = []
|
|
client_calls: list[object] = []
|
|
|
|
class SelectedHandleObserved(RuntimeError):
|
|
pass
|
|
|
|
async def forbidden_rediscovery(address: str, *, timeout: float) -> None:
|
|
rediscovery_calls.append((address, timeout))
|
|
raise AssertionError("a fresh explicit scan handle must be used directly")
|
|
|
|
class CapturingClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
client_calls.append(device)
|
|
raise SelectedHandleObserved
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
|
|
_seed_scan_lease({device_id: retained_handle}, observed_at=100.0)
|
|
monkeypatch.setattr(
|
|
wifi_module.BleakScanner,
|
|
"find_device_by_address",
|
|
forbidden_rediscovery,
|
|
)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", CapturingClient)
|
|
|
|
with pytest.raises(SelectedHandleObserved) as caught:
|
|
asyncio.run(
|
|
provision_wifi_once(
|
|
device_id,
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=1.0,
|
|
)
|
|
)
|
|
|
|
assert rediscovery_calls == []
|
|
assert client_calls == [retained_handle]
|
|
assert caught.value.operation_stage == "connect" # type: ignore[attr-defined]
|
|
assert caught.value.device_write_attempted is False # type: ignore[attr-defined]
|
|
assert caught.value.device_write_confirmed is False # type: ignore[attr-defined]
|
|
|
|
|
|
def test_provisioning_write_does_not_fallback_when_fresh_scan_omits_device(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
rediscovery_calls: list[tuple[str, float]] = []
|
|
|
|
async def forbidden_rediscovery(address: str, *, timeout: float) -> None:
|
|
rediscovery_calls.append((address, timeout))
|
|
return None
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
|
|
_seed_scan_lease({}, observed_at=100.0)
|
|
monkeypatch.setattr(
|
|
wifi_module.BleakScanner,
|
|
"find_device_by_address",
|
|
forbidden_rediscovery,
|
|
)
|
|
|
|
with pytest.raises(BleakDeviceNotFoundError) as caught:
|
|
asyncio.run(
|
|
provision_wifi_once(
|
|
"not-in-fresh-scan",
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=1.0,
|
|
)
|
|
)
|
|
|
|
assert rediscovery_calls == []
|
|
assert caught.value.operation_stage == "resolution" # type: ignore[attr-defined]
|
|
assert caught.value.device_write_attempted is False # type: ignore[attr-defined]
|
|
assert caught.value.device_write_confirmed is False # type: ignore[attr-defined]
|
|
|
|
|
|
def test_provisioning_uses_latest_selected_handle_without_age_based_rediscovery(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device_id = "synthetic-corebluetooth-uuid"
|
|
expired_handle = object()
|
|
rediscovered_handle = object()
|
|
rediscovery_calls: list[tuple[str, float]] = []
|
|
client_calls: list[object] = []
|
|
clock = [100.0]
|
|
|
|
class RediscoveredHandleObserved(RuntimeError):
|
|
pass
|
|
|
|
async def rediscover(address: str, *, timeout: float) -> object:
|
|
rediscovery_calls.append((address, timeout))
|
|
return rediscovered_handle
|
|
|
|
class CapturingClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
client_calls.append(device)
|
|
raise RediscoveredHandleObserved
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: clock[0])
|
|
_seed_scan_lease({device_id: expired_handle}, observed_at=clock[0])
|
|
clock[0] += scanner_module.BLE_RUNTIME_HANDLE_LEASE_TTL_SECONDS + 0.001
|
|
monkeypatch.setattr(wifi_module.BleakScanner, "find_device_by_address", rediscover)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", CapturingClient)
|
|
|
|
with pytest.raises(RediscoveredHandleObserved):
|
|
asyncio.run(
|
|
provision_wifi_once(
|
|
device_id,
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=1.0,
|
|
)
|
|
)
|
|
|
|
assert rediscovery_calls == []
|
|
assert client_calls == [expired_handle]
|
|
assert scanner_module.discovered_device(device_id) is None
|
|
|
|
|
|
def test_provisioning_baseline_error_keeps_type_and_adds_safe_gatt_facts(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device_id = "synthetic-corebluetooth-uuid"
|
|
retained_handle = object()
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write"],
|
|
max_write_without_response_size=512,
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
class FailingBaselineClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
assert device is retained_handle
|
|
self.services = FakeServices()
|
|
self.name = "XGR-K1"
|
|
|
|
async def __aenter__(self) -> Any:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
async def read_gatt_char(self, _characteristic: object) -> bytes:
|
|
raise BleakGATTProtocolError(0x0E)
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
|
|
_seed_scan_lease({device_id: retained_handle}, observed_at=100.0)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", FailingBaselineClient)
|
|
|
|
with pytest.raises(BleakGATTProtocolError) as caught:
|
|
asyncio.run(
|
|
provision_wifi_once(
|
|
device_id,
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=1.0,
|
|
)
|
|
)
|
|
|
|
error = caught.value
|
|
assert error.operation_stage == "baseline-read" # type: ignore[attr-defined]
|
|
assert error.device_write_attempted is False # type: ignore[attr-defined]
|
|
assert error.device_write_confirmed is False # type: ignore[attr-defined]
|
|
assert error.resolved_write_mode == "with_response" # type: ignore[attr-defined]
|
|
assert error.write_characteristic_properties == ("write",) # type: ignore[attr-defined]
|
|
assert error.max_write_without_response_size == 512 # type: ignore[attr-defined]
|
|
assert error.frame_length == FRAME_LENGTH # type: ignore[attr-defined]
|
|
assert error.att_error_code == 0x0E # type: ignore[attr-defined]
|
|
assert error.att_error_name == "UNLIKELY_ERROR" # type: ignore[attr-defined]
|
|
|
|
|
|
def test_provisioning_write_error_adds_only_safe_resolved_gatt_facts(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device_id = "synthetic-corebluetooth-uuid"
|
|
retained_handle = object()
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write"],
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
class FailingWriteClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
assert device is retained_handle
|
|
self.services = FakeServices()
|
|
self.name = "XGR-K1"
|
|
|
|
async def __aenter__(self) -> Any:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
async def read_gatt_char(self, _characteristic: object) -> bytes:
|
|
return bytes(52)
|
|
|
|
async def write_gatt_char(
|
|
self,
|
|
_characteristic: object,
|
|
_value: bytes,
|
|
*,
|
|
response: bool,
|
|
) -> None:
|
|
assert response is True
|
|
raise BleakGATTProtocolError(0x03)
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
|
|
_seed_scan_lease({device_id: retained_handle}, observed_at=100.0)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", FailingWriteClient)
|
|
|
|
with pytest.raises(BleakGATTProtocolError) as caught:
|
|
asyncio.run(
|
|
provision_wifi_once(
|
|
device_id,
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=1.0,
|
|
)
|
|
)
|
|
|
|
error = caught.value
|
|
assert error.operation_stage == "gatt-write" # type: ignore[attr-defined]
|
|
assert error.resolved_write_mode == "with_response" # type: ignore[attr-defined]
|
|
assert error.write_characteristic_properties == ("write",) # type: ignore[attr-defined]
|
|
assert error.max_write_without_response_size is None # type: ignore[attr-defined]
|
|
assert error.frame_length == FRAME_LENGTH # type: ignore[attr-defined]
|
|
diagnostic_text = repr(vars(error))
|
|
assert "LabNet" not in diagnostic_text
|
|
assert "synthetic-password" not in diagnostic_text
|
|
assert "payload" not in vars(error)
|
|
assert "frame" not in vars(error)
|
|
|
|
|
|
def test_provisioning_status_poll_error_reports_confirmed_write(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device_id = "synthetic-corebluetooth-uuid"
|
|
retained_handle = object()
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write"],
|
|
max_write_without_response_size=512,
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
baseline = bytearray(52)
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
class FailingPollClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
assert device is retained_handle
|
|
self.services = FakeServices()
|
|
self.name = "XGR-K1"
|
|
self.is_connected = True
|
|
self.read_count = 0
|
|
|
|
async def __aenter__(self) -> Any:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
async def read_gatt_char(self, _characteristic: object) -> bytes:
|
|
self.read_count += 1
|
|
if self.read_count == 1:
|
|
return bytes(baseline)
|
|
raise BleakGATTProtocolError(0x12)
|
|
|
|
async def write_gatt_char(self, *_args: object, **_kwargs: object) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
|
|
_seed_scan_lease({device_id: retained_handle}, observed_at=100.0)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", FailingPollClient)
|
|
|
|
with pytest.raises(BleakGATTProtocolError) as caught:
|
|
asyncio.run(
|
|
provision_wifi_once(
|
|
device_id,
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=1.0,
|
|
)
|
|
)
|
|
|
|
error = caught.value
|
|
assert error.operation_stage == "status-poll" # type: ignore[attr-defined]
|
|
assert error.device_write_attempted is True # type: ignore[attr-defined]
|
|
assert error.device_write_confirmed is True # type: ignore[attr-defined]
|
|
assert error.resolved_write_mode == "with_response" # type: ignore[attr-defined]
|
|
assert error.write_characteristic_properties == ("write",) # type: ignore[attr-defined]
|
|
assert error.max_write_without_response_size == 512 # type: ignore[attr-defined]
|
|
assert error.frame_length == FRAME_LENGTH # type: ignore[attr-defined]
|
|
assert error.att_error_code == 0x12 # type: ignore[attr-defined]
|
|
assert error.att_error_name == "DATABASE_OUT_OF_SYNC" # type: ignore[attr-defined]
|
|
|
|
|
|
def test_provision_deadline_during_gatt_write_reports_ambiguous_metadata(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device_id = "synthetic-corebluetooth-uuid"
|
|
retained_handle = object()
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write-without-response"],
|
|
max_write_without_response_size=512,
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
baseline = bytes(52)
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
async def scenario() -> None:
|
|
cancellation_observed = asyncio.Event()
|
|
cleanup_release = asyncio.Event()
|
|
|
|
class StuckWriteClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
assert device is retained_handle
|
|
self.services = FakeServices()
|
|
self.name = "XGR-K1"
|
|
self.is_connected = True
|
|
|
|
async def __aenter__(self) -> Any:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
async def read_gatt_char(self, _characteristic: object) -> bytes:
|
|
return baseline
|
|
|
|
async def write_gatt_char(self, *_args: object, **_kwargs: object) -> None:
|
|
try:
|
|
await asyncio.Event().wait()
|
|
except asyncio.CancelledError:
|
|
cancellation_observed.set()
|
|
await cleanup_release.wait()
|
|
raise
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
|
|
_seed_scan_lease({device_id: retained_handle}, observed_at=100.0)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", StuckWriteClient)
|
|
monkeypatch.setattr(
|
|
wifi_module,
|
|
"BLE_PROVISION_HARD_TIMEOUT_GRACE_SECONDS",
|
|
0.01,
|
|
)
|
|
|
|
try:
|
|
with pytest.raises(BleOperationHardTimeout) as raised:
|
|
await provision_wifi_once(
|
|
device_id,
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=0.01,
|
|
)
|
|
assert raised.value.reason_code == "ble-provisioning-timeout"
|
|
assert raised.value.operation_stage == "gatt-write"
|
|
assert raised.value.device_write_attempted is True
|
|
assert raised.value.device_write_confirmed is False
|
|
assert "LabNet" not in str(raised.value)
|
|
assert "synthetic-password" not in str(raised.value)
|
|
await cancellation_observed.wait()
|
|
assert ble_runtime_snapshot()["cleanup_pending"] is True
|
|
finally:
|
|
cleanup_release.set()
|
|
assert await wait_for_ble_runtime_idle()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_captured_recovery_handle_reads_7f02_before_exactly_one_station_write(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device = BLEDevice("RECOVERY-UUID", "XGR-RECOVERY", details=object())
|
|
service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["write"],
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=wifi_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=wifi_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == wifi_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == wifi_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
baseline = bytearray(52)
|
|
baseline[0] = 7
|
|
baseline[1:8] = b"WIFI_AP"
|
|
baseline[33] = 4
|
|
baseline[34:38] = bytes((192, 168, 56, 1))
|
|
connected = bytearray(52)
|
|
connected[0] = 11
|
|
connected[1:12] = b"WIFI_CLIENT"
|
|
connected[33] = 4
|
|
connected[34:38] = bytes((192, 168, 68, 50))
|
|
events: list[str] = []
|
|
|
|
class RecordingClient:
|
|
def __init__(self, selected: object, **_kwargs: object) -> None:
|
|
assert selected is device
|
|
self.services = FakeServices()
|
|
self.name = "XGR-RECOVERY"
|
|
self.is_connected = True
|
|
|
|
async def __aenter__(self) -> Any:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
async def read_gatt_char(self, _characteristic: object) -> bytes:
|
|
events.append("read-7f02")
|
|
return bytes(baseline if events.count("write-7f01") == 0 else connected)
|
|
|
|
async def write_gatt_char(
|
|
self,
|
|
_characteristic: object,
|
|
_value: bytes,
|
|
*,
|
|
response: bool,
|
|
) -> None:
|
|
assert response is True
|
|
events.append("write-7f01")
|
|
|
|
monkeypatch.setattr(wifi_module, "BleakClient", RecordingClient)
|
|
|
|
async def scenario() -> None:
|
|
owner_epoch = bind_ble_runtime_owner_loop()
|
|
captured = scanner_module.CapturedDiscoveredDevice(
|
|
device=device,
|
|
macos_uuid=device.address,
|
|
owner_epoch=owner_epoch,
|
|
)
|
|
scanner_module.pin_connected_device_handle(
|
|
captured,
|
|
device_session_id="device-session-a",
|
|
)
|
|
result = await provision_wifi_once(
|
|
device.address,
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=1.0,
|
|
write_mode="with_response",
|
|
captured_device=captured,
|
|
)
|
|
assert result["outcome"] == "lan_address_observed"
|
|
assert scanner_module.connected_device_recovery_snapshot(
|
|
device.address,
|
|
device_session_id="device-session-a",
|
|
)["gatt_validated_recently"] is True
|
|
|
|
asyncio.run(scenario())
|
|
assert events == ["read-7f02", "write-7f01", "read-7f02"]
|
|
|
|
|
|
def test_powered_off_recovery_handle_fails_before_station_write(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device = BLEDevice("OFFLINE-UUID", "XGR-OFFLINE", details=object())
|
|
|
|
class OfflineClient:
|
|
def __init__(self, selected: object, **_kwargs: object) -> None:
|
|
assert selected is device
|
|
|
|
async def __aenter__(self) -> Any:
|
|
raise BleakDeviceNotFoundError(device.address, "powered off")
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
async def write_gatt_char(self, *_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("powered-off recovery must fail before 7f01")
|
|
|
|
monkeypatch.setattr(wifi_module, "BleakClient", OfflineClient)
|
|
|
|
async def scenario() -> None:
|
|
owner_epoch = bind_ble_runtime_owner_loop()
|
|
captured = scanner_module.CapturedDiscoveredDevice(
|
|
device=device,
|
|
macos_uuid=device.address,
|
|
owner_epoch=owner_epoch,
|
|
)
|
|
scanner_module.pin_connected_device_handle(
|
|
captured,
|
|
device_session_id="device-session-a",
|
|
)
|
|
with pytest.raises(BleakDeviceNotFoundError) as raised:
|
|
await provision_wifi_once(
|
|
device.address,
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=1.0,
|
|
write_mode="with_response",
|
|
captured_device=captured,
|
|
)
|
|
assert raised.value.operation_stage == "connect" # type: ignore[attr-defined]
|
|
assert raised.value.device_write_attempted is False # type: ignore[attr-defined]
|
|
assert (
|
|
scanner_module.connected_device_capture(
|
|
device.address,
|
|
device_session_id="device-session-a",
|
|
)
|
|
is None
|
|
)
|
|
assert scanner_module.connected_device_recovery_snapshot(
|
|
device.address,
|
|
device_session_id="device-session-a",
|
|
)["status"] == "unavailable"
|
|
|
|
asyncio.run(scenario())
|