from copy import deepcopy import pytest from k1link.device_plugins.xgrids_k1.wifi_failure import reviewed_station_failure_code def station_error(code: int = 4) -> dict[str, object]: return { "code": "BleakGATTProtocolError", "operation_stage": "gatt-write", "device_write_attempted": True, "device_write_confirmed": False, "resolved_write_mode": "with_response", "frame_length": 99, "ble_att_error_code": code, "ble_att_error_name": "INVALID_PDU", "safe_to_retry": False, "side_effect_status": "unknown", } @pytest.mark.parametrize("mode", ["bridge", "direct-connect"]) @pytest.mark.parametrize( "code,reason", [(4, "k1-wifi-network-not-found"), (6, "k1-wifi-credentials-required")] ) def test_reviewed_station_reply_retains_raw_facts_and_retry_fence( mode: str, code: int, reason: str ) -> None: error = station_error(code) before = deepcopy(error) assert ( reviewed_station_failure_code(error, firmware_version="3.0.2", connection_mode=mode) == reason ) assert error == before @pytest.mark.parametrize( "field,value", [ ("code", "RuntimeError"), ("operation_stage", "baseline-read"), ("device_write_attempted", False), ("device_write_confirmed", True), ("resolved_write_mode", "without_response"), ("frame_length", 100), ("ble_att_error_code", 14), ("ble_att_error_code", "4"), ("ble_att_error_code", {}), ], ) def test_unrelated_transport_failures_are_not_wifi_diagnoses(field: str, value: object) -> None: error = {**station_error(), field: value} assert ( reviewed_station_failure_code(error, firmware_version="3.0.2", connection_mode="bridge") is None ) @pytest.mark.parametrize( "firmware,mode", [("3.0.1", "bridge"), ("unknown", "bridge"), ("3.0.2", "quick-connect")] ) def test_other_firmware_and_quick_connect_do_not_inherit_station_codes( firmware: str, mode: str ) -> None: assert ( reviewed_station_failure_code( station_error(), firmware_version=firmware, connection_mode=mode ) is None )