Keep MD5 mismatch non-retryable with its own error message

esphome/espota2.py

@@ -174,18 +174,13 @@ _ERROR_MESSAGES: dict[int, str] = {
     RESPONSE_ERROR_UNKNOWN: "Unknown error from ESP",
 }
 
-# Device-reported errors that do not persist across attempts: an MD5 mismatch
-# means the transfer arrived corrupted and the device aborted without
-# committing, so a fresh upload may succeed.
-_RETRYABLE_ERROR_CODES: frozenset[int] = frozenset({RESPONSE_ERROR_MD5_MISMATCH})
-
 
 class OTAError(EsphomeError):
     pass
 
 
 class OTANetworkError(OTAError):
-    """Transient OTA failure (timeout, reset, closed connection, corrupted transfer); retrying may succeed."""
+    """Network-level OTA failure (timeout, reset, closed connection); retrying may succeed."""
 
 
 def _committed_error(err: OTANetworkError) -> OTAError:
@@ -278,8 +273,6 @@ def check_error(data: list[int] | bytes, expect: int | list[int] | None) -> None
     dat = data[0]
     error_msg = _ERROR_MESSAGES.get(dat)
     if error_msg is not None:
-        if dat in _RETRYABLE_ERROR_CODES:
-            raise OTANetworkError(error_msg)
         raise OTAError(error_msg)
     if expect is None:
         return

tests/unit_tests/test_espota2.py

@@ -274,12 +274,6 @@ def test_check_error_unexpected_response() -> None:
         espota2.check_error([0x7F], [espota2.RESPONSE_OK, espota2.RESPONSE_AUTH_OK])
 
 
-def test_check_error_md5_mismatch_is_retryable() -> None:
-    """Test check_error raises the retryable OTANetworkError for an MD5 mismatch."""
-    with pytest.raises(espota2.OTANetworkError, match="MD5 code mismatch"):
-        espota2.check_error([espota2.RESPONSE_ERROR_MD5_MISMATCH], None)
-
-
 def test_check_error_empty_data() -> None:
     """Test check_error raises the retryable OTANetworkError when the device closes the connection."""
     with pytest.raises(
@@ -682,6 +676,26 @@ def test_perform_ota_post_commit_failure_not_retryable(
     assert not isinstance(exc.value, espota2.OTANetworkError)
 
 
+@pytest.mark.usefixtures("mock_time")
+def test_perform_ota_md5_mismatch_not_marked_committed(
+    mock_socket: Mock, mock_file: io.BytesIO
+) -> None:
+    """Test an MD5 mismatch keeps its own message and stays non-retryable."""
+    mock_socket.recv.side_effect = [
+        *_no_auth_handshake(espota2.OTA_VERSION_1_0),
+        bytes([espota2.RESPONSE_RECEIVE_OK]),  # Device received everything
+        bytes([espota2.RESPONSE_ERROR_MD5_MISMATCH]),  # Device aborted the update
+    ]
+
+    with pytest.raises(espota2.OTAError, match="MD5 code mismatch") as exc:
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
+
+    # The device aborted without committing, so the message must not claim
+    # the update may have been installed, and the error must not be retried
+    assert not isinstance(exc.value, espota2.OTANetworkError)
+    assert "committed" not in str(exc.value)
+
+
 @pytest.mark.usefixtures("mock_time")
 def test_perform_ota_end_ack_send_failure_is_success(
     mock_socket: Mock, mock_file: io.BytesIO