From 305b4504de66ee14157a17ed16ac21977de6726d Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Thu, 11 Sep 2025 19:14:03 -0500 Subject: [PATCH] wip --- esphome/dashboard/web_server.py | 6 +- tests/dashboard/status/test_dns.py | 195 ++++++++-------------------- tests/dashboard/status/test_mdns.py | 28 ++-- 3 files changed, 76 insertions(+), 153 deletions(-) diff --git a/esphome/dashboard/web_server.py b/esphome/dashboard/web_server.py index 90a7cab3b5..2ab449b8da 100644 --- a/esphome/dashboard/web_server.py +++ b/esphome/dashboard/web_server.py @@ -358,7 +358,7 @@ class EsphomePortCommandWebSocket(EsphomeCommandWebSocket): normalized = use_address.rstrip(".").lower() cache_args.extend( [ - "--mdns-lookup-cache", + "--mdns-address-cache", f"{normalized}={','.join(sort_ip_addresses(cached))}", ] ) @@ -372,7 +372,7 @@ class EsphomePortCommandWebSocket(EsphomeCommandWebSocket): normalized = use_address.rstrip(".").lower() cache_args.extend( [ - "--dns-lookup-cache", + "--dns-address-cache", f"{normalized}={','.join(sort_ip_addresses(cached))}", ] ) @@ -396,7 +396,7 @@ class EsphomePortCommandWebSocket(EsphomeCommandWebSocket): normalized = mdns_name.rstrip(".").lower() cache_args.extend( [ - "--mdns-lookup-cache", + "--mdns-address-cache", f"{normalized}={','.join(sort_ip_addresses(cached))}", ] ) diff --git a/tests/dashboard/status/test_dns.py b/tests/dashboard/status/test_dns.py index 519defcbe1..9ca48ba2d8 100644 --- a/tests/dashboard/status/test_dns.py +++ b/tests/dashboard/status/test_dns.py @@ -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() diff --git a/tests/dashboard/status/test_mdns.py b/tests/dashboard/status/test_mdns.py index b451157922..b20ba6699d 100644 --- a/tests/dashboard/status/test_mdns.py +++ b/tests/dashboard/status/test_mdns.py @@ -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