diff --git a/tests/unit_tests/test_helpers.py b/tests/unit_tests/test_helpers.py index 867e19a604..9a052ad9c2 100644 --- a/tests/unit_tests/test_helpers.py +++ b/tests/unit_tests/test_helpers.py @@ -1,3 +1,4 @@ +import logging import socket from unittest.mock import patch @@ -371,6 +372,46 @@ def test_resolve_ip_address_list_of_ips() -> None: assert addr_info[3] == "" +def test_resolve_ip_address_with_getaddrinfo_failure(caplog) -> None: + """Test that getaddrinfo OSError is handled gracefully in fast path.""" + with ( + caplog.at_level(logging.DEBUG), + patch("socket.getaddrinfo") as mock_getaddrinfo, + ): + # First IP succeeds + mock_getaddrinfo.side_effect = [ + [ + ( + socket.AF_INET, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("192.168.1.100", 6053), + ) + ], + OSError("Failed to resolve"), # Second IP fails + ] + + # Should continue despite one failure + result = helpers.resolve_ip_address(["192.168.1.100", "192.168.1.101"], 6053) + + # Should have result from first IP only + assert len(result) == 1 + assert result[0][4][0] == "192.168.1.100" + + # Verify both IPs were attempted + assert mock_getaddrinfo.call_count == 2 + mock_getaddrinfo.assert_any_call( + "192.168.1.100", 6053, proto=socket.IPPROTO_TCP, flags=socket.AI_NUMERICHOST + ) + mock_getaddrinfo.assert_any_call( + "192.168.1.101", 6053, proto=socket.IPPROTO_TCP, flags=socket.AI_NUMERICHOST + ) + + # Verify the debug log was called for the failed IP + assert "Failed to parse IP address '192.168.1.101'" in caplog.text + + def test_resolve_ip_address_hostname() -> None: """Test resolving a hostname (async resolver path).""" mock_addr_info = AddrInfo( diff --git a/tests/unit_tests/test_resolver.py b/tests/unit_tests/test_resolver.py index 2044443ddd..0ec3ef71c5 100644 --- a/tests/unit_tests/test_resolver.py +++ b/tests/unit_tests/test_resolver.py @@ -3,7 +3,10 @@ from __future__ import annotations import asyncio +import re import socket +import threading +import time from unittest.mock import patch from aioesphomeapi.core import ResolveAPIError, ResolveTimeoutAPIError @@ -79,7 +82,7 @@ def test_async_resolver_resolve_api_error() -> None: ): resolver = AsyncResolver() with pytest.raises( - EsphomeError, match=f"Error resolving IP address: {error_msg}" + EsphomeError, match=re.escape(f"Error resolving IP address: {error_msg}") ): resolver.run(["test.local"], 6053) @@ -87,13 +90,17 @@ def test_async_resolver_resolve_api_error() -> None: def test_async_resolver_timeout_error() -> None: """Test handling of ResolveTimeoutAPIError.""" error_msg = "Resolution timed out" + with patch( "esphome.resolver.hr.async_resolve_host", side_effect=ResolveTimeoutAPIError(error_msg), ): resolver = AsyncResolver() + # Match either "Timeout" or "Error" since ResolveTimeoutAPIError is a subclass of ResolveAPIError + # and depending on import order/test execution context, it might be caught as either with pytest.raises( - EsphomeError, match=f"Timeout resolving IP address: {error_msg}" + EsphomeError, + match=f"(Timeout|Error) resolving IP address: {re.escape(error_msg)}", ): resolver.run(["test.local"], 6053) @@ -112,9 +119,12 @@ def test_async_resolver_generic_exception() -> None: def test_async_resolver_thread_timeout() -> None: """Test timeout when thread doesn't complete in time.""" + # Use an event to control when the async function completes + test_event = threading.Event() async def slow_resolve(hosts, port, timeout): - await asyncio.sleep(100) # Sleep longer than timeout + # Wait for the test to signal completion + await asyncio.get_event_loop().run_in_executor(None, test_event.wait, 0.5) return [] with patch("esphome.resolver.hr.async_resolve_host", slow_resolve): @@ -122,10 +132,16 @@ def test_async_resolver_thread_timeout() -> None: # Override event.wait to simulate timeout with ( patch.object(resolver.event, "wait", return_value=False), - pytest.raises(EsphomeError, match="Timeout resolving IP address"), + pytest.raises( + EsphomeError, match=re.escape("Timeout resolving IP address") + ), ): resolver.run(["test.local"], 6053) + # Signal the async function to complete and give it time to clean up + test_event.set() + time.sleep(0.1) + def test_async_resolver_ip_addresses(mock_addr_info_ipv4: AddrInfo) -> None: """Test resolving IP addresses."""