228 lines
7.2 KiB
Python
228 lines
7.2 KiB
Python
import asyncio
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from bleak.exc import BleakDeviceNotFoundError
|
|
|
|
import k1link.device_plugins.xgrids_k1.ble.wifi_provisioning as wifi_module
|
|
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,
|
|
)
|
|
|
|
|
|
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",
|
|
"ipv4": "192.168.56.1",
|
|
"status_code": 1,
|
|
"reserved": 0,
|
|
"trailer_hex": "5858",
|
|
}
|
|
|
|
|
|
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
|
|
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:
|
|
return characteristic if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID else None
|
|
|
|
class FakeClient:
|
|
def __init__(self, _device: object, **_kwargs: object) -> None:
|
|
self.services = FakeServices()
|
|
self.name = "XGR-K1"
|
|
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")
|
|
|
|
monkeypatch.setattr(wifi_module, "discovered_device", lambda _uuid: object())
|
|
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["status"]["ipv4"] == "10.255.254.77"
|
|
|
|
|
|
def test_read_wifi_status_recovery_rediscover_ignores_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
|
|
stale_handle = object()
|
|
recovered_handle = object()
|
|
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:
|
|
return characteristic if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID else None
|
|
|
|
class FakeClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
assert device is recovered_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:
|
|
return recovered_handle
|
|
|
|
monkeypatch.setattr(wifi_module, "discovered_device", lambda _uuid: stale_handle)
|
|
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"
|
|
|
|
|
|
def test_provisioning_write_requires_fresh_rediscovery_before_connecting(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
stale_handle = object()
|
|
rediscovery_calls: list[tuple[str, float]] = []
|
|
client_calls: list[object] = []
|
|
|
|
async def missing_device(address: str, *, timeout: float) -> None:
|
|
rediscovery_calls.append((address, timeout))
|
|
return None
|
|
|
|
class ForbiddenClient:
|
|
def __init__(self, device: object, **_kwargs: object) -> None:
|
|
client_calls.append(device)
|
|
raise AssertionError("a failed fresh discovery must stop before the BLE write session")
|
|
|
|
monkeypatch.setattr(wifi_module, "discovered_device", lambda _uuid: stale_handle)
|
|
monkeypatch.setattr(
|
|
wifi_module.BleakScanner,
|
|
"find_device_by_address",
|
|
missing_device,
|
|
)
|
|
monkeypatch.setattr(wifi_module, "BleakClient", ForbiddenClient)
|
|
|
|
with pytest.raises(BleakDeviceNotFoundError):
|
|
asyncio.run(
|
|
provision_wifi_once(
|
|
"synthetic-corebluetooth-uuid",
|
|
"LabNet",
|
|
"synthetic-password",
|
|
timeout_seconds=1.0,
|
|
)
|
|
)
|
|
|
|
assert rediscovery_calls == [("synthetic-corebluetooth-uuid", 1.0)]
|
|
assert client_calls == []
|