From 077041e0720f8777ce0d386d4d340d2a2ebc06bc Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 12 Aug 2026 20:45:06 -0500 Subject: [PATCH] Deduplicate resolved endpoints before sizing the attempt budget --- esphome/espota2.py | 12 ++++++++++++ tests/unit_tests/test_espota2.py | 19 +++++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/esphome/espota2.py b/esphome/espota2.py index 61e897f601..b7db9dd4a7 100644 --- a/esphome/espota2.py +++ b/esphome/espota2.py @@ -580,6 +580,18 @@ def run_ota_impl_( _LOGGER.error("No addresses to connect to for %s", remote_host) return 1, None + # The same device often resolves through several names (mDNS name, + # use_address, MQTT discovery), so drop duplicate endpoints; they add no + # new path to the device but would inflate the attempt budget below. + seen_endpoints: set[tuple[int, tuple]] = set() + unique_res = [] + for r in res: + endpoint = (r[0], r[4]) + if endpoint not in seen_endpoints: + seen_endpoints.add(endpoint) + unique_res.append(r) + res = unique_res + # Every address is tried at least once and EXTRA_UPLOAD_ATTEMPTS retries # are shared across the addresses, cycling through them. Wait before an # attempt when the previous one actually reached the device, or when diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py index db4a4b1117..91882c4950 100644 --- a/tests/unit_tests/test_espota2.py +++ b/tests/unit_tests/test_espota2.py @@ -851,6 +851,25 @@ def test_run_ota_impl_multiple_addresses_cycle( assert mock_sleep.call_count == 2 +@pytest.mark.usefixtures("mock_socket_constructor") +def test_run_ota_impl_duplicate_addresses_deduplicated( + mock_socket: Mock, firmware_file: Path, mock_resolve_ip: Mock, mock_sleep: Mock +) -> None: + """Test duplicate resolved endpoints do not inflate the attempt budget.""" + entry = (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.168.1.100", 3232)) + mock_resolve_ip.return_value = [entry, entry] + mock_socket.connect.side_effect = OSError("Connection refused") + + result_code, result_host = espota2.run_ota_impl_( + "test.local", 3232, "password", str(firmware_file) + ) + + assert result_code == 1 + assert result_host is None + # The duplicate collapses to one endpoint, so the budget is 1 + EXTRA + assert mock_socket.connect.call_count == espota2.EXTRA_UPLOAD_ATTEMPTS + 1 + + @pytest.mark.usefixtures("mock_socket_constructor", "mock_resolve_ip_dual") def test_run_ota_impl_second_address_succeeds_without_delay( mock_socket: Mock,