python-kasa · rytilahti · Jun 27, 2024 · Jun 26, 2024 · Jun 26, 2024
diff --git a/kasa/aestransport.py b/kasa/aestransport.py
@@ -140,7 +140,13 @@ def hash_credentials(login_v2: bool, credentials: Credentials) -> tuple[str, str
         return un, pw
 
     def _handle_response_error_code(self, resp_dict: Any, msg: str) -> None:
-        error_code = SmartErrorCode(resp_dict.get("error_code"))  # type: ignore[arg-type]
+        error_code_raw = resp_dict.get("error_code")
+        try:
+            error_code = SmartErrorCode(error_code_raw)  # type: ignore[arg-type]
+        except ValueError:
+            _LOGGER.warning("Received unknown error code: %s", error_code_raw)
+            error_code = SmartErrorCode.INTERNAL_UNKNOWN_ERROR
+
         if error_code == SmartErrorCode.SUCCESS:
             return
         msg = f"{msg}: {self._host}: {error_code.name}({error_code.value})"

diff --git a/kasa/exceptions.py b/kasa/exceptions.py
@@ -119,6 +119,9 @@ def __str__(self):
     DST_ERROR = -2301
     DST_SAVE_ERROR = -2302
 
+    # Library internal for unknown error codes
+    INTERNAL_UNKNOWN_ERROR = -100_000
+
 
 SMART_RETRYABLE_ERRORS = [
     SmartErrorCode.TRANSPORT_NOT_AVAILABLE_ERROR,

diff --git a/kasa/smartprotocol.py b/kasa/smartprotocol.py
@@ -239,12 +239,20 @@ async def _handle_response_lists(
             response_result[response_list_name].extend(next_batch[response_list_name])
 
     def _handle_response_error_code(self, resp_dict: dict, method, raise_on_error=True):
-        error_code = SmartErrorCode(resp_dict.get("error_code"))  # type: ignore[arg-type]
+        error_code_raw = resp_dict.get("error_code")
+        try:
+            error_code = SmartErrorCode(error_code_raw)  # type: ignore[arg-type]
+        except ValueError:
+            _LOGGER.warning("Received unknown error code: %s", error_code_raw)
+            error_code = SmartErrorCode.INTERNAL_UNKNOWN_ERROR
+
         if error_code == SmartErrorCode.SUCCESS:
             return
+
         if not raise_on_error:
             resp_dict["result"] = error_code
             return
+
         msg = (
             f"Error querying device: {self._host}: "
             + f"{error_code.name}({error_code.value})"

diff --git a/kasa/tests/test_aestransport.py b/kasa/tests/test_aestransport.py
@@ -276,6 +276,33 @@ async def test_passthrough_errors(mocker, error_code):
         await transport.send(json_dumps(request))
 
 
+@pytest.mark.parametrize("error_code", [-13333, 13333])
+async def test_unknown_errors(mocker, error_code):
+    host = "127.0.0.1"
+    mock_aes_device = MockAesDevice(host, 200, error_code, 0)
+    mocker.patch.object(aiohttp.ClientSession, "post", side_effect=mock_aes_device.post)
+
+    config = DeviceConfig(host, credentials=Credentials("foo", "bar"))
+    transport = AesTransport(config=config)
+    transport._handshake_done = True
+    transport._session_expire_at = time.time() + 86400
+    transport._encryption_session = mock_aes_device.encryption_session
+    transport._token_url = transport._app_url.with_query(
+        f"token={mock_aes_device.token}"
+    )
+
+    request = {
+        "method": "get_device_info",
+        "params": None,
+        "request_time_milis": round(time.time() * 1000),
+        "requestID": 1,
+        "terminal_uuid": "foobar",
+    }
+    with pytest.raises(KasaException):
+        res = await transport.send(json_dumps(request))
+        assert res is SmartErrorCode.INTERNAL_UNKNOWN_ERROR
+
+
 async def test_port_override():
     """Test that port override sets the app_url."""
     host = "127.0.0.1"

diff --git a/kasa/tests/test_smartprotocol.py b/kasa/tests/test_smartprotocol.py
@@ -35,6 +35,25 @@ async def test_smart_device_errors(dummy_protocol, mocker, error_code):
     assert send_mock.call_count == expected_calls
 
 
+@pytest.mark.parametrize("error_code", [-13333, 13333])
+async def test_smart_device_unknown_errors(
+    dummy_protocol, mocker, error_code, caplog: pytest.LogCaptureFixture
+):
+    """Test handling of unknown error codes."""
+    mock_response = {"result": {"great": "success"}, "error_code": error_code}
+
+    send_mock = mocker.patch.object(
+        dummy_protocol._transport, "send", return_value=mock_response
+    )
+
+    with pytest.raises(KasaException):
+        res = await dummy_protocol.query(DUMMY_QUERY)
+        assert res is SmartErrorCode.INTERNAL_UNKNOWN_ERROR
+
+    send_mock.assert_called_once()
+    assert f"Received unknown error code: {error_code}" in caplog.text
+
+
 @pytest.mark.parametrize("error_code", ERRORS, ids=lambda e: e.name)
 async def test_smart_device_errors_in_multiple_request(
     dummy_protocol, mocker, error_code