diff --git a/tests/unit_tests/test_address_cache.py b/tests/unit_tests/test_address_cache.py index f02c129256..de43830d53 100644 --- a/tests/unit_tests/test_address_cache.py +++ b/tests/unit_tests/test_address_cache.py @@ -1,260 +1,305 @@ """Tests for the address_cache module.""" +from __future__ import annotations + +import logging + +import pytest +from pytest import LogCaptureFixture + from esphome.address_cache import AddressCache, normalize_hostname -class TestNormalizeHostname: - """Test the normalize_hostname function.""" - - def test_normalize_simple_hostname(self): - """Test normalizing a simple hostname.""" - assert normalize_hostname("device") == "device" - assert normalize_hostname("device.local") == "device.local" - assert normalize_hostname("server.example.com") == "server.example.com" - - def test_normalize_removes_trailing_dots(self): - """Test that trailing dots are removed.""" - assert normalize_hostname("device.") == "device" - assert normalize_hostname("device.local.") == "device.local" - assert normalize_hostname("server.example.com.") == "server.example.com" - assert normalize_hostname("device...") == "device" - - def test_normalize_converts_to_lowercase(self): - """Test that hostnames are converted to lowercase.""" - assert normalize_hostname("DEVICE") == "device" - assert normalize_hostname("Device.Local") == "device.local" - assert normalize_hostname("Server.Example.COM") == "server.example.com" - - def test_normalize_combined(self): - """Test combination of trailing dots and case conversion.""" - assert normalize_hostname("DEVICE.LOCAL.") == "device.local" - assert normalize_hostname("Server.Example.COM...") == "server.example.com" +def test_normalize_simple_hostname() -> None: + """Test normalizing a simple hostname.""" + assert normalize_hostname("device") == "device" + assert normalize_hostname("device.local") == "device.local" + assert normalize_hostname("server.example.com") == "server.example.com" -class TestAddressCache: - """Test the AddressCache class.""" +def test_normalize_removes_trailing_dots() -> None: + """Test that trailing dots are removed.""" + assert normalize_hostname("device.") == "device" + assert normalize_hostname("device.local.") == "device.local" + assert normalize_hostname("server.example.com.") == "server.example.com" + assert normalize_hostname("device...") == "device" - def test_init_empty(self): - """Test initialization with empty caches.""" - cache = AddressCache() - assert cache.mdns_cache == {} - assert cache.dns_cache == {} - assert not cache.has_cache() - def test_init_with_caches(self): - """Test initialization with provided caches.""" - mdns_cache = {"device.local": ["192.168.1.10"]} - dns_cache = {"server.com": ["10.0.0.1"]} - cache = AddressCache(mdns_cache=mdns_cache, dns_cache=dns_cache) - assert cache.mdns_cache == mdns_cache - assert cache.dns_cache == dns_cache - assert cache.has_cache() +def test_normalize_converts_to_lowercase() -> None: + """Test that hostnames are converted to lowercase.""" + assert normalize_hostname("DEVICE") == "device" + assert normalize_hostname("Device.Local") == "device.local" + assert normalize_hostname("Server.Example.COM") == "server.example.com" - def test_get_mdns_addresses(self): - """Test getting mDNS addresses.""" - cache = AddressCache( - mdns_cache={"device.local": ["192.168.1.10", "192.168.1.11"]} - ) - # Direct lookup - assert cache.get_mdns_addresses("device.local") == [ - "192.168.1.10", - "192.168.1.11", - ] +def test_normalize_combined() -> None: + """Test combination of trailing dots and case conversion.""" + assert normalize_hostname("DEVICE.LOCAL.") == "device.local" + assert normalize_hostname("Server.Example.COM...") == "server.example.com" - # Case insensitive lookup - assert cache.get_mdns_addresses("Device.Local") == [ - "192.168.1.10", - "192.168.1.11", - ] - # With trailing dot - assert cache.get_mdns_addresses("device.local.") == [ - "192.168.1.10", - "192.168.1.11", - ] +def test_init_empty() -> None: + """Test initialization with empty caches.""" + cache = AddressCache() + assert cache.mdns_cache == {} + assert cache.dns_cache == {} + assert not cache.has_cache() - # Not found - assert cache.get_mdns_addresses("unknown.local") is None - def test_get_dns_addresses(self): - """Test getting DNS addresses.""" - cache = AddressCache(dns_cache={"server.com": ["10.0.0.1", "10.0.0.2"]}) +def test_init_with_caches() -> None: + """Test initialization with provided caches.""" + mdns_cache: dict[str, list[str]] = {"device.local": ["192.168.1.10"]} + dns_cache: dict[str, list[str]] = {"server.com": ["10.0.0.1"]} + cache = AddressCache(mdns_cache=mdns_cache, dns_cache=dns_cache) + assert cache.mdns_cache == mdns_cache + assert cache.dns_cache == dns_cache + assert cache.has_cache() - # Direct lookup - assert cache.get_dns_addresses("server.com") == ["10.0.0.1", "10.0.0.2"] - # Case insensitive lookup - assert cache.get_dns_addresses("Server.COM") == ["10.0.0.1", "10.0.0.2"] +def test_get_mdns_addresses() -> None: + """Test getting mDNS addresses.""" + cache = AddressCache(mdns_cache={"device.local": ["192.168.1.10", "192.168.1.11"]}) - # With trailing dot - assert cache.get_dns_addresses("server.com.") == ["10.0.0.1", "10.0.0.2"] + # Direct lookup + assert cache.get_mdns_addresses("device.local") == [ + "192.168.1.10", + "192.168.1.11", + ] - # Not found - assert cache.get_dns_addresses("unknown.com") is None + # Case insensitive lookup + assert cache.get_mdns_addresses("Device.Local") == [ + "192.168.1.10", + "192.168.1.11", + ] - def test_get_addresses_auto_detection(self): - """Test automatic cache selection based on hostname.""" - cache = AddressCache( - mdns_cache={"device.local": ["192.168.1.10"]}, - dns_cache={"server.com": ["10.0.0.1"]}, - ) + # With trailing dot + assert cache.get_mdns_addresses("device.local.") == [ + "192.168.1.10", + "192.168.1.11", + ] - # Should use mDNS cache for .local domains - assert cache.get_addresses("device.local") == ["192.168.1.10"] - assert cache.get_addresses("device.local.") == ["192.168.1.10"] - assert cache.get_addresses("Device.Local") == ["192.168.1.10"] + # Not found + assert cache.get_mdns_addresses("unknown.local") is None - # Should use DNS cache for non-.local domains - assert cache.get_addresses("server.com") == ["10.0.0.1"] - assert cache.get_addresses("server.com.") == ["10.0.0.1"] - assert cache.get_addresses("Server.COM") == ["10.0.0.1"] - # Not found - assert cache.get_addresses("unknown.local") is None - assert cache.get_addresses("unknown.com") is None +def test_get_dns_addresses() -> None: + """Test getting DNS addresses.""" + cache = AddressCache(dns_cache={"server.com": ["10.0.0.1", "10.0.0.2"]}) - def test_has_cache(self): - """Test checking if cache has entries.""" - # Empty cache - cache = AddressCache() - assert not cache.has_cache() + # Direct lookup + assert cache.get_dns_addresses("server.com") == ["10.0.0.1", "10.0.0.2"] - # Only mDNS cache - cache = AddressCache(mdns_cache={"device.local": ["192.168.1.10"]}) - assert cache.has_cache() + # Case insensitive lookup + assert cache.get_dns_addresses("Server.COM") == ["10.0.0.1", "10.0.0.2"] - # Only DNS cache - cache = AddressCache(dns_cache={"server.com": ["10.0.0.1"]}) - assert cache.has_cache() + # With trailing dot + assert cache.get_dns_addresses("server.com.") == ["10.0.0.1", "10.0.0.2"] - # Both caches - cache = AddressCache( - mdns_cache={"device.local": ["192.168.1.10"]}, - dns_cache={"server.com": ["10.0.0.1"]}, - ) - assert cache.has_cache() + # Not found + assert cache.get_dns_addresses("unknown.com") is None - def test_from_cli_args_empty(self): - """Test creating cache from empty CLI arguments.""" - cache = AddressCache.from_cli_args([], []) - assert cache.mdns_cache == {} - assert cache.dns_cache == {} - def test_from_cli_args_single_entry(self): - """Test creating cache from single CLI argument.""" - mdns_args = ["device.local=192.168.1.10"] - dns_args = ["server.com=10.0.0.1"] +def test_get_addresses_auto_detection() -> None: + """Test automatic cache selection based on hostname.""" + cache = AddressCache( + mdns_cache={"device.local": ["192.168.1.10"]}, + dns_cache={"server.com": ["10.0.0.1"]}, + ) - cache = AddressCache.from_cli_args(mdns_args, dns_args) + # Should use mDNS cache for .local domains + assert cache.get_addresses("device.local") == ["192.168.1.10"] + assert cache.get_addresses("device.local.") == ["192.168.1.10"] + assert cache.get_addresses("Device.Local") == ["192.168.1.10"] - assert cache.mdns_cache == {"device.local": ["192.168.1.10"]} - assert cache.dns_cache == {"server.com": ["10.0.0.1"]} + # Should use DNS cache for non-.local domains + assert cache.get_addresses("server.com") == ["10.0.0.1"] + assert cache.get_addresses("server.com.") == ["10.0.0.1"] + assert cache.get_addresses("Server.COM") == ["10.0.0.1"] - def test_from_cli_args_multiple_ips(self): - """Test creating cache with multiple IPs per host.""" - mdns_args = ["device.local=192.168.1.10,192.168.1.11"] - dns_args = ["server.com=10.0.0.1,10.0.0.2,10.0.0.3"] + # Not found + assert cache.get_addresses("unknown.local") is None + assert cache.get_addresses("unknown.com") is None - cache = AddressCache.from_cli_args(mdns_args, dns_args) - assert cache.mdns_cache == {"device.local": ["192.168.1.10", "192.168.1.11"]} - assert cache.dns_cache == {"server.com": ["10.0.0.1", "10.0.0.2", "10.0.0.3"]} +def test_has_cache() -> None: + """Test checking if cache has entries.""" + # Empty cache + cache = AddressCache() + assert not cache.has_cache() - def test_from_cli_args_multiple_entries(self): - """Test creating cache with multiple host entries.""" - mdns_args = [ - "device1.local=192.168.1.10", - "device2.local=192.168.1.20,192.168.1.21", - ] - dns_args = ["server1.com=10.0.0.1", "server2.com=10.0.0.2"] + # Only mDNS cache + cache = AddressCache(mdns_cache={"device.local": ["192.168.1.10"]}) + assert cache.has_cache() - cache = AddressCache.from_cli_args(mdns_args, dns_args) + # Only DNS cache + cache = AddressCache(dns_cache={"server.com": ["10.0.0.1"]}) + assert cache.has_cache() - assert cache.mdns_cache == { - "device1.local": ["192.168.1.10"], - "device2.local": ["192.168.1.20", "192.168.1.21"], - } - assert cache.dns_cache == { - "server1.com": ["10.0.0.1"], - "server2.com": ["10.0.0.2"], - } + # Both caches + cache = AddressCache( + mdns_cache={"device.local": ["192.168.1.10"]}, + dns_cache={"server.com": ["10.0.0.1"]}, + ) + assert cache.has_cache() - def test_from_cli_args_normalization(self): - """Test that CLI arguments are normalized.""" - mdns_args = ["Device1.Local.=192.168.1.10", "DEVICE2.LOCAL=192.168.1.20"] - dns_args = ["Server1.COM.=10.0.0.1", "SERVER2.com=10.0.0.2"] - cache = AddressCache.from_cli_args(mdns_args, dns_args) +def test_from_cli_args_empty() -> None: + """Test creating cache from empty CLI arguments.""" + cache = AddressCache.from_cli_args([], []) + assert cache.mdns_cache == {} + assert cache.dns_cache == {} - # Hostnames should be normalized (lowercase, no trailing dots) - assert cache.mdns_cache == { - "device1.local": ["192.168.1.10"], - "device2.local": ["192.168.1.20"], - } - assert cache.dns_cache == { - "server1.com": ["10.0.0.1"], - "server2.com": ["10.0.0.2"], - } - def test_from_cli_args_whitespace_handling(self): - """Test that whitespace in IPs is handled.""" - mdns_args = ["device.local= 192.168.1.10 , 192.168.1.11 "] - dns_args = ["server.com= 10.0.0.1 , 10.0.0.2 "] +def test_from_cli_args_single_entry() -> None: + """Test creating cache from single CLI argument.""" + mdns_args: list[str] = ["device.local=192.168.1.10"] + dns_args: list[str] = ["server.com=10.0.0.1"] - cache = AddressCache.from_cli_args(mdns_args, dns_args) + cache = AddressCache.from_cli_args(mdns_args, dns_args) - assert cache.mdns_cache == {"device.local": ["192.168.1.10", "192.168.1.11"]} - assert cache.dns_cache == {"server.com": ["10.0.0.1", "10.0.0.2"]} + assert cache.mdns_cache == {"device.local": ["192.168.1.10"]} + assert cache.dns_cache == {"server.com": ["10.0.0.1"]} - def test_from_cli_args_invalid_format(self, caplog): - """Test handling of invalid argument format.""" - mdns_args = ["invalid_format", "device.local=192.168.1.10"] - dns_args = ["server.com=10.0.0.1", "also_invalid"] - cache = AddressCache.from_cli_args(mdns_args, dns_args) +def test_from_cli_args_multiple_ips() -> None: + """Test creating cache with multiple IPs per host.""" + mdns_args: list[str] = ["device.local=192.168.1.10,192.168.1.11"] + dns_args: list[str] = ["server.com=10.0.0.1,10.0.0.2,10.0.0.3"] - # Valid entries should still be processed - assert cache.mdns_cache == {"device.local": ["192.168.1.10"]} - assert cache.dns_cache == {"server.com": ["10.0.0.1"]} + cache = AddressCache.from_cli_args(mdns_args, dns_args) - # Check that warnings were logged for invalid entries - assert "Invalid cache format: invalid_format" in caplog.text - assert "Invalid cache format: also_invalid" in caplog.text + assert cache.mdns_cache == {"device.local": ["192.168.1.10", "192.168.1.11"]} + assert cache.dns_cache == {"server.com": ["10.0.0.1", "10.0.0.2", "10.0.0.3"]} - def test_from_cli_args_ipv6(self): - """Test handling of IPv6 addresses.""" - mdns_args = ["device.local=fe80::1,2001:db8::1"] - dns_args = ["server.com=2001:db8::2,::1"] - cache = AddressCache.from_cli_args(mdns_args, dns_args) +def test_from_cli_args_multiple_entries() -> None: + """Test creating cache with multiple host entries.""" + mdns_args: list[str] = [ + "device1.local=192.168.1.10", + "device2.local=192.168.1.20,192.168.1.21", + ] + dns_args: list[str] = ["server1.com=10.0.0.1", "server2.com=10.0.0.2"] - assert cache.mdns_cache == {"device.local": ["fe80::1", "2001:db8::1"]} - assert cache.dns_cache == {"server.com": ["2001:db8::2", "::1"]} + cache = AddressCache.from_cli_args(mdns_args, dns_args) - def test_logging_output(self, caplog): - """Test that appropriate debug logging occurs.""" - import logging + assert cache.mdns_cache == { + "device1.local": ["192.168.1.10"], + "device2.local": ["192.168.1.20", "192.168.1.21"], + } + assert cache.dns_cache == { + "server1.com": ["10.0.0.1"], + "server2.com": ["10.0.0.2"], + } - caplog.set_level(logging.DEBUG) - cache = AddressCache( - mdns_cache={"device.local": ["192.168.1.10"]}, - dns_cache={"server.com": ["10.0.0.1"]}, - ) +def test_from_cli_args_normalization() -> None: + """Test that CLI arguments are normalized.""" + mdns_args: list[str] = ["Device1.Local.=192.168.1.10", "DEVICE2.LOCAL=192.168.1.20"] + dns_args: list[str] = ["Server1.COM.=10.0.0.1", "SERVER2.com=10.0.0.2"] - # Test successful lookups log at debug level - result = cache.get_mdns_addresses("device.local") - assert result == ["192.168.1.10"] - assert "Using mDNS cache for device.local" in caplog.text + cache = AddressCache.from_cli_args(mdns_args, dns_args) - caplog.clear() - result = cache.get_dns_addresses("server.com") - assert result == ["10.0.0.1"] - assert "Using DNS cache for server.com" in caplog.text + # Hostnames should be normalized (lowercase, no trailing dots) + assert cache.mdns_cache == { + "device1.local": ["192.168.1.10"], + "device2.local": ["192.168.1.20"], + } + assert cache.dns_cache == { + "server1.com": ["10.0.0.1"], + "server2.com": ["10.0.0.2"], + } - # Test that failed lookups don't log - caplog.clear() - result = cache.get_mdns_addresses("unknown.local") - assert result is None - assert "Using mDNS cache" not in caplog.text + +def test_from_cli_args_whitespace_handling() -> None: + """Test that whitespace in IPs is handled.""" + mdns_args: list[str] = ["device.local= 192.168.1.10 , 192.168.1.11 "] + dns_args: list[str] = ["server.com= 10.0.0.1 , 10.0.0.2 "] + + cache = AddressCache.from_cli_args(mdns_args, dns_args) + + assert cache.mdns_cache == {"device.local": ["192.168.1.10", "192.168.1.11"]} + assert cache.dns_cache == {"server.com": ["10.0.0.1", "10.0.0.2"]} + + +def test_from_cli_args_invalid_format(caplog: LogCaptureFixture) -> None: + """Test handling of invalid argument format.""" + mdns_args: list[str] = ["invalid_format", "device.local=192.168.1.10"] + dns_args: list[str] = ["server.com=10.0.0.1", "also_invalid"] + + cache = AddressCache.from_cli_args(mdns_args, dns_args) + + # Valid entries should still be processed + assert cache.mdns_cache == {"device.local": ["192.168.1.10"]} + assert cache.dns_cache == {"server.com": ["10.0.0.1"]} + + # Check that warnings were logged for invalid entries + assert "Invalid cache format: invalid_format" in caplog.text + assert "Invalid cache format: also_invalid" in caplog.text + + +def test_from_cli_args_ipv6() -> None: + """Test handling of IPv6 addresses.""" + mdns_args: list[str] = ["device.local=fe80::1,2001:db8::1"] + dns_args: list[str] = ["server.com=2001:db8::2,::1"] + + cache = AddressCache.from_cli_args(mdns_args, dns_args) + + assert cache.mdns_cache == {"device.local": ["fe80::1", "2001:db8::1"]} + assert cache.dns_cache == {"server.com": ["2001:db8::2", "::1"]} + + +def test_logging_output(caplog: LogCaptureFixture) -> None: + """Test that appropriate debug logging occurs.""" + caplog.set_level(logging.DEBUG) + + cache = AddressCache( + mdns_cache={"device.local": ["192.168.1.10"]}, + dns_cache={"server.com": ["10.0.0.1"]}, + ) + + # Test successful lookups log at debug level + result: list[str] | None = cache.get_mdns_addresses("device.local") + assert result == ["192.168.1.10"] + assert "Using mDNS cache for device.local" in caplog.text + + caplog.clear() + result = cache.get_dns_addresses("server.com") + assert result == ["10.0.0.1"] + assert "Using DNS cache for server.com" in caplog.text + + # Test that failed lookups don't log + caplog.clear() + result = cache.get_mdns_addresses("unknown.local") + assert result is None + assert "Using mDNS cache" not in caplog.text + + +@pytest.mark.parametrize( + "hostname,expected", + [ + ("test.local", "test.local"), + ("Test.Local.", "test.local"), + ("TEST.LOCAL...", "test.local"), + ("example.com", "example.com"), + ("EXAMPLE.COM.", "example.com"), + ], +) +def test_normalize_hostname_parametrized(hostname: str, expected: str) -> None: + """Test hostname normalization with various inputs.""" + assert normalize_hostname(hostname) == expected + + +@pytest.mark.parametrize( + "mdns_arg,expected", + [ + ("host=1.2.3.4", {"host": ["1.2.3.4"]}), + ("Host.Local=1.2.3.4,5.6.7.8", {"host.local": ["1.2.3.4", "5.6.7.8"]}), + ("HOST.LOCAL.=::1", {"host.local": ["::1"]}), + ], +) +def test_parse_cache_args_parametrized( + mdns_arg: str, expected: dict[str, list[str]] +) -> None: + """Test parsing of cache arguments with various formats.""" + cache = AddressCache.from_cli_args([mdns_arg], []) + assert cache.mdns_cache == expected