From a282920d7c9fd79b287de7eee3662dd7c2841629 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Thu, 4 Sep 2025 20:06:55 -0500 Subject: [PATCH] fix, cover --- esphome/resolver.py | 4 +- tests/unit_tests/test_helpers.py | 255 ++++++++++++++++++++++++++++++ tests/unit_tests/test_resolver.py | 157 ++++++++++++++++++ 3 files changed, 414 insertions(+), 2 deletions(-) create mode 100644 tests/unit_tests/test_resolver.py diff --git a/esphome/resolver.py b/esphome/resolver.py index a245737962..f70ecec357 100644 --- a/esphome/resolver.py +++ b/esphome/resolver.py @@ -52,10 +52,10 @@ class AsyncResolver: raise EsphomeError("Timeout resolving IP address") if exc := self.exception: - if isinstance(exc, ResolveAPIError): - raise EsphomeError(f"Error resolving IP address: {exc}") from exc if isinstance(exc, ResolveTimeoutAPIError): raise EsphomeError(f"Timeout resolving IP address: {exc}") from exc + if isinstance(exc, ResolveAPIError): + raise EsphomeError(f"Error resolving IP address: {exc}") from exc raise exc return self.result diff --git a/tests/unit_tests/test_helpers.py b/tests/unit_tests/test_helpers.py index b353d1aa99..706acdd359 100644 --- a/tests/unit_tests/test_helpers.py +++ b/tests/unit_tests/test_helpers.py @@ -1,8 +1,13 @@ +import socket +from unittest.mock import patch + +from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr from hypothesis import given from hypothesis.strategies import ip_addresses import pytest from esphome import helpers +from esphome.core import EsphomeError @pytest.mark.parametrize( @@ -277,3 +282,253 @@ def test_sort_ip_addresses(text: list[str], expected: list[str]) -> None: actual = helpers.sort_ip_addresses(text) assert actual == expected + + +# DNS resolution tests +def test_is_ip_address_ipv4(): + """Test is_ip_address with IPv4 addresses.""" + assert helpers.is_ip_address("192.168.1.1") is True + assert helpers.is_ip_address("127.0.0.1") is True + assert helpers.is_ip_address("255.255.255.255") is True + assert helpers.is_ip_address("0.0.0.0") is True + + +def test_is_ip_address_ipv6(): + """Test is_ip_address with IPv6 addresses.""" + assert helpers.is_ip_address("::1") is True + assert helpers.is_ip_address("2001:db8::1") is True + assert helpers.is_ip_address("fe80::1") is True + assert helpers.is_ip_address("::") is True + + +def test_is_ip_address_invalid(): + """Test is_ip_address with non-IP strings.""" + assert helpers.is_ip_address("hostname") is False + assert helpers.is_ip_address("hostname.local") is False + assert helpers.is_ip_address("256.256.256.256") is False + assert helpers.is_ip_address("192.168.1") is False + assert helpers.is_ip_address("") is False + + +def test_resolve_ip_address_single_ipv4(): + """Test resolving a single IPv4 address (fast path).""" + result = helpers.resolve_ip_address("192.168.1.100", 6053) + + assert len(result) == 1 + assert result[0][0] == socket.AF_INET # family + assert result[0][1] == socket.SOCK_STREAM # type + assert result[0][2] == socket.IPPROTO_TCP # proto + assert result[0][3] == "" # canonname + assert result[0][4] == ("192.168.1.100", 6053) # sockaddr + + +def test_resolve_ip_address_single_ipv6(): + """Test resolving a single IPv6 address (fast path).""" + result = helpers.resolve_ip_address("::1", 6053) + + assert len(result) == 1 + assert result[0][0] == socket.AF_INET6 # family + assert result[0][1] == socket.SOCK_STREAM # type + assert result[0][2] == socket.IPPROTO_TCP # proto + assert result[0][3] == "" # canonname + # IPv6 sockaddr has 4 elements + assert len(result[0][4]) == 4 + assert result[0][4][0] == "::1" # address + assert result[0][4][1] == 6053 # port + + +def test_resolve_ip_address_list_of_ips(): + """Test resolving a list of IP addresses (fast path).""" + ips = ["192.168.1.100", "10.0.0.1", "::1"] + result = helpers.resolve_ip_address(ips, 6053) + + # Should return results sorted by preference (IPv6 first, then IPv4) + assert len(result) >= 2 # At least IPv4 addresses should work + + # Check that results are properly formatted + for addr_info in result: + assert addr_info[0] in (socket.AF_INET, socket.AF_INET6) + assert addr_info[1] == socket.SOCK_STREAM + assert addr_info[2] == socket.IPPROTO_TCP + assert addr_info[3] == "" + + +def test_resolve_ip_address_hostname(): + """Test resolving a hostname (async resolver path).""" + mock_addr_info = AddrInfo( + family=socket.AF_INET, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + sockaddr=IPv4Sockaddr(address="192.168.1.100", port=6053), + ) + + with patch("esphome.resolver.AsyncResolver") as MockResolver: + mock_resolver = MockResolver.return_value + mock_resolver.run.return_value = [mock_addr_info] + + result = helpers.resolve_ip_address("test.local", 6053) + + assert len(result) == 1 + assert result[0][0] == socket.AF_INET + assert result[0][4] == ("192.168.1.100", 6053) + mock_resolver.run.assert_called_once_with(["test.local"], 6053) + + +def test_resolve_ip_address_mixed_list(): + """Test resolving a mix of IPs and hostnames.""" + mock_addr_info = AddrInfo( + family=socket.AF_INET, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + sockaddr=IPv4Sockaddr(address="192.168.1.200", port=6053), + ) + + with patch("esphome.resolver.AsyncResolver") as MockResolver: + mock_resolver = MockResolver.return_value + mock_resolver.run.return_value = [mock_addr_info] + + # Mix of IP and hostname - should use async resolver + result = helpers.resolve_ip_address(["192.168.1.100", "test.local"], 6053) + + assert len(result) == 1 + assert result[0][4][0] == "192.168.1.200" + mock_resolver.run.assert_called_once_with(["192.168.1.100", "test.local"], 6053) + + +def test_resolve_ip_address_url(): + """Test extracting hostname from URL.""" + mock_addr_info = AddrInfo( + family=socket.AF_INET, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + sockaddr=IPv4Sockaddr(address="192.168.1.100", port=6053), + ) + + with patch("esphome.resolver.AsyncResolver") as MockResolver: + mock_resolver = MockResolver.return_value + mock_resolver.run.return_value = [mock_addr_info] + + result = helpers.resolve_ip_address("http://test.local", 6053) + + assert len(result) == 1 + mock_resolver.run.assert_called_once_with(["test.local"], 6053) + + +def test_resolve_ip_address_ipv6_conversion(): + """Test proper IPv6 address info conversion.""" + mock_addr_info = AddrInfo( + family=socket.AF_INET6, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + sockaddr=IPv6Sockaddr(address="2001:db8::1", port=6053, flowinfo=1, scope_id=2), + ) + + with patch("esphome.resolver.AsyncResolver") as MockResolver: + mock_resolver = MockResolver.return_value + mock_resolver.run.return_value = [mock_addr_info] + + result = helpers.resolve_ip_address("test.local", 6053) + + assert len(result) == 1 + assert result[0][0] == socket.AF_INET6 + assert result[0][4] == ("2001:db8::1", 6053, 1, 2) + + +def test_resolve_ip_address_error_handling(): + """Test error handling from AsyncResolver.""" + with patch("esphome.resolver.AsyncResolver") as MockResolver: + mock_resolver = MockResolver.return_value + mock_resolver.run.side_effect = EsphomeError("Resolution failed") + + with pytest.raises(EsphomeError, match="Resolution failed"): + helpers.resolve_ip_address("test.local", 6053) + + +def test_addr_preference_ipv4(): + """Test address preference for IPv4.""" + addr_info = ( + socket.AF_INET, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("192.168.1.1", 6053), + ) + assert helpers.addr_preference_(addr_info) == 2 + + +def test_addr_preference_ipv6(): + """Test address preference for regular IPv6.""" + addr_info = ( + socket.AF_INET6, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("2001:db8::1", 6053, 0, 0), + ) + assert helpers.addr_preference_(addr_info) == 1 + + +def test_addr_preference_ipv6_link_local_no_scope(): + """Test address preference for link-local IPv6 without scope.""" + addr_info = ( + socket.AF_INET6, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("fe80::1", 6053, 0, 0), # link-local with scope_id=0 + ) + assert helpers.addr_preference_(addr_info) == 3 + + +def test_addr_preference_ipv6_link_local_with_scope(): + """Test address preference for link-local IPv6 with scope.""" + addr_info = ( + socket.AF_INET6, + socket.SOCK_STREAM, + socket.IPPROTO_TCP, + "", + ("fe80::1", 6053, 0, 2), # link-local with scope_id=2 + ) + assert helpers.addr_preference_(addr_info) == 1 # Has scope, so it's usable + + +def test_resolve_ip_address_sorting(): + """Test that results are sorted by preference.""" + # Create multiple address infos with different preferences + mock_addr_infos = [ + AddrInfo( + family=socket.AF_INET6, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + sockaddr=IPv6Sockaddr( + address="fe80::1", port=6053, flowinfo=0, scope_id=0 + ), # Preference 3 (link-local no scope) + ), + AddrInfo( + family=socket.AF_INET, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + sockaddr=IPv4Sockaddr( + address="192.168.1.100", port=6053 + ), # Preference 2 (IPv4) + ), + AddrInfo( + family=socket.AF_INET6, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + sockaddr=IPv6Sockaddr( + address="2001:db8::1", port=6053, flowinfo=0, scope_id=0 + ), # Preference 1 (IPv6) + ), + ] + + with patch("esphome.resolver.AsyncResolver") as MockResolver: + mock_resolver = MockResolver.return_value + mock_resolver.run.return_value = mock_addr_infos + + result = helpers.resolve_ip_address("test.local", 6053) + + # Should be sorted: IPv6 first, then IPv4, then link-local without scope + assert result[0][4][0] == "2001:db8::1" # IPv6 (preference 1) + assert result[1][4][0] == "192.168.1.100" # IPv4 (preference 2) + assert result[2][4][0] == "fe80::1" # Link-local no scope (preference 3) diff --git a/tests/unit_tests/test_resolver.py b/tests/unit_tests/test_resolver.py new file mode 100644 index 0000000000..d49a367085 --- /dev/null +++ b/tests/unit_tests/test_resolver.py @@ -0,0 +1,157 @@ +"""Tests for the DNS resolver module.""" + +from __future__ import annotations + +import asyncio +import socket +from unittest.mock import patch + +from aioesphomeapi.core import ResolveAPIError, ResolveTimeoutAPIError +from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr +import pytest + +from esphome.core import EsphomeError +from esphome.resolver import RESOLVE_TIMEOUT, AsyncResolver + + +@pytest.fixture +def mock_addr_info_ipv4(): + """Create a mock IPv4 AddrInfo.""" + return AddrInfo( + family=socket.AF_INET, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + sockaddr=IPv4Sockaddr(address="192.168.1.100", port=6053), + ) + + +@pytest.fixture +def mock_addr_info_ipv6(): + """Create a mock IPv6 AddrInfo.""" + return AddrInfo( + family=socket.AF_INET6, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + sockaddr=IPv6Sockaddr(address="2001:db8::1", port=6053, flowinfo=0, scope_id=0), + ) + + +def test_async_resolver_successful_resolution(mock_addr_info_ipv4): + """Test successful DNS resolution.""" + with patch( + "esphome.resolver.hr.async_resolve_host", + return_value=[mock_addr_info_ipv4], + ) as mock_resolve: + resolver = AsyncResolver() + result = resolver.run(["test.local"], 6053) + + assert result == [mock_addr_info_ipv4] + mock_resolve.assert_called_once_with( + ["test.local"], 6053, timeout=RESOLVE_TIMEOUT + ) + + +def test_async_resolver_multiple_hosts(mock_addr_info_ipv4, mock_addr_info_ipv6): + """Test resolving multiple hosts.""" + mock_results = [mock_addr_info_ipv4, mock_addr_info_ipv6] + + with patch( + "esphome.resolver.hr.async_resolve_host", + return_value=mock_results, + ) as mock_resolve: + resolver = AsyncResolver() + result = resolver.run(["test1.local", "test2.local"], 6053) + + assert result == mock_results + mock_resolve.assert_called_once_with( + ["test1.local", "test2.local"], 6053, timeout=RESOLVE_TIMEOUT + ) + + +def test_async_resolver_resolve_api_error(): + """Test handling of ResolveAPIError.""" + error_msg = "Failed to resolve" + with patch( + "esphome.resolver.hr.async_resolve_host", + side_effect=ResolveAPIError(error_msg), + ): + resolver = AsyncResolver() + with pytest.raises( + EsphomeError, match=f"Error resolving IP address: {error_msg}" + ): + resolver.run(["test.local"], 6053) + + +def test_async_resolver_timeout_error(): + """Test handling of ResolveTimeoutAPIError.""" + error_msg = "Resolution timed out" + with patch( + "esphome.resolver.hr.async_resolve_host", + side_effect=ResolveTimeoutAPIError(error_msg), + ): + resolver = AsyncResolver() + with pytest.raises( + EsphomeError, match=f"Timeout resolving IP address: {error_msg}" + ): + resolver.run(["test.local"], 6053) + + +def test_async_resolver_generic_exception(): + """Test handling of generic exceptions.""" + error = RuntimeError("Unexpected error") + with patch( + "esphome.resolver.hr.async_resolve_host", + side_effect=error, + ): + resolver = AsyncResolver() + with pytest.raises(RuntimeError, match="Unexpected error"): + resolver.run(["test.local"], 6053) + + +def test_async_resolver_thread_timeout(): + """Test timeout when thread doesn't complete in time.""" + + async def slow_resolve(hosts, port, timeout): + await asyncio.sleep(100) # Sleep longer than timeout + return [] + + with patch("esphome.resolver.hr.async_resolve_host", slow_resolve): + resolver = AsyncResolver() + # Override event.wait to simulate timeout + with ( + patch.object(resolver.event, "wait", return_value=False), + pytest.raises(EsphomeError, match="Timeout resolving IP address"), + ): + resolver.run(["test.local"], 6053) + + +def test_async_resolver_ip_addresses(mock_addr_info_ipv4): + """Test resolving IP addresses.""" + with patch( + "esphome.resolver.hr.async_resolve_host", + return_value=[mock_addr_info_ipv4], + ) as mock_resolve: + resolver = AsyncResolver() + result = resolver.run(["192.168.1.100"], 6053) + + assert result == [mock_addr_info_ipv4] + mock_resolve.assert_called_once_with( + ["192.168.1.100"], 6053, timeout=RESOLVE_TIMEOUT + ) + + +def test_async_resolver_mixed_addresses(mock_addr_info_ipv4, mock_addr_info_ipv6): + """Test resolving mix of hostnames and IP addresses.""" + mock_results = [mock_addr_info_ipv4, mock_addr_info_ipv6] + + with patch( + "esphome.resolver.hr.async_resolve_host", + return_value=mock_results, + ) as mock_resolve: + resolver = AsyncResolver() + result = resolver.run(["test.local", "192.168.1.100", "::1"], 6053) + + assert result == mock_results + mock_resolve.assert_called_once_with( + ["test.local", "192.168.1.100", "::1"], 6053, timeout=RESOLVE_TIMEOUT + )