diff --git a/esphome/espota2.py b/esphome/espota2.py index 2dd5409452..61e897f601 100644 --- a/esphome/espota2.py +++ b/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 diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py index dc9c586788..db4a4b1117 100644 --- a/tests/unit_tests/test_espota2.py +++ b/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