376 lines
13 KiB
Python
376 lines
13 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, BleakGATTProtocolError
|
|
|
|
import k1link.device_plugins.xgrids_k1.ble.ap_activation as ap_module
|
|
import k1link.device_plugins.xgrids_k1.ble.scanner as scanner_module
|
|
from k1link.device_plugins.xgrids_k1.ble.ap_activation import (
|
|
COMMAND_OFFSET,
|
|
ENABLE_AP_COMMAND,
|
|
FRAME_LENGTH,
|
|
build_ap_activation_frame,
|
|
is_ap_ready_status,
|
|
)
|
|
from k1link.device_plugins.xgrids_k1.ble.runtime_arbiter import (
|
|
BleRuntimeBusy,
|
|
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 WifiStatus
|
|
|
|
|
|
@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_ap_activation_frame_matches_reviewed_lixelgo_layout() -> None:
|
|
frame = build_ap_activation_frame()
|
|
|
|
assert len(frame) == FRAME_LENGTH == 100
|
|
assert frame[COMMAND_OFFSET] == ENABLE_AP_COMMAND == 1
|
|
assert frame[:COMMAND_OFFSET] == bytes(COMMAND_OFFSET)
|
|
|
|
|
|
def _status(*, reserved: int) -> WifiStatus:
|
|
return {
|
|
"value_length": 52,
|
|
"mode": "WIFI_AP",
|
|
"ipv4": "192.168.56.1",
|
|
"status_code": 1,
|
|
"reserved": reserved,
|
|
"trailer_hex": "",
|
|
}
|
|
|
|
|
|
def test_ap_control_mode_without_ready_flag_is_not_ready() -> None:
|
|
assert not is_ap_ready_status(_status(reserved=0))
|
|
|
|
|
|
def test_ap_ready_requires_the_reviewed_byte_51_flag() -> None:
|
|
assert is_ap_ready_status(_status(reserved=1))
|
|
|
|
|
|
def test_ap_activation_uses_retained_handle_without_rediscovery(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device_id = "synthetic-corebluetooth-uuid"
|
|
retained_handle = BLEDevice(device_id, "XGR-K1", details=object())
|
|
rediscovery_calls: list[tuple[str, float]] = []
|
|
client_calls: list[tuple[object, float, bool]] = []
|
|
|
|
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,
|
|
*,
|
|
timeout: float,
|
|
pair: bool,
|
|
) -> None:
|
|
client_calls.append((device, timeout, pair))
|
|
raise SelectedHandleObserved
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
|
|
_seed_scan_lease({device_id: retained_handle}, observed_at=100.0)
|
|
monkeypatch.setattr(
|
|
ap_module.BleakScanner,
|
|
"find_device_by_address",
|
|
forbidden_rediscovery,
|
|
)
|
|
monkeypatch.setattr(ap_module, "BleakClient", CapturingClient)
|
|
|
|
async def scenario() -> SelectedHandleObserved:
|
|
owner_epoch = bind_ble_runtime_owner_loop()
|
|
captured = scanner_module.CapturedDiscoveredDevice(
|
|
device=retained_handle,
|
|
macos_uuid=device_id,
|
|
owner_epoch=owner_epoch,
|
|
)
|
|
scanner_module.pin_connected_device_handle(
|
|
captured,
|
|
device_session_id="device-session-a",
|
|
)
|
|
with pytest.raises(SelectedHandleObserved) as caught:
|
|
await ap_module.activate_device_ap_once(
|
|
device_id,
|
|
timeout_seconds=1.0,
|
|
connect_timeout_seconds=30.0,
|
|
captured_device=captured,
|
|
)
|
|
assert (
|
|
scanner_module.connected_device_capture(
|
|
device_id,
|
|
device_session_id="device-session-a",
|
|
)
|
|
is None
|
|
)
|
|
return caught.value
|
|
|
|
error = asyncio.run(scenario())
|
|
|
|
assert rediscovery_calls == []
|
|
assert client_calls == [(retained_handle, 30.0, False)]
|
|
assert error.operation_stage == "connect" # 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]
|
|
|
|
|
|
def test_ap_activation_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(
|
|
ap_module.BleakScanner,
|
|
"find_device_by_address",
|
|
forbidden_rediscovery,
|
|
)
|
|
|
|
with pytest.raises(BleakDeviceNotFoundError) as caught:
|
|
asyncio.run(
|
|
ap_module.activate_device_ap_once(
|
|
"not-in-fresh-scan",
|
|
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_ap_activation_write_error_keeps_type_and_adds_safe_gatt_facts(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device_id = "synthetic-corebluetooth-uuid"
|
|
retained_handle = object()
|
|
service = SimpleNamespace(uuid=ap_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=ap_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=ap_module.SERVICE_UUID,
|
|
properties=["write-without-response", "write"],
|
|
max_write_without_response_size=512,
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=ap_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=ap_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
baseline = bytearray(52)
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == ap_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == ap_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == ap_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"
|
|
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 bytes(baseline)
|
|
|
|
async def write_gatt_char(self, *_args: object, **_kwargs: object) -> None:
|
|
raise BleakGATTProtocolError(0x03)
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
|
|
_seed_scan_lease({device_id: retained_handle}, observed_at=100.0)
|
|
monkeypatch.setattr(ap_module, "BleakClient", FailingWriteClient)
|
|
|
|
with pytest.raises(BleakGATTProtocolError) as caught:
|
|
asyncio.run(
|
|
ap_module.activate_device_ap_once(
|
|
device_id,
|
|
timeout_seconds=1.0,
|
|
)
|
|
)
|
|
|
|
error = caught.value
|
|
assert error.operation_stage == "gatt-write" # type: ignore[attr-defined]
|
|
assert error.device_write_attempted is True # type: ignore[attr-defined]
|
|
assert error.device_write_confirmed is False # type: ignore[attr-defined]
|
|
assert error.resolved_write_mode == "without_response" # type: ignore[attr-defined]
|
|
assert error.write_characteristic_properties == ( # type: ignore[attr-defined]
|
|
"write",
|
|
"write-without-response",
|
|
)
|
|
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 == 0x03 # type: ignore[attr-defined]
|
|
assert error.att_error_name == "WRITE_NOT_PERMITTED" # type: ignore[attr-defined]
|
|
|
|
|
|
def test_ap_activation_keeps_same_client_alive_through_caller_handoff(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
device_id = "synthetic-corebluetooth-uuid"
|
|
retained_handle = BLEDevice(device_id, "XGR-K1", details=object())
|
|
service = SimpleNamespace(uuid=ap_module.SERVICE_UUID)
|
|
write_characteristic = SimpleNamespace(
|
|
uuid=ap_module.WRITE_CHARACTERISTIC_UUID,
|
|
service_uuid=ap_module.SERVICE_UUID,
|
|
properties=["write"],
|
|
max_write_without_response_size=512,
|
|
)
|
|
status_characteristic = SimpleNamespace(
|
|
uuid=ap_module.STATUS_CHARACTERISTIC_UUID,
|
|
service_uuid=ap_module.SERVICE_UUID,
|
|
properties=["read"],
|
|
)
|
|
entered_clients = 0
|
|
exited_clients = 0
|
|
writes: list[tuple[bytes, bool]] = []
|
|
|
|
class FakeServices:
|
|
def get_service(self, uuid: str) -> object | None:
|
|
return service if uuid == ap_module.SERVICE_UUID else None
|
|
|
|
def get_characteristic(self, uuid: str) -> object | None:
|
|
if uuid == ap_module.WRITE_CHARACTERISTIC_UUID:
|
|
return write_characteristic
|
|
if uuid == ap_module.STATUS_CHARACTERISTIC_UUID:
|
|
return status_characteristic
|
|
return None
|
|
|
|
class ReadyClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
assert device is retained_handle
|
|
self.services = FakeServices()
|
|
self.name = "XGR-K1"
|
|
self.is_connected = False
|
|
self._write_completed = False
|
|
|
|
async def __aenter__(self) -> Any:
|
|
nonlocal entered_clients
|
|
entered_clients += 1
|
|
self.is_connected = True
|
|
return self
|
|
|
|
async def __aexit__(self, *_args: object) -> None:
|
|
nonlocal exited_clients
|
|
exited_clients += 1
|
|
self.is_connected = False
|
|
|
|
async def read_gatt_char(self, _characteristic: object) -> bytes:
|
|
if not self._write_completed:
|
|
return bytes(52)
|
|
ready = bytearray(52)
|
|
mode = b"WIFI_AP"
|
|
ready[0] = len(mode)
|
|
ready[1 : 1 + len(mode)] = mode
|
|
ready[33] = 4
|
|
ready[34:38] = bytes((192, 168, 56, 1))
|
|
ready[50] = 1
|
|
ready[51] = 1
|
|
return bytes(ready)
|
|
|
|
async def write_gatt_char(
|
|
self,
|
|
_characteristic: object,
|
|
value: bytes,
|
|
*,
|
|
response: bool,
|
|
) -> None:
|
|
writes.append((bytes(value), response))
|
|
self._write_completed = True
|
|
|
|
monkeypatch.setattr(scanner_module, "monotonic", lambda: 100.0)
|
|
_seed_scan_lease({device_id: retained_handle}, observed_at=100.0)
|
|
monkeypatch.setattr(ap_module, "BleakClient", ReadyClient)
|
|
|
|
async def scenario() -> None:
|
|
owner_epoch = bind_ble_runtime_owner_loop()
|
|
captured = scanner_module.CapturedDiscoveredDevice(
|
|
device=retained_handle,
|
|
macos_uuid=device_id,
|
|
owner_epoch=owner_epoch,
|
|
)
|
|
scanner_module.pin_connected_device_handle(
|
|
captured,
|
|
device_session_id="device-session-a",
|
|
)
|
|
async with ap_module.device_ap_activation_session(
|
|
device_id,
|
|
timeout_seconds=0.1,
|
|
poll_interval_seconds=0.01,
|
|
captured_device=captured,
|
|
) as result:
|
|
assert result["ready_observed"] is True
|
|
assert entered_clients == 1
|
|
assert exited_clients == 0
|
|
assert ble_runtime_snapshot()["active_operation_kind"] == "ap-enable"
|
|
with pytest.raises(BleRuntimeBusy) as busy:
|
|
await scanner_module.scan(0.01)
|
|
assert busy.value.active_operation_kind == "ap-enable"
|
|
# This represents the host CoreWLAN association window: the setup
|
|
# deadline is over, but the exact same BLE client must remain alive.
|
|
await asyncio.sleep(0.02)
|
|
assert exited_clients == 0
|
|
assert scanner_module.connected_device_recovery_snapshot(
|
|
device_id,
|
|
device_session_id="device-session-a",
|
|
)["gatt_validated_recently"] is True
|
|
|
|
assert exited_clients == 1
|
|
assert await wait_for_ble_runtime_idle()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
assert len(writes) == 1
|
|
payload, response = writes[0]
|
|
assert len(payload) == FRAME_LENGTH
|
|
assert payload[:COMMAND_OFFSET] == bytes(COMMAND_OFFSET)
|
|
assert payload[COMMAND_OFFSET] == ENABLE_AP_COMMAND
|
|
assert response is True
|