import asyncio from types import SimpleNamespace from typing import Any import pytest 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, 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"