sascha/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2025-10-31 07:03:55 +00:00
Code Issues Releases Wiki Activity

wip

Browse Source
This commit is contained in:
J. Nick Koston
2025-09-11 19:14:03 -05:00
parent 5dbe56849a
commit 305b4504de
3 changed files with 76 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

195
tests/dashboard/status/test_dns.py
View File

@@ -11,192 +11,111 @@ from esphome.dashboard.dns import DNSCache
@pytest.fixture
def dns_cache() -> DNSCache:
def dns_cache_fixture() -> DNSCache:
"""Create a DNSCache instance."""
return DNSCache()
def test_get_cached_addresses_not_in_cache(dns_cache: DNSCache) -> None:
def test_get_cached_addresses_not_in_cache(dns_cache_fixture: DNSCache) -> None:
"""Test get_cached_addresses when hostname is not in cache."""
now = time.monotonic()
result = dns_cache.get_cached_addresses("unknown.example.com", now)
result = dns_cache_fixture.get_cached_addresses("unknown.example.com", now)
assert result is None
def test_get_cached_addresses_expired(dns_cache: DNSCache) -> None:
def test_get_cached_addresses_expired(dns_cache_fixture: DNSCache) -> None:
"""Test get_cached_addresses when cache entry is expired."""
now = time.monotonic()
# Add entry that's already expired
dns_cache.cache["example.com"] = (["192.168.1.10"], now - 1)
dns_cache_fixture._cache["example.com"] = (now - 1, ["192.168.1.10"])
result = dns_cache.get_cached_addresses("example.com", now)
result = dns_cache_fixture.get_cached_addresses("example.com", now)
assert result is None
# Expired entry should be removed
assert "example.com" not in dns_cache.cache
# Expired entry should still be in cache (not removed by get_cached_addresses)
assert "example.com" in dns_cache_fixture._cache
def test_get_cached_addresses_valid(dns_cache: DNSCache) -> None:
def test_get_cached_addresses_valid(dns_cache_fixture: DNSCache) -> None:
"""Test get_cached_addresses with valid cache entry."""
now = time.monotonic()
# Add entry that expires in 60 seconds
dns_cache.cache["example.com"] = (["192.168.1.10", "192.168.1.11"], now + 60)
dns_cache_fixture._cache["example.com"] = (
now + 60,
["192.168.1.10", "192.168.1.11"],
)
result = dns_cache.get_cached_addresses("example.com", now)
result = dns_cache_fixture.get_cached_addresses("example.com", now)
assert result == ["192.168.1.10", "192.168.1.11"]
# Entry should still be in cache
assert "example.com" in dns_cache.cache
assert "example.com" in dns_cache_fixture._cache
def test_get_cached_addresses_hostname_normalization(dns_cache: DNSCache) -> None:
def test_get_cached_addresses_hostname_normalization(
dns_cache_fixture: DNSCache,
) -> None:
"""Test get_cached_addresses normalizes hostname."""
now = time.monotonic()
# Add entry with lowercase hostname
dns_cache.cache["example.com"] = (["192.168.1.10"], now + 60)
dns_cache_fixture._cache["example.com"] = (now + 60, ["192.168.1.10"])
# Test with various forms
assert dns_cache.get_cached_addresses("EXAMPLE.COM", now) == ["192.168.1.10"]
assert dns_cache.get_cached_addresses("example.com.", now) == ["192.168.1.10"]
assert dns_cache.get_cached_addresses("EXAMPLE.COM.", now) == ["192.168.1.10"]
assert dns_cache_fixture.get_cached_addresses("EXAMPLE.COM", now) == [
"192.168.1.10"
]
assert dns_cache_fixture.get_cached_addresses("example.com.", now) == [
"192.168.1.10"
]
assert dns_cache_fixture.get_cached_addresses("EXAMPLE.COM.", now) == [
"192.168.1.10"
]
def test_get_cached_addresses_ipv6(dns_cache: DNSCache) -> None:
def test_get_cached_addresses_ipv6(dns_cache_fixture: DNSCache) -> None:
"""Test get_cached_addresses with IPv6 addresses."""
now = time.monotonic()
dns_cache.cache["example.com"] = (["2001:db8::1", "fe80::1"], now + 60)
dns_cache_fixture._cache["example.com"] = (now + 60, ["2001:db8::1", "fe80::1"])
result = dns_cache.get_cached_addresses("example.com", now)
result = dns_cache_fixture.get_cached_addresses("example.com", now)
assert result == ["2001:db8::1", "fe80::1"]
def test_get_cached_addresses_empty_list(dns_cache: DNSCache) -> None:
def test_get_cached_addresses_empty_list(dns_cache_fixture: DNSCache) -> None:
"""Test get_cached_addresses with empty address list."""
now = time.monotonic()
dns_cache.cache["example.com"] = ([], now + 60)
dns_cache_fixture._cache["example.com"] = (now + 60, [])
result = dns_cache.get_cached_addresses("example.com", now)
result = dns_cache_fixture.get_cached_addresses("example.com", now)
assert result == []
def test_resolve_addresses_already_cached(dns_cache: DNSCache) -> None:
"""Test resolve_addresses when hostname is already cached."""
def test_get_cached_addresses_exception_in_cache(dns_cache_fixture: DNSCache) -> None:
"""Test get_cached_addresses when cache contains an exception."""
now = time.monotonic()
dns_cache.cache["example.com"] = (["192.168.1.10"], now + 60)
# Store an exception (from failed resolution)
dns_cache_fixture._cache["example.com"] = (now + 60, OSError("Resolution failed"))
with patch("socket.getaddrinfo") as mock_getaddrinfo:
result = dns_cache.resolve_addresses("example.com", ["example.com"])
assert result == ["192.168.1.10"]
# Should not call getaddrinfo for cached entry
mock_getaddrinfo.assert_not_called()
result = dns_cache_fixture.get_cached_addresses("example.com", now)
assert result is None # Should return None for exceptions
def test_resolve_addresses_not_cached(dns_cache: DNSCache) -> None:
"""Test resolve_addresses when hostname needs resolution."""
with patch("socket.getaddrinfo") as mock_getaddrinfo:
mock_getaddrinfo.return_value = [
(None, None, None, None, ("192.168.1.10", 0)),
(None, None, None, None, ("192.168.1.11", 0)),
]
result = dns_cache.resolve_addresses("example.com", ["example.com"])
assert result == ["192.168.1.10", "192.168.1.11"]
mock_getaddrinfo.assert_called_once_with("example.com", 0)
# Should be cached now
assert "example.com" in dns_cache.cache
def test_resolve_addresses_multiple_hostnames(dns_cache: DNSCache) -> None:
"""Test resolve_addresses with multiple hostnames."""
def test_async_resolve_not_called(dns_cache_fixture: DNSCache) -> None:
"""Test that get_cached_addresses never calls async_resolve."""
now = time.monotonic()
dns_cache.cache["cached.com"] = (["192.168.1.10"], now + 60)
with patch("socket.getaddrinfo") as mock_getaddrinfo:
mock_getaddrinfo.return_value = [
(None, None, None, None, ("10.0.0.1", 0)),
]
with patch.object(dns_cache_fixture, "async_resolve") as mock_resolve:
# Test non-cached
result = dns_cache_fixture.get_cached_addresses("uncached.com", now)
assert result is None
mock_resolve.assert_not_called()
result = dns_cache.resolve_addresses(
"primary.com", ["cached.com", "primary.com", "fallback.com"]
)
# Should return cached result for first match
# Test expired
dns_cache_fixture._cache["expired.com"] = (now - 1, ["192.168.1.10"])
result = dns_cache_fixture.get_cached_addresses("expired.com", now)
assert result is None
mock_resolve.assert_not_called()
# Test valid
dns_cache_fixture._cache["valid.com"] = (now + 60, ["192.168.1.10"])
result = dns_cache_fixture.get_cached_addresses("valid.com", now)
assert result == ["192.168.1.10"]
mock_getaddrinfo.assert_not_called()
def test_resolve_addresses_resolution_error(dns_cache: DNSCache) -> None:
"""Test resolve_addresses when resolution fails."""
with patch("socket.getaddrinfo") as mock_getaddrinfo:
mock_getaddrinfo.side_effect = OSError("Name resolution failed")
result = dns_cache.resolve_addresses("example.com", ["example.com"])
assert result == []
# Failed resolution should not be cached
assert "example.com" not in dns_cache.cache
def test_resolve_addresses_ipv6_resolution(dns_cache: DNSCache) -> None:
"""Test resolve_addresses with IPv6 results."""
with patch("socket.getaddrinfo") as mock_getaddrinfo:
mock_getaddrinfo.return_value = [
(None, None, None, None, ("2001:db8::1", 0, 0, 0)),
(None, None, None, None, ("fe80::1", 0, 0, 0)),
]
result = dns_cache.resolve_addresses("example.com", ["example.com"])
assert result == ["2001:db8::1", "fe80::1"]
def test_resolve_addresses_duplicate_removal(dns_cache: DNSCache) -> None:
"""Test resolve_addresses removes duplicate addresses."""
with patch("socket.getaddrinfo") as mock_getaddrinfo:
mock_getaddrinfo.return_value = [
(None, None, None, None, ("192.168.1.10", 0)),
(None, None, None, None, ("192.168.1.10", 0)), # Duplicate
(None, None, None, None, ("192.168.1.11", 0)),
]
result = dns_cache.resolve_addresses("example.com", ["example.com"])
assert result == ["192.168.1.10", "192.168.1.11"]
def test_resolve_addresses_hostname_normalization(dns_cache: DNSCache) -> None:
"""Test resolve_addresses normalizes hostnames."""
with patch("socket.getaddrinfo") as mock_getaddrinfo:
mock_getaddrinfo.return_value = [
(None, None, None, None, ("192.168.1.10", 0)),
]
# Resolve with uppercase and trailing dot
result = dns_cache.resolve_addresses("EXAMPLE.COM.", ["EXAMPLE.COM."])
assert result == ["192.168.1.10"]
# Should be cached with normalized name
assert "example.com" in dns_cache.cache
# Should use cached result for different forms
result = dns_cache.resolve_addresses("example.com", ["example.com"])
assert result == ["192.168.1.10"]
# Only called once due to caching
mock_getaddrinfo.assert_called_once()
def test_cache_expiration_ttl(dns_cache: DNSCache) -> None:
"""Test that cache entries expire after TTL."""
with patch("socket.getaddrinfo") as mock_getaddrinfo:
mock_getaddrinfo.return_value = [
(None, None, None, None, ("192.168.1.10", 0)),
]
# First resolution
result = dns_cache.resolve_addresses("example.com", ["example.com"])
assert result == ["192.168.1.10"]
assert mock_getaddrinfo.call_count == 1
# Simulate time passing beyond TTL
with patch("time.monotonic") as mock_time:
mock_time.return_value = time.monotonic() + 301 # TTL is 300 seconds
# Should trigger new resolution
result = dns_cache.resolve_addresses("example.com", ["example.com"])
assert result == ["192.168.1.10"]
assert mock_getaddrinfo.call_count == 2
mock_resolve.assert_not_called()

28
tests/dashboard/status/test_mdns.py
View File

@@ -153,19 +153,23 @@ def test_get_cached_addresses_empty_list(mdns_status: MDNSStatus) -> None:
def test_async_setup_success(mock_dashboard: Mock) -> None:
"""Test successful async_setup."""
mdns_status = MDNSStatus(mock_dashboard)
with patch("esphome.dashboard.status.mdns.AsyncEsphomeZeroconf") as mock_zc:
mock_zc.return_value = Mock()
result = mdns_status.async_setup()
assert result is True
assert mdns_status.aiozc is not None
with patch("asyncio.get_running_loop") as mock_loop:
mock_loop.return_value = Mock()
mdns_status = MDNSStatus(mock_dashboard)
with patch("esphome.dashboard.status.mdns.AsyncEsphomeZeroconf") as mock_zc:
mock_zc.return_value = Mock()
result = mdns_status.async_setup()
assert result is True
assert mdns_status.aiozc is not None
def test_async_setup_failure(mock_dashboard: Mock) -> None:
"""Test async_setup with OSError."""
mdns_status = MDNSStatus(mock_dashboard)
with patch("esphome.dashboard.status.mdns.AsyncEsphomeZeroconf") as mock_zc:
mock_zc.side_effect = OSError("Network error")
result = mdns_status.async_setup()
assert result is False
assert mdns_status.aiozc is None
with patch("asyncio.get_running_loop") as mock_loop:
mock_loop.return_value = Mock()
mdns_status = MDNSStatus(mock_dashboard)
with patch("esphome.dashboard.status.mdns.AsyncEsphomeZeroconf") as mock_zc:
mock_zc.side_effect = OSError("Network error")
result = mdns_status.async_setup()
assert result is False
assert mdns_status.aiozc is None