1
0
mirror of https://github.com/esphome/esphome.git synced 2025-09-19 03:32:20 +01:00
This commit is contained in:
J. Nick Koston
2025-09-11 19:00:31 -05:00
parent b9bf81fffc
commit 7dcedbae09

View File

@@ -1,260 +1,305 @@
"""Tests for the address_cache module.""" """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 from esphome.address_cache import AddressCache, normalize_hostname
class TestNormalizeHostname: def test_normalize_simple_hostname() -> None:
"""Test the normalize_hostname function.""" """Test normalizing a simple hostname."""
assert normalize_hostname("device") == "device"
def test_normalize_simple_hostname(self): assert normalize_hostname("device.local") == "device.local"
"""Test normalizing a simple hostname.""" assert normalize_hostname("server.example.com") == "server.example.com"
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"
class TestAddressCache: def test_normalize_removes_trailing_dots() -> None:
"""Test the AddressCache class.""" """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): def test_normalize_converts_to_lowercase() -> None:
"""Test initialization with provided caches.""" """Test that hostnames are converted to lowercase."""
mdns_cache = {"device.local": ["192.168.1.10"]} assert normalize_hostname("DEVICE") == "device"
dns_cache = {"server.com": ["10.0.0.1"]} assert normalize_hostname("Device.Local") == "device.local"
cache = AddressCache(mdns_cache=mdns_cache, dns_cache=dns_cache) assert normalize_hostname("Server.Example.COM") == "server.example.com"
assert cache.mdns_cache == mdns_cache
assert cache.dns_cache == dns_cache
assert cache.has_cache()
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 def test_normalize_combined() -> None:
assert cache.get_mdns_addresses("device.local") == [ """Test combination of trailing dots and case conversion."""
"192.168.1.10", assert normalize_hostname("DEVICE.LOCAL.") == "device.local"
"192.168.1.11", 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 def test_init_empty() -> None:
assert cache.get_mdns_addresses("device.local.") == [ """Test initialization with empty caches."""
"192.168.1.10", cache = AddressCache()
"192.168.1.11", 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): def test_init_with_caches() -> None:
"""Test getting DNS addresses.""" """Test initialization with provided caches."""
cache = AddressCache(dns_cache={"server.com": ["10.0.0.1", "10.0.0.2"]}) 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 def test_get_mdns_addresses() -> None:
assert cache.get_dns_addresses("Server.COM") == ["10.0.0.1", "10.0.0.2"] """Test getting mDNS addresses."""
cache = AddressCache(mdns_cache={"device.local": ["192.168.1.10", "192.168.1.11"]})
# With trailing dot # Direct lookup
assert cache.get_dns_addresses("server.com.") == ["10.0.0.1", "10.0.0.2"] assert cache.get_mdns_addresses("device.local") == [
"192.168.1.10",
"192.168.1.11",
]
# Not found # Case insensitive lookup
assert cache.get_dns_addresses("unknown.com") is None assert cache.get_mdns_addresses("Device.Local") == [
"192.168.1.10",
"192.168.1.11",
]
def test_get_addresses_auto_detection(self): # With trailing dot
"""Test automatic cache selection based on hostname.""" assert cache.get_mdns_addresses("device.local.") == [
cache = AddressCache( "192.168.1.10",
mdns_cache={"device.local": ["192.168.1.10"]}, "192.168.1.11",
dns_cache={"server.com": ["10.0.0.1"]}, ]
)
# Should use mDNS cache for .local domains # Not found
assert cache.get_addresses("device.local") == ["192.168.1.10"] assert cache.get_mdns_addresses("unknown.local") is None
assert cache.get_addresses("device.local.") == ["192.168.1.10"]
assert cache.get_addresses("Device.Local") == ["192.168.1.10"]
# 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 def test_get_dns_addresses() -> None:
assert cache.get_addresses("unknown.local") is None """Test getting DNS addresses."""
assert cache.get_addresses("unknown.com") is None cache = AddressCache(dns_cache={"server.com": ["10.0.0.1", "10.0.0.2"]})
def test_has_cache(self): # Direct lookup
"""Test checking if cache has entries.""" assert cache.get_dns_addresses("server.com") == ["10.0.0.1", "10.0.0.2"]
# Empty cache
cache = AddressCache()
assert not cache.has_cache()
# Only mDNS cache # Case insensitive lookup
cache = AddressCache(mdns_cache={"device.local": ["192.168.1.10"]}) assert cache.get_dns_addresses("Server.COM") == ["10.0.0.1", "10.0.0.2"]
assert cache.has_cache()
# Only DNS cache # With trailing dot
cache = AddressCache(dns_cache={"server.com": ["10.0.0.1"]}) assert cache.get_dns_addresses("server.com.") == ["10.0.0.1", "10.0.0.2"]
assert cache.has_cache()
# Both caches # Not found
cache = AddressCache( assert cache.get_dns_addresses("unknown.com") is None
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_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): def test_get_addresses_auto_detection() -> None:
"""Test creating cache from single CLI argument.""" """Test automatic cache selection based on hostname."""
mdns_args = ["device.local=192.168.1.10"] cache = AddressCache(
dns_args = ["server.com=10.0.0.1"] 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"]} # Should use DNS cache for non-.local domains
assert cache.dns_cache == {"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"]
assert cache.get_addresses("Server.COM") == ["10.0.0.1"]
def test_from_cli_args_multiple_ips(self): # Not found
"""Test creating cache with multiple IPs per host.""" assert cache.get_addresses("unknown.local") is None
mdns_args = ["device.local=192.168.1.10,192.168.1.11"] assert cache.get_addresses("unknown.com") is None
dns_args = ["server.com=10.0.0.1,10.0.0.2,10.0.0.3"]
cache = AddressCache.from_cli_args(mdns_args, dns_args)
assert cache.mdns_cache == {"device.local": ["192.168.1.10", "192.168.1.11"]} def test_has_cache() -> None:
assert cache.dns_cache == {"server.com": ["10.0.0.1", "10.0.0.2", "10.0.0.3"]} """Test checking if cache has entries."""
# Empty cache
cache = AddressCache()
assert not cache.has_cache()
def test_from_cli_args_multiple_entries(self): # Only mDNS cache
"""Test creating cache with multiple host entries.""" cache = AddressCache(mdns_cache={"device.local": ["192.168.1.10"]})
mdns_args = [ assert cache.has_cache()
"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"]
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 == { # Both caches
"device1.local": ["192.168.1.10"], cache = AddressCache(
"device2.local": ["192.168.1.20", "192.168.1.21"], mdns_cache={"device.local": ["192.168.1.10"]},
} dns_cache={"server.com": ["10.0.0.1"]},
assert cache.dns_cache == { )
"server1.com": ["10.0.0.1"], assert cache.has_cache()
"server2.com": ["10.0.0.2"],
}
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): def test_from_cli_args_single_entry() -> None:
"""Test that whitespace in IPs is handled.""" """Test creating cache from single CLI argument."""
mdns_args = ["device.local= 192.168.1.10 , 192.168.1.11 "] mdns_args: list[str] = ["device.local=192.168.1.10"]
dns_args = ["server.com= 10.0.0.1 , 10.0.0.2 "] 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.mdns_cache == {"device.local": ["192.168.1.10"]}
assert cache.dns_cache == {"server.com": ["10.0.0.1", "10.0.0.2"]} 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 cache = AddressCache.from_cli_args(mdns_args, dns_args)
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 cache.mdns_cache == {"device.local": ["192.168.1.10", "192.168.1.11"]}
assert "Invalid cache format: invalid_format" in caplog.text assert cache.dns_cache == {"server.com": ["10.0.0.1", "10.0.0.2", "10.0.0.3"]}
assert "Invalid cache format: also_invalid" in caplog.text
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"]} cache = AddressCache.from_cli_args(mdns_args, dns_args)
assert cache.dns_cache == {"server.com": ["2001:db8::2", "::1"]}
def test_logging_output(self, caplog): assert cache.mdns_cache == {
"""Test that appropriate debug logging occurs.""" "device1.local": ["192.168.1.10"],
import logging "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( def test_from_cli_args_normalization() -> None:
mdns_cache={"device.local": ["192.168.1.10"]}, """Test that CLI arguments are normalized."""
dns_cache={"server.com": ["10.0.0.1"]}, 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 cache = AddressCache.from_cli_args(mdns_args, dns_args)
result = cache.get_mdns_addresses("device.local")
assert result == ["192.168.1.10"]
assert "Using mDNS cache for device.local" in caplog.text
caplog.clear() # Hostnames should be normalized (lowercase, no trailing dots)
result = cache.get_dns_addresses("server.com") assert cache.mdns_cache == {
assert result == ["10.0.0.1"] "device1.local": ["192.168.1.10"],
assert "Using DNS cache for server.com" in caplog.text "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() def test_from_cli_args_whitespace_handling() -> None:
result = cache.get_mdns_addresses("unknown.local") """Test that whitespace in IPs is handled."""
assert result is None mdns_args: list[str] = ["device.local= 192.168.1.10 , 192.168.1.11 "]
assert "Using mDNS cache" not in caplog.text 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