[dashboard] Transfer DNS/mDNS cache from dashboard to CLI to avoid blocking (#10685)

tests/dashboard/status/test_dns.py

@@ -0,0 +1,121 @@
+"""Unit tests for esphome.dashboard.dns module."""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import patch
+
+import pytest
+
+from esphome.dashboard.dns import DNSCache
+
+
+@pytest.fixture
+def dns_cache_fixture() -> DNSCache:
+    """Create a DNSCache instance."""
+    return DNSCache()
+
+
+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_fixture.get_cached_addresses("unknown.example.com", now)
+    assert result is 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_fixture._cache["example.com"] = (now - 1, ["192.168.1.10"])
+
+    result = dns_cache_fixture.get_cached_addresses("example.com", now)
+    assert result is None
+    # 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_fixture: DNSCache) -> None:
+    """Test get_cached_addresses with valid cache entry."""
+    now = time.monotonic()
+    # Add entry that expires in 60 seconds
+    dns_cache_fixture._cache["example.com"] = (
+        now + 60,
+        ["192.168.1.10", "192.168.1.11"],
+    )
+
+    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_fixture._cache
+
+
+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_fixture._cache["example.com"] = (now + 60, ["192.168.1.10"])
+
+    # Test with various forms
+    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_fixture: DNSCache) -> None:
+    """Test get_cached_addresses with IPv6 addresses."""
+    now = time.monotonic()
+    dns_cache_fixture._cache["example.com"] = (now + 60, ["2001:db8::1", "fe80::1"])
+
+    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_fixture: DNSCache) -> None:
+    """Test get_cached_addresses with empty address list."""
+    now = time.monotonic()
+    dns_cache_fixture._cache["example.com"] = (now + 60, [])
+
+    result = dns_cache_fixture.get_cached_addresses("example.com", now)
+    assert result == []
+
+
+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()
+    # Store an exception (from failed resolution)
+    dns_cache_fixture._cache["example.com"] = (now + 60, OSError("Resolution failed"))
+
+    result = dns_cache_fixture.get_cached_addresses("example.com", now)
+    assert result is None  # Should return None for exceptions
+
+
+def test_async_resolve_not_called(dns_cache_fixture: DNSCache) -> None:
+    """Test that get_cached_addresses never calls async_resolve."""
+    now = time.monotonic()
+
+    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()
+
+        # 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_resolve.assert_not_called()

tests/dashboard/status/test_mdns.py

@@ -0,0 +1,168 @@
+"""Unit tests for esphome.dashboard.status.mdns module."""
+
+from __future__ import annotations
+
+from unittest.mock import Mock, patch
+
+import pytest
+import pytest_asyncio
+from zeroconf import AddressResolver, IPVersion
+
+from esphome.dashboard.status.mdns import MDNSStatus
+
+
+@pytest_asyncio.fixture
+async def mdns_status(mock_dashboard: Mock) -> MDNSStatus:
+    """Create an MDNSStatus instance in async context."""
+    # We're in an async context so get_running_loop will work
+    return MDNSStatus(mock_dashboard)
+
+
+@pytest.mark.asyncio
+async def test_get_cached_addresses_no_zeroconf(mdns_status: MDNSStatus) -> None:
+    """Test get_cached_addresses when no zeroconf instance is available."""
+    mdns_status.aiozc = None
+    result = mdns_status.get_cached_addresses("device.local")
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_get_cached_addresses_not_in_cache(mdns_status: MDNSStatus) -> None:
+    """Test get_cached_addresses when address is not in cache."""
+    mdns_status.aiozc = Mock()
+    mdns_status.aiozc.zeroconf = Mock()
+
+    with patch("esphome.dashboard.status.mdns.AddressResolver") as mock_resolver:
+        mock_info = Mock(spec=AddressResolver)
+        mock_info.load_from_cache.return_value = False
+        mock_resolver.return_value = mock_info
+
+        result = mdns_status.get_cached_addresses("device.local")
+        assert result is None
+        mock_info.load_from_cache.assert_called_once_with(mdns_status.aiozc.zeroconf)
+
+
+@pytest.mark.asyncio
+async def test_get_cached_addresses_found_in_cache(mdns_status: MDNSStatus) -> None:
+    """Test get_cached_addresses when address is found in cache."""
+    mdns_status.aiozc = Mock()
+    mdns_status.aiozc.zeroconf = Mock()
+
+    with patch("esphome.dashboard.status.mdns.AddressResolver") as mock_resolver:
+        mock_info = Mock(spec=AddressResolver)
+        mock_info.load_from_cache.return_value = True
+        mock_info.parsed_scoped_addresses.return_value = ["192.168.1.10", "fe80::1"]
+        mock_resolver.return_value = mock_info
+
+        result = mdns_status.get_cached_addresses("device.local")
+        assert result == ["192.168.1.10", "fe80::1"]
+        mock_info.load_from_cache.assert_called_once_with(mdns_status.aiozc.zeroconf)
+        mock_info.parsed_scoped_addresses.assert_called_once_with(IPVersion.All)
+
+
+@pytest.mark.asyncio
+async def test_get_cached_addresses_with_trailing_dot(mdns_status: MDNSStatus) -> None:
+    """Test get_cached_addresses with hostname having trailing dot."""
+    mdns_status.aiozc = Mock()
+    mdns_status.aiozc.zeroconf = Mock()
+
+    with patch("esphome.dashboard.status.mdns.AddressResolver") as mock_resolver:
+        mock_info = Mock(spec=AddressResolver)
+        mock_info.load_from_cache.return_value = True
+        mock_info.parsed_scoped_addresses.return_value = ["192.168.1.10"]
+        mock_resolver.return_value = mock_info
+
+        result = mdns_status.get_cached_addresses("device.local.")
+        assert result == ["192.168.1.10"]
+        # Should normalize to device.local. for zeroconf
+        mock_resolver.assert_called_once_with("device.local.")
+
+
+@pytest.mark.asyncio
+async def test_get_cached_addresses_uppercase_hostname(mdns_status: MDNSStatus) -> None:
+    """Test get_cached_addresses with uppercase hostname."""
+    mdns_status.aiozc = Mock()
+    mdns_status.aiozc.zeroconf = Mock()
+
+    with patch("esphome.dashboard.status.mdns.AddressResolver") as mock_resolver:
+        mock_info = Mock(spec=AddressResolver)
+        mock_info.load_from_cache.return_value = True
+        mock_info.parsed_scoped_addresses.return_value = ["192.168.1.10"]
+        mock_resolver.return_value = mock_info
+
+        result = mdns_status.get_cached_addresses("DEVICE.LOCAL")
+        assert result == ["192.168.1.10"]
+        # Should normalize to device.local. for zeroconf
+        mock_resolver.assert_called_once_with("device.local.")
+
+
+@pytest.mark.asyncio
+async def test_get_cached_addresses_simple_hostname(mdns_status: MDNSStatus) -> None:
+    """Test get_cached_addresses with simple hostname (no domain)."""
+    mdns_status.aiozc = Mock()
+    mdns_status.aiozc.zeroconf = Mock()
+
+    with patch("esphome.dashboard.status.mdns.AddressResolver") as mock_resolver:
+        mock_info = Mock(spec=AddressResolver)
+        mock_info.load_from_cache.return_value = True
+        mock_info.parsed_scoped_addresses.return_value = ["192.168.1.10"]
+        mock_resolver.return_value = mock_info
+
+        result = mdns_status.get_cached_addresses("device")
+        assert result == ["192.168.1.10"]
+        # Should append .local. for zeroconf
+        mock_resolver.assert_called_once_with("device.local.")
+
+
+@pytest.mark.asyncio
+async def test_get_cached_addresses_ipv6_only(mdns_status: MDNSStatus) -> None:
+    """Test get_cached_addresses returning only IPv6 addresses."""
+    mdns_status.aiozc = Mock()
+    mdns_status.aiozc.zeroconf = Mock()
+
+    with patch("esphome.dashboard.status.mdns.AddressResolver") as mock_resolver:
+        mock_info = Mock(spec=AddressResolver)
+        mock_info.load_from_cache.return_value = True
+        mock_info.parsed_scoped_addresses.return_value = ["fe80::1", "2001:db8::1"]
+        mock_resolver.return_value = mock_info
+
+        result = mdns_status.get_cached_addresses("device.local")
+        assert result == ["fe80::1", "2001:db8::1"]
+
+
+@pytest.mark.asyncio
+async def test_get_cached_addresses_empty_list(mdns_status: MDNSStatus) -> None:
+    """Test get_cached_addresses returning empty list from cache."""
+    mdns_status.aiozc = Mock()
+    mdns_status.aiozc.zeroconf = Mock()
+
+    with patch("esphome.dashboard.status.mdns.AddressResolver") as mock_resolver:
+        mock_info = Mock(spec=AddressResolver)
+        mock_info.load_from_cache.return_value = True
+        mock_info.parsed_scoped_addresses.return_value = []
+        mock_resolver.return_value = mock_info
+
+        result = mdns_status.get_cached_addresses("device.local")
+        assert result == []
+
+
+@pytest.mark.asyncio
+async 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
+
+
+@pytest.mark.asyncio
+async 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