Guard against empty address list and polish review nits

This commit is contained in:
J. Nick Koston
2026-08-12 19:21:13 -05:00
parent 3ee8aaf77c
commit c1481bbb5d
2 changed files with 26 additions and 5 deletions
+8 -3
View File
@@ -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, ...);
+18 -2
View File
@@ -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