| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -2,9 +2,11 @@ | |||
| 2 | 2 | ||
| 3 | 3 | from __future__ import annotations | |
| 4 | 4 | ||
| 5 | + import base64 | ||
| 5 | 6 | import copy | |
| 6 | 7 | import logging | |
| 7 | 8 | import time | |
| 9 | + import types | ||
| 8 | 10 | from collections import OrderedDict | |
| 9 | 11 | from typing import TYPE_CHECKING, Any, cast | |
| 10 | 12 | from unittest.mock import patch | |
@@ -13,7 +15,15 @@ | |||
| 13 | 15 | from freezegun.api import FrozenDateTimeFactory | |
| 14 | 16 | from pytest_mock import MockerFixture | |
| 15 | 17 | ||
| 16 | - from kasa import Device, DeviceType, KasaException, Module | ||
| 18 | + from kasa import ( | ||
| 19 | + AuthenticationError, | ||
| 20 | + Credentials, | ||
| 21 | + Device, | ||
| 22 | + DeviceType, | ||
| 23 | + KasaException, | ||
| 24 | + Module, | ||
| 25 | + ) | ||
| 26 | + from kasa.device import WifiNetwork | ||
| 17 | 27 | from kasa.exceptions import DeviceError, SmartErrorCode | |
| 18 | 28 | from kasa.smart import SmartDevice | |
| 19 | 29 | from kasa.smart.modules.energy import Energy | |
@@ -1010,3 +1020,347 @@ async def test_unpair(dev: SmartDevice, mocker: MockerFixture): | |||
| 1010 | 1020 | await unpair_feat.set_value(None) | |
| 1011 | 1021 | ||
| 1012 | 1022 | unpair_call.assert_called_with(child.device_id) | |
| 1023 | + | ||
| 1024 | + | ||
| 1025 | + @pytest.mark.requires_dummy | ||
| 1026 | + async def test_wifi_scan_legacy(dummy_protocol): | ||
| 1027 | + """Verify legacy wifi scan parses base64 ssid values.""" | ||
| 1028 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1029 | + ssid = "LegacyWifi" | ||
| 1030 | + ap_list = [ | ||
| 1031 | + { | ||
| 1032 | + "ssid": base64.b64encode(ssid.encode()).decode(), | ||
| 1033 | + "cipher_type": 4, | ||
| 1034 | + "key_type": 3, | ||
| 1035 | + "channel": 6, | ||
| 1036 | + "signal_level": 2, | ||
| 1037 | + "bssid": "aa:bb:cc:dd:ee:ff", | ||
| 1038 | + } | ||
| 1039 | + ] | ||
| 1040 | + | ||
| 1041 | + async def _query(request, *_, **__): | ||
| 1042 | + assert request == {"get_wireless_scan_info": {"start_index": 0}} | ||
| 1043 | + return {"get_wireless_scan_info": {"ap_list": ap_list}} | ||
| 1044 | + | ||
| 1045 | + dev.protocol.query = _query | ||
| 1046 | + networks = await dev.wifi_scan() | ||
| 1047 | + | ||
| 1048 | + assert dev.wpa3_supported is None | ||
| 1049 | + assert dev.public_key is None | ||
| 1050 | + assert networks[0].ssid == ssid | ||
| 1051 | + assert networks[0].bssid == "aa:bb:cc:dd:ee:ff" | ||
| 1052 | + | ||
| 1053 | + | ||
| 1054 | + @pytest.mark.requires_dummy | ||
| 1055 | + async def test_wifi_scan_fallback_scanaplist(dummy_protocol): | ||
| 1056 | + """Verify scanApList fallback stores wpa3/public_key and parses fields.""" | ||
| 1057 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1058 | + ap_list = [ | ||
| 1059 | + { | ||
| 1060 | + "ssid": "ModernWifi", | ||
| 1061 | + "bssid": "11:22:33:44:55:66", | ||
| 1062 | + "auth": 3, | ||
| 1063 | + "encryption": 2, | ||
| 1064 | + "rssi": 2, | ||
| 1065 | + } | ||
| 1066 | + ] | ||
| 1067 | + | ||
| 1068 | + async def _query(request, *_, **__): | ||
| 1069 | + if "get_wireless_scan_info" in request: | ||
| 1070 | + raise KasaException("scan not supported") | ||
| 1071 | + assert request == {"scanApList": {"onboarding": {"scan": {}}}} | ||
| 1072 | + return { | ||
| 1073 | + "wpa3_supported": "true", | ||
| 1074 | + "public_key": " key ", | ||
| 1075 | + "scanApList": {"onboarding": {"scan": {"ap_list": ap_list}}}, | ||
| 1076 | + } | ||
| 1077 | + | ||
| 1078 | + dev.protocol.query = _query | ||
| 1079 | + networks = await dev.wifi_scan() | ||
| 1080 | + | ||
| 1081 | + assert dev.wpa3_supported is True | ||
| 1082 | + assert dev.public_key == "key" | ||
| 1083 | + assert networks[0].ssid == "ModernWifi" | ||
| 1084 | + assert networks[0].auth == 3 | ||
| 1085 | + assert networks[0].encryption == 2 | ||
| 1086 | + assert networks[0].rssi == 2 | ||
| 1087 | + | ||
| 1088 | + | ||
| 1089 | + @pytest.mark.requires_dummy | ||
| 1090 | + async def test_wifi_scan_fallback_bool_and_missing_key(dummy_protocol): | ||
| 1091 | + """Verify scanApList fallback handles bool wpa3 and non-string public key.""" | ||
| 1092 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1093 | + ap_list = [ | ||
| 1094 | + { | ||
| 1095 | + "ssid": "ModernWifi2", | ||
| 1096 | + "bssid": "22:33:44:55:66:77", | ||
| 1097 | + "auth": 4, | ||
| 1098 | + "encryption": 1, | ||
| 1099 | + "rssi": 3, | ||
| 1100 | + } | ||
| 1101 | + ] | ||
| 1102 | + | ||
| 1103 | + async def _query(request, *_, **__): | ||
| 1104 | + if "get_wireless_scan_info" in request: | ||
| 1105 | + raise KasaException("scan not supported") | ||
| 1106 | + assert request == {"scanApList": {"onboarding": {"scan": {}}}} | ||
| 1107 | + return { | ||
| 1108 | + "wpa3_supported": True, | ||
| 1109 | + "public_key": 123, | ||
| 1110 | + "scanApList": {"onboarding": {"scan": {"ap_list": ap_list}}}, | ||
| 1111 | + } | ||
| 1112 | + | ||
| 1113 | + dev.protocol.query = _query | ||
| 1114 | + networks = await dev.wifi_scan() | ||
| 1115 | + | ||
| 1116 | + assert dev.wpa3_supported is True | ||
| 1117 | + assert dev.public_key is None | ||
| 1118 | + assert networks[0].ssid == "ModernWifi2" | ||
| 1119 | + assert networks[0].auth == 4 | ||
| 1120 | + | ||
| 1121 | + | ||
| 1122 | + @pytest.mark.requires_dummy | ||
| 1123 | + async def test_wifi_scan_fallback_false_and_empty_key(dummy_protocol): | ||
| 1124 | + """Verify scanApList fallback handles false wpa3 and empty public key.""" | ||
| 1125 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1126 | + ap_list = [ | ||
| 1127 | + { | ||
| 1128 | + "ssid": "ModernWifi3", | ||
| 1129 | + "bssid": "33:44:55:66:77:88", | ||
| 1130 | + "auth": 2, | ||
| 1131 | + "encryption": 1, | ||
| 1132 | + "rssi": 1, | ||
| 1133 | + } | ||
| 1134 | + ] | ||
| 1135 | + | ||
| 1136 | + async def _query(request, *_, **__): | ||
| 1137 | + if "get_wireless_scan_info" in request: | ||
| 1138 | + raise KasaException("scan not supported") | ||
| 1139 | + assert request == {"scanApList": {"onboarding": {"scan": {}}}} | ||
| 1140 | + return { | ||
| 1141 | + "wpa3_supported": "false", | ||
| 1142 | + "public_key": " ", | ||
| 1143 | + "scanApList": {"onboarding": {"scan": {"ap_list": ap_list}}}, | ||
| 1144 | + } | ||
| 1145 | + | ||
| 1146 | + dev.protocol.query = _query | ||
| 1147 | + networks = await dev.wifi_scan() | ||
| 1148 | + | ||
| 1149 | + assert dev.wpa3_supported is False | ||
| 1150 | + assert dev.public_key is None | ||
| 1151 | + assert networks[0].ssid == "ModernWifi3" | ||
| 1152 | + | ||
| 1153 | + | ||
| 1154 | + @pytest.mark.requires_dummy | ||
| 1155 | + async def test_wifi_scan_invalid_scan_branches(dummy_protocol): | ||
| 1156 | + """Cover invalid scan branches inside wifi_scan helper functions.""" | ||
| 1157 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1158 | + | ||
| 1159 | + def _get_code(name: str) -> types.CodeType: | ||
| 1160 | + for const in dev.wifi_scan.__code__.co_consts: | ||
| 1161 | + if isinstance(const, types.CodeType) and const.co_name == name: | ||
| 1162 | + return const | ||
| 1163 | + raise AssertionError(f"Missing {name} code object") | ||
| 1164 | + | ||
| 1165 | + def _make_cell(value): | ||
| 1166 | + def _inner(): | ||
| 1167 | + return value | ||
| 1168 | + | ||
| 1169 | + return _inner.__closure__[0] | ||
| 1170 | + | ||
| 1171 | + scan_code = _get_code("_scan") | ||
| 1172 | + scan_fn = types.FunctionType( | ||
| 1173 | + scan_code, | ||
| 1174 | + dev.wifi_scan.__globals__, | ||
| 1175 | + name="_scan", | ||
| 1176 | + closure=(_make_cell(dev),), | ||
| 1177 | + ) | ||
| 1178 | + with pytest.raises(KasaException, match="Unknown wifi scan target"): | ||
| 1179 | + await scan_fn("bogus_target") | ||
| 1180 | + | ||
| 1181 | + handle_code = _get_code("_handle_scan_response") | ||
| 1182 | + handle_fn = types.FunctionType( | ||
| 1183 | + handle_code, | ||
| 1184 | + dev.wifi_scan.__globals__, | ||
| 1185 | + name="_handle_scan_response", | ||
| 1186 | + ) | ||
| 1187 | + with pytest.raises(KasaException, match="Unknown wifi scan response"): | ||
| 1188 | + handle_fn({}, "bogus_scan") | ||
| 1189 | + | ||
| 1190 | + net_code = _get_code("_net_for_scan_info") | ||
| 1191 | + net_fn = types.FunctionType( | ||
| 1192 | + net_code, | ||
| 1193 | + dev.wifi_scan.__globals__, | ||
| 1194 | + name="_net_for_scan_info", | ||
| 1195 | + ) | ||
| 1196 | + with pytest.raises(KasaException, match="Unknown wifi scan info"): | ||
| 1197 | + net_fn({}, "bogus_scan") | ||
| 1198 | + | ||
| 1199 | + | ||
| 1200 | + @pytest.mark.requires_dummy | ||
| 1201 | + async def test_wifi_join_connectap_payload_timeout(dummy_protocol): | ||
| 1202 | + """Verify connectAp payload types and timeout handling.""" | ||
| 1203 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1204 | + dev.protocol._transport._credentials = Credentials("user", "pass") | ||
| 1205 | + dev.networks = [ | ||
| 1206 | + WifiNetwork( | ||
| 1207 | + ssid="ModernWifi", | ||
| 1208 | + auth=3, | ||
| 1209 | + encryption=2, | ||
| 1210 | + rssi=2, | ||
| 1211 | + bssid="11:22:33:44:55:66", | ||
| 1212 | + ) | ||
| 1213 | + ] | ||
| 1214 | + | ||
| 1215 | + async def _query(request, retry_count=3, *_, **__): | ||
| 1216 | + assert retry_count == 0 | ||
| 1217 | + payload = request["connectAp"]["onboarding"]["connect"] | ||
| 1218 | + assert isinstance(payload["ssid"], str) | ||
| 1219 | + assert isinstance(payload["bssid"], str) | ||
| 1220 | + assert isinstance(payload["auth"], int) | ||
| 1221 | + assert isinstance(payload["encryption"], int) | ||
| 1222 | + assert isinstance(payload["rssi"], int) | ||
| 1223 | + assert isinstance(payload["password"], str) | ||
| 1224 | + raise KasaException("timeout") | ||
| 1225 | + | ||
| 1226 | + dev.protocol.query = _query | ||
| 1227 | + resp = await dev.wifi_join("ModernWifi", "secret") | ||
| 1228 | + assert resp == {} | ||
| 1229 | + | ||
| 1230 | + | ||
| 1231 | + @pytest.mark.requires_dummy | ||
| 1232 | + async def test_wifi_join_connectap_device_error(dummy_protocol): | ||
| 1233 | + """Verify connectAp device errors are re-raised.""" | ||
| 1234 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1235 | + dev.protocol._transport._credentials = Credentials("user", "pass") | ||
| 1236 | + dev.networks = [ | ||
| 1237 | + WifiNetwork( | ||
| 1238 | + ssid="ModernWifi", | ||
| 1239 | + auth=3, | ||
| 1240 | + encryption=2, | ||
| 1241 | + rssi=2, | ||
| 1242 | + bssid="11:22:33:44:55:66", | ||
| 1243 | + ) | ||
| 1244 | + ] | ||
| 1245 | + | ||
| 1246 | + async def _query(*_, **__): | ||
| 1247 | + raise DeviceError("bad request", error_code=SmartErrorCode.PARAMS_ERROR) | ||
| 1248 | + | ||
| 1249 | + dev.protocol.query = _query | ||
| 1250 | + with pytest.raises(DeviceError): | ||
| 1251 | + await dev.wifi_join("ModernWifi", "secret") | ||
| 1252 | + | ||
| 1253 | + | ||
| 1254 | + @pytest.mark.requires_dummy | ||
| 1255 | + async def test_wifi_join_connectap_bad_public_key(dummy_protocol, monkeypatch): | ||
| 1256 | + """Verify connectAp raises on bad public key data.""" | ||
| 1257 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1258 | + dev.protocol._transport._credentials = Credentials("user", "pass") | ||
| 1259 | + dev.public_key = "not-base64" | ||
| 1260 | + dev.networks = [ | ||
| 1261 | + WifiNetwork( | ||
| 1262 | + ssid="ModernWifi", | ||
| 1263 | + auth=3, | ||
| 1264 | + encryption=2, | ||
| 1265 | + rssi=2, | ||
| 1266 | + bssid="11:22:33:44:55:66", | ||
| 1267 | + ) | ||
| 1268 | + ] | ||
| 1269 | + | ||
| 1270 | + def _raise_kasa(*_, **__): | ||
| 1271 | + raise KasaException("bad public key") | ||
| 1272 | + | ||
| 1273 | + monkeypatch.setattr("base64.b64decode", _raise_kasa) | ||
| 1274 | + | ||
| 1275 | + with pytest.raises(KasaException, match="bad public key"): | ||
| 1276 | + await dev.wifi_join("ModernWifi", "secret") | ||
| 1277 | + | ||
| 1278 | + | ||
| 1279 | + @pytest.mark.requires_dummy | ||
| 1280 | + async def test_wifi_join_connectap_non_rsa_key(dummy_protocol, monkeypatch): | ||
| 1281 | + """Verify connectAp rejects non-RSA public keys.""" | ||
| 1282 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1283 | + dev.protocol._transport._credentials = Credentials("user", "pass") | ||
| 1284 | + dev.public_key = dev.STATIC_PUBLIC_KEY_B64 | ||
| 1285 | + dev.networks = [ | ||
| 1286 | + WifiNetwork( | ||
| 1287 | + ssid="ModernWifi", | ||
| 1288 | + auth=3, | ||
| 1289 | + encryption=2, | ||
| 1290 | + rssi=2, | ||
| 1291 | + bssid="11:22:33:44:55:66", | ||
| 1292 | + ) | ||
| 1293 | + ] | ||
| 1294 | + | ||
| 1295 | + def _fake_load_der_public_key(_): | ||
| 1296 | + return object() | ||
| 1297 | + | ||
| 1298 | + monkeypatch.setattr( | ||
| 1299 | + "kasa.smart.smartdevice.serialization.load_der_public_key", | ||
| 1300 | + _fake_load_der_public_key, | ||
| 1301 | + ) | ||
| 1302 | + | ||
| 1303 | + with pytest.raises(TypeError, match="RSA public key"): | ||
| 1304 | + await dev.wifi_join("ModernWifi", "secret") | ||
| 1305 | + | ||
| 1306 | + | ||
| 1307 | + @pytest.mark.requires_dummy | ||
| 1308 | + async def test_wifi_join_legacy_set_qs_info(dummy_protocol): | ||
| 1309 | + """Verify legacy onboarding payload and timeout handling.""" | ||
| 1310 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1311 | + dev.protocol._transport._credentials = Credentials("user", "pass") | ||
| 1312 | + dev._last_update = {"get_device_time": {"year": 2024}} | ||
| 1313 | + dev.networks = [WifiNetwork(ssid="LegacyWifi", bssid="aa:bb:cc:dd:ee:ff")] | ||
| 1314 | + | ||
| 1315 | + async def _query(request, retry_count=3, *_, **__): | ||
| 1316 | + assert retry_count == 0 | ||
| 1317 | + payload = request["set_qs_info"] | ||
| 1318 | + assert payload["account"]["username"] == base64.b64encode(b"user").decode() | ||
| 1319 | + assert payload["account"]["password"] == base64.b64encode(b"pass").decode() | ||
| 1320 | + assert payload["wireless"]["ssid"] == base64.b64encode(b"LegacyWifi").decode() | ||
| 1321 | + raise KasaException("timeout") | ||
| 1322 | + | ||
| 1323 | + dev.protocol.query = _query | ||
| 1324 | + resp = await dev.wifi_join("LegacyWifi", "secret") | ||
| 1325 | + assert resp == {} | ||
| 1326 | + | ||
| 1327 | + | ||
| 1328 | + @pytest.mark.requires_dummy | ||
| 1329 | + async def test_wifi_join_no_credentials(dummy_protocol): | ||
| 1330 | + """Verify wifi_join raises without credentials.""" | ||
| 1331 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1332 | + with pytest.raises(AuthenticationError): | ||
| 1333 | + await dev.wifi_join("ssid", "password") | ||
| 1334 | + | ||
| 1335 | + | ||
| 1336 | + @pytest.mark.requires_dummy | ||
| 1337 | + async def test_wifi_join_no_networks(dummy_protocol): | ||
| 1338 | + """Verify wifi_join raises when scan returns no networks.""" | ||
| 1339 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1340 | + dev.protocol._transport._credentials = Credentials("user", "pass") | ||
| 1341 | + | ||
| 1342 | + async def _wifi_scan(): | ||
| 1343 | + return [] | ||
| 1344 | + | ||
| 1345 | + dev.wifi_scan = _wifi_scan | ||
| 1346 | + with pytest.raises(KasaException, match="Unable to determine wifi scan type"): | ||
| 1347 | + await dev.wifi_join("ssid", "password") | ||
| 1348 | + | ||
| 1349 | + | ||
| 1350 | + @pytest.mark.requires_dummy | ||
| 1351 | + async def test_wifi_join_ssid_not_found(dummy_protocol): | ||
| 1352 | + """Verify wifi_join raises when ssid is not found in scan results.""" | ||
| 1353 | + dev = SmartDevice("127.0.0.1", protocol=dummy_protocol) | ||
| 1354 | + dev.protocol._transport._credentials = Credentials("user", "pass") | ||
| 1355 | + dev.networks = [ | ||
| 1356 | + WifiNetwork( | ||
| 1357 | + ssid="OtherWifi", | ||
| 1358 | + auth=3, | ||
| 1359 | + encryption=2, | ||
| 1360 | + rssi=2, | ||
| 1361 | + bssid="11:22:33:44:55:66", | ||
| 1362 | + ) | ||
| 1363 | + ] | ||
| 1364 | + | ||
| 1365 | + with pytest.raises(KasaException, match="Unable to determine wifi scan type"): | ||
| 1366 | + await dev.wifi_join("MissingWifi", "secret") | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments