mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-11-10 11:29:51 +00:00
953f7d01d7
* Improve DNS plug-in restart Instead of simply go by PrimaryConnectioon change, use the DnsManager Configuration property. This property is ultimately used to write the DNS plug-in configuration, so it is really the relevant information we pass on to the plug-in. * Check for changes and restart DNS plugin * Check for changes in plug-in DNS Cache last local (NetworkManager) provided DNS servers. Check against this DNS server list when deciding when to restart the DNS plug-in. * Check connectivity unthrottled in certain situations * Fix pytest * Fix pytest * Improve test coverage for DNS plugins restart functionality * Apply suggestions from code review Co-authored-by: Mike Degatano <michael.degatano@gmail.com> * Debounce local DNS changes and event based connectivity checks * Remove connection check logic * Remove unthrottled connectivity check * Fix delayed call * Store restart task and cancel in case a restart is running * Improve DNS configuration change tests * Remove stale code * Improve DNS plug-in tests, less mocking * Cover multiple private functions at once Improve tests around notify_locals_changed() to cover multiple functions at once. --------- Co-authored-by: Mike Degatano <michael.degatano@gmail.com>
135 lines
4.6 KiB
Python
135 lines
4.6 KiB
Python
"""Test supported features."""
|
|
|
|
# pylint: disable=protected-access
|
|
import asyncio
|
|
from unittest.mock import PropertyMock, patch
|
|
|
|
from dbus_fast import Variant
|
|
import pytest
|
|
|
|
from supervisor.coresys import CoreSys
|
|
from supervisor.plugins.dns import PluginDns
|
|
|
|
from tests.dbus_service_mocks.network_manager import (
|
|
NetworkManager as NetworkManagerService,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("force", [True, False])
|
|
async def test_connectivity_not_connected(
|
|
coresys: CoreSys, force: bool, network_manager_service: NetworkManagerService
|
|
):
|
|
"""Test host unknown connectivity."""
|
|
assert coresys.host.network.connectivity
|
|
|
|
network_manager_service.connectivity = 0
|
|
await coresys.host.network.check_connectivity(force=force)
|
|
assert not coresys.host.network.connectivity
|
|
|
|
|
|
async def test_connectivity_connected(
|
|
coresys: CoreSys, network_manager_service: NetworkManagerService
|
|
):
|
|
"""Test host full connectivity."""
|
|
network_manager_service.CheckConnectivity.calls.clear()
|
|
|
|
await coresys.host.network.check_connectivity()
|
|
assert coresys.host.network.connectivity
|
|
assert network_manager_service.CheckConnectivity.calls == []
|
|
|
|
await coresys.host.network.check_connectivity(force=True)
|
|
assert coresys.host.network.connectivity
|
|
assert network_manager_service.CheckConnectivity.calls == [()]
|
|
|
|
|
|
@pytest.mark.parametrize("force", [True, False])
|
|
async def test_connectivity_events(coresys: CoreSys, force: bool):
|
|
"""Test connectivity events."""
|
|
coresys.host.network.connectivity = None
|
|
await asyncio.sleep(0)
|
|
|
|
with patch.object(
|
|
type(coresys.homeassistant.websocket), "async_send_message"
|
|
) as send_message:
|
|
await coresys.host.network.check_connectivity(force=force)
|
|
await asyncio.sleep(0)
|
|
|
|
assert coresys.host.network.connectivity is True
|
|
send_message.assert_called_once_with(
|
|
{
|
|
"type": "supervisor/event",
|
|
"data": {
|
|
"event": "supervisor_update",
|
|
"update_key": "network",
|
|
"data": {"host_internet": True},
|
|
},
|
|
}
|
|
)
|
|
|
|
send_message.reset_mock()
|
|
with patch.object(
|
|
type(coresys.dbus.network),
|
|
"connectivity_enabled",
|
|
new=PropertyMock(return_value=False),
|
|
):
|
|
await coresys.host.network.check_connectivity(force=force)
|
|
await asyncio.sleep(0)
|
|
|
|
assert coresys.host.network.connectivity is None
|
|
send_message.assert_called_once_with(
|
|
{
|
|
"type": "supervisor/event",
|
|
"data": {
|
|
"event": "supervisor_update",
|
|
"update_key": "network",
|
|
"data": {"host_internet": None},
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
async def test_dns_configuration_change_triggers_notify_locals_changed(
|
|
coresys: CoreSys, dns_manager_service
|
|
):
|
|
"""Test that DNS configuration changes trigger notify_locals_changed."""
|
|
await coresys.host.network.load()
|
|
|
|
with patch.object(PluginDns, "notify_locals_changed") as notify_locals_changed:
|
|
# Test that non-Configuration changes don't trigger notify_locals_changed
|
|
dns_manager_service.emit_properties_changed({"Mode": "default"})
|
|
await dns_manager_service.ping()
|
|
notify_locals_changed.assert_not_called()
|
|
|
|
# Test that Configuration changes trigger notify_locals_changed
|
|
configuration = [
|
|
{
|
|
"nameservers": Variant("as", ["192.168.2.2"]),
|
|
"domains": Variant("as", ["lan"]),
|
|
"interface": Variant("s", "eth0"),
|
|
"priority": Variant("i", 100),
|
|
"vpn": Variant("b", False),
|
|
}
|
|
]
|
|
|
|
dns_manager_service.emit_properties_changed({"Configuration": configuration})
|
|
await dns_manager_service.ping()
|
|
notify_locals_changed.assert_called_once()
|
|
|
|
notify_locals_changed.reset_mock()
|
|
# Test that subsequent Configuration changes also trigger notify_locals_changed
|
|
different_configuration = [
|
|
{
|
|
"nameservers": Variant("as", ["8.8.8.8"]),
|
|
"domains": Variant("as", ["example.com"]),
|
|
"interface": Variant("s", "wlan0"),
|
|
"priority": Variant("i", 200),
|
|
"vpn": Variant("b", True),
|
|
}
|
|
]
|
|
|
|
dns_manager_service.emit_properties_changed(
|
|
{"Configuration": different_configuration}
|
|
)
|
|
await dns_manager_service.ping()
|
|
notify_locals_changed.assert_called_once()
|