diff --git a/esphome/espota2.py b/esphome/espota2.py index 9565cc3d31..5d0d28c22e 100644 --- a/esphome/espota2.py +++ b/esphome/espota2.py @@ -511,8 +511,9 @@ def perform_ota( # The device treats a missing end acknowledgement as non-fatal and is # already rebooting into the new firmware, so the update succeeded _LOGGER.warning("Failed sending end acknowledgement: %s", err) - - _LOGGER.info("OTA successful") + _LOGGER.info("OTA successful (end acknowledgement not delivered)") + else: + _LOGGER.info("OTA successful") # Do not connect logs until it is fully on time.sleep(1) @@ -546,6 +547,10 @@ def run_ota_impl_( ) raise OTAError(err) from err + if not res: + _LOGGER.error("No addresses to connect to for %s", remote_host) + return 1, None + # Every address is tried at least once and the budget grants # MAX_UPLOAD_ATTEMPTS - 1 extra retries, cycling through the addresses. # Wait before an attempt when the previous one actually reached the @@ -585,7 +590,7 @@ def run_ota_impl_( except OTANetworkError as err: # Transient network failure; retry last_error = str(err) - _LOGGER.warning(last_error) + _LOGGER.warning("%s", last_error) continue except OTAError as err: # Device-reported error (wrong password, wrong flash size, ...); diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py index b55758fa19..3848915778 100644 --- a/tests/unit_tests/test_espota2.py +++ b/tests/unit_tests/test_espota2.py @@ -578,7 +578,6 @@ def test_perform_ota_upload_error(mock_socket: Mock, mock_file: io.BytesIO) -> N espota2.perform_ota(mock_socket, None, mock_file, "test.bin") -@pytest.mark.usefixtures("mock_time") def _no_auth_handshake(version: int) -> list[bytes]: """Recv responses for a handshake without auth, up to the MD5 check.""" return [ @@ -591,6 +590,7 @@ def _no_auth_handshake(version: int) -> list[bytes]: ] +@pytest.mark.usefixtures("mock_time") def test_perform_ota_chunk_send_error(mock_socket: Mock, mock_file: io.BytesIO) -> None: """Test OTA raises the retryable OTANetworkError when sending a chunk fails.""" mock_socket.recv.side_effect = _no_auth_handshake(espota2.OTA_VERSION_2_0) @@ -762,7 +762,8 @@ def test_run_ota_impl_multiple_addresses_cycle( assert result_code == 1 assert result_host is None - # Every address gets a first visit plus MAX_UPLOAD_ATTEMPTS - 1 retries + # Each address is visited once, then the two spare attempts cycle back + # through them; the budget is shared, not per address assert mock_socket.connect.call_args_list == [ call(DUAL_STACK_SA6), call(DUAL_STACK_SA4), @@ -837,6 +838,21 @@ def test_run_ota_impl_device_error_not_retried( mock_sleep.assert_not_called() +def test_run_ota_impl_no_addresses( + firmware_file: Path, mock_resolve_ip: Mock, mock_sleep: Mock +) -> None: + """Test run_ota_impl_ fails cleanly when resolution yields no addresses.""" + mock_resolve_ip.return_value = [] + + result_code, result_host = espota2.run_ota_impl_( + "test.local", 3232, "password", str(firmware_file) + ) + + assert result_code == 1 + assert result_host is None + mock_sleep.assert_not_called() + + def test_run_ota_impl_resolve_failed(tmp_path: Path, mock_resolve_ip: Mock) -> None: """Test run_ota_impl_ when DNS resolution fails.""" # Create a real firmware file