mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-11-11 20:10:16 +00:00
8a95113ebd
* Fix NetworkManager connection name for VLANs The connection name for VLANs should include the parent interface name for better identification. This was originally the intention, but the interface object's name property was used which appears empty at that point. * Disallow creating multiple connections for the same VLAN id Only allow a single connection per interface and VLAN id. The regular network commands can be used to alter the configuration. * Fix pytest * Simply connection id name generation Always rely on the Supervisor interface representation's name attribute to generate the NetworkManager connection id. Make sure that the name is correctly set when creating VLAN interfaces as well. * Special case VLAN configuration We can't use the match information when comparing Supervisor interface representation with D-Bus representations. Special case VLAN and compare using VLAN ID and parent interface. Note that this currently compares connection UUID of the parent interface. * Fix pytest * Separate VLAN creation logic from apply_changes Apply changes is really all about updating the NetworkManager settings of a particular network interface. The base in apply_changes() is NetworkInterface class, which is the NetworkManager Device abstraction. All physical interfaces have such a Device hence it is always present. The only exception is when creating a VLAN: Since it is a virtual device, there is no device when creating a VLAN. This separate the two cases. This makes it much easier to reason if a VLAN already exists or not, and to handle the case where a VLAN needs to be created. For all other network interfaces, the apply_changes() method can now rely on the presence of the NetworkInterface Device abstraction. * Add VLAN test interface and VLAN exists test Add a test which checks that an error gets raised when a VLAN for a particular interface/id combination already exists. * Address pylint * Fix test_ignore_veth_only_changes pytest * Make VLAN interface disabled to avoid test issues * Reference setting 38 in mocked connection * Make sure interface type matches Require a interface type match before doing any comparision. * Add Supervisor host network configuration tests * Fix device type checking * Fix pytest * Fix tests by taking VLAN interface into account * Fix test_load_with_network_connection_issues This seems like a hack, but it turns out that the additional active connection caused coresys.host.network.update() to be called, which implicitly "fake" activated the connection. Now it seems that our mocking causes IPv4 gateway to be set. So in a way, the test checked a particular mock behavior instead of actual intention. The crucial part of this test is that we make sure the settings remain unchanged. This is done by ensuring that the the method is still auto. * Fix test_check_network_interface_ipv4.py Now that we have the VLAN interface active too it will raise an issue as well. * Apply suggestions from code review Co-authored-by: Mike Degatano <michael.degatano@gmail.com> * Fix ruff check issue --------- Co-authored-by: Mike Degatano <michael.degatano@gmail.com>
282 lines
10 KiB
Python
282 lines
10 KiB
Python
"""Test network manager."""
|
|
|
|
import asyncio
|
|
from ipaddress import IPv4Address, IPv6Address
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from dbus_fast import Variant
|
|
import pytest
|
|
|
|
from supervisor.const import CoreState
|
|
from supervisor.coresys import CoreSys
|
|
from supervisor.dbus.const import InterfaceMethod
|
|
from supervisor.exceptions import HostNotSupportedError
|
|
from supervisor.homeassistant.const import WSEvent, WSType
|
|
from supervisor.host.const import WifiMode
|
|
|
|
from tests.dbus_service_mocks.base import DBusServiceMock
|
|
from tests.dbus_service_mocks.network_active_connection import (
|
|
ActiveConnection as ActiveConnectionService,
|
|
)
|
|
from tests.dbus_service_mocks.network_connection_settings import (
|
|
SETTINGS_1_FIXTURE,
|
|
ConnectionSettings as ConnectionSettingsService,
|
|
)
|
|
from tests.dbus_service_mocks.network_device_wireless import (
|
|
DeviceWireless as DeviceWirelessService,
|
|
)
|
|
from tests.dbus_service_mocks.network_manager import (
|
|
NetworkManager as NetworkManagerService,
|
|
)
|
|
|
|
|
|
@pytest.fixture(name="wireless_service")
|
|
async def fixture_wireless_service(
|
|
network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
|
|
) -> DeviceWirelessService:
|
|
"""Return mock device wireless service."""
|
|
yield network_manager_services["network_device_wireless"]
|
|
|
|
|
|
async def test_load(
|
|
coresys: CoreSys,
|
|
network_manager_service: NetworkManagerService,
|
|
connection_settings_service: ConnectionSettingsService,
|
|
):
|
|
"""Test network manager load."""
|
|
network_manager_service.ActivateConnection.calls.clear()
|
|
network_manager_service.CheckConnectivity.calls.clear()
|
|
|
|
await coresys.host.network.load()
|
|
|
|
assert coresys.host.network.connectivity is True
|
|
|
|
assert len(coresys.host.network.dns_servers) == 1
|
|
assert str(coresys.host.network.dns_servers[0]) == "192.168.30.1"
|
|
|
|
assert len(coresys.host.network.interfaces) == 3
|
|
name_dict = {intr.name: intr for intr in coresys.host.network.interfaces}
|
|
assert "eth0" in name_dict
|
|
assert name_dict["eth0"].mac == "AA:BB:CC:DD:EE:FF"
|
|
assert name_dict["eth0"].enabled is True
|
|
assert name_dict["eth0"].ipv4.gateway == IPv4Address("192.168.2.1")
|
|
assert name_dict["eth0"].ipv4.ready is True
|
|
assert name_dict["eth0"].ipv4setting.method == InterfaceMethod.AUTO
|
|
assert name_dict["eth0"].ipv4setting.address == []
|
|
assert name_dict["eth0"].ipv4setting.gateway is None
|
|
assert name_dict["eth0"].ipv4setting.nameservers == [IPv4Address("192.168.2.1")]
|
|
assert name_dict["eth0"].ipv6.gateway == IPv6Address("fe80::da58:d7ff:fe00:9c69")
|
|
assert name_dict["eth0"].ipv6.ready is True
|
|
assert name_dict["eth0"].ipv6setting.method == InterfaceMethod.AUTO
|
|
assert name_dict["eth0"].ipv6setting.address == []
|
|
assert name_dict["eth0"].ipv6setting.gateway is None
|
|
assert name_dict["eth0"].ipv6setting.nameservers == [
|
|
IPv6Address("2001:4860:4860::8888")
|
|
]
|
|
assert "wlan0" in name_dict
|
|
assert name_dict["wlan0"].enabled is False
|
|
|
|
assert connection_settings_service.settings["ipv4"]["method"].value == "auto"
|
|
assert connection_settings_service.settings["ipv4"]["address-data"] == Variant(
|
|
"aa{sv}", []
|
|
)
|
|
assert "gateway" not in connection_settings_service.settings["ipv4"]
|
|
assert connection_settings_service.settings["ipv4"]["dns"] == Variant(
|
|
"au", [16951488]
|
|
)
|
|
assert connection_settings_service.settings["ipv6"]["method"].value == "auto"
|
|
assert connection_settings_service.settings["ipv6"]["address-data"] == Variant(
|
|
"aa{sv}", []
|
|
)
|
|
assert "gateway" not in connection_settings_service.settings["ipv6"]
|
|
assert connection_settings_service.settings["ipv6"]["dns"] == Variant(
|
|
"aay", [bytearray(b" \x01H`H`\x00\x00\x00\x00\x00\x00\x00\x00\x88\x88")]
|
|
)
|
|
assert "eth0.10" in name_dict
|
|
assert name_dict["eth0.10"].enabled is True
|
|
|
|
assert len(network_manager_service.ActivateConnection.calls) == 2
|
|
assert (
|
|
"/org/freedesktop/NetworkManager/Settings/38",
|
|
"/org/freedesktop/NetworkManager/Devices/38",
|
|
"/",
|
|
) in network_manager_service.ActivateConnection.calls
|
|
assert (
|
|
"/org/freedesktop/NetworkManager/Settings/1",
|
|
"/org/freedesktop/NetworkManager/Devices/1",
|
|
"/",
|
|
) in network_manager_service.ActivateConnection.calls
|
|
assert network_manager_service.CheckConnectivity.calls == []
|
|
|
|
|
|
async def test_load_with_disabled_methods(
|
|
coresys: CoreSys,
|
|
network_manager_service: NetworkManagerService,
|
|
connection_settings_service: ConnectionSettingsService,
|
|
):
|
|
"""Test load does not disable methods of interfaces."""
|
|
network_manager_service.ActivateConnection.calls.clear()
|
|
|
|
disabled = {"method": Variant("s", "disabled")}
|
|
connection_settings_service.settings = SETTINGS_1_FIXTURE | {
|
|
"ipv4": disabled,
|
|
"ipv6": disabled,
|
|
}
|
|
await coresys.dbus.network.get("eth0").settings.reload()
|
|
|
|
await coresys.host.network.load()
|
|
assert (
|
|
"/org/freedesktop/NetworkManager/Settings/1",
|
|
"/org/freedesktop/NetworkManager/Devices/1",
|
|
"/",
|
|
) not in network_manager_service.ActivateConnection.calls
|
|
|
|
|
|
async def test_load_with_network_connection_issues(
|
|
coresys: CoreSys,
|
|
network_manager_service: NetworkManagerService,
|
|
active_connection_service: ActiveConnectionService,
|
|
):
|
|
"""Test load does not update interfaces with network connection issues."""
|
|
network_manager_service.ActivateConnection.calls.clear()
|
|
|
|
active_connection_service.emit_properties_changed(
|
|
{"StateFlags": 0x10, "Ip4Config": "/"}
|
|
)
|
|
await active_connection_service.ping()
|
|
|
|
await coresys.host.network.load()
|
|
await coresys.host.network.update()
|
|
|
|
assert (
|
|
"/org/freedesktop/NetworkManager/Settings/1",
|
|
"/org/freedesktop/NetworkManager/Devices/1",
|
|
"/",
|
|
) not in network_manager_service.ActivateConnection.calls
|
|
assert len(coresys.host.network.interfaces) == 3
|
|
name_dict = {intr.name: intr for intr in coresys.host.network.interfaces}
|
|
assert "eth0" in name_dict
|
|
assert name_dict["eth0"].enabled is True
|
|
assert name_dict["eth0"].ipv4setting.method == InterfaceMethod.AUTO
|
|
assert name_dict["eth0"].ipv6setting.method == InterfaceMethod.AUTO
|
|
|
|
|
|
async def test_scan_wifi(coresys: CoreSys):
|
|
"""Test scanning wifi."""
|
|
with pytest.raises(HostNotSupportedError):
|
|
await coresys.host.network.scan_wifi(coresys.host.network.get("eth0"))
|
|
|
|
with patch("supervisor.host.network.asyncio.sleep"):
|
|
aps = await coresys.host.network.scan_wifi(coresys.host.network.get("wlan0"))
|
|
|
|
assert len(aps) == 2
|
|
assert aps[0].mac == "E4:57:40:A9:D7:DE"
|
|
assert aps[0].mode == WifiMode.INFRASTRUCTURE
|
|
assert aps[1].mac == "18:4B:0D:23:A1:9C"
|
|
assert aps[1].mode == WifiMode.INFRASTRUCTURE
|
|
|
|
|
|
async def test_scan_wifi_with_failures(
|
|
coresys: CoreSys, wireless_service: DeviceWirelessService, caplog
|
|
):
|
|
"""Test scanning wifi with accesspoint processing failures."""
|
|
wireless_service.all_access_points = [
|
|
"/org/freedesktop/NetworkManager/AccessPoint/43099",
|
|
"/org/freedesktop/NetworkManager/AccessPoint/43100",
|
|
"/org/freedesktop/NetworkManager/AccessPoint/99999",
|
|
]
|
|
|
|
with patch("supervisor.host.network.asyncio.sleep"):
|
|
aps = await coresys.host.network.scan_wifi(coresys.host.network.get("wlan0"))
|
|
|
|
assert len(aps) == 2
|
|
assert "Can't process an AP" in caplog.text
|
|
|
|
|
|
async def test_host_connectivity_changed(
|
|
coresys: CoreSys,
|
|
network_manager_service: NetworkManagerService,
|
|
ha_ws_client: AsyncMock,
|
|
):
|
|
"""Test host connectivity changed."""
|
|
await coresys.host.load()
|
|
assert coresys.host.network.connectivity is True
|
|
|
|
network_manager_service.emit_properties_changed({"Connectivity": 1})
|
|
await network_manager_service.ping()
|
|
assert coresys.host.network.connectivity is False
|
|
await asyncio.sleep(0)
|
|
assert {
|
|
"type": WSType.SUPERVISOR_EVENT,
|
|
"data": {
|
|
"event": WSEvent.SUPERVISOR_UPDATE,
|
|
"update_key": "network",
|
|
"data": {"host_internet": False},
|
|
},
|
|
} in [call.args[0] for call in ha_ws_client.async_send_command.call_args_list]
|
|
|
|
ha_ws_client.async_send_command.reset_mock()
|
|
network_manager_service.emit_properties_changed({}, ["Connectivity"])
|
|
await network_manager_service.ping()
|
|
await network_manager_service.ping()
|
|
assert coresys.host.network.connectivity is True
|
|
await asyncio.sleep(0)
|
|
assert {
|
|
"type": WSType.SUPERVISOR_EVENT,
|
|
"data": {
|
|
"event": WSEvent.SUPERVISOR_UPDATE,
|
|
"update_key": "network",
|
|
"data": {"host_internet": True},
|
|
},
|
|
} in [call.args[0] for call in ha_ws_client.async_send_command.call_args_list]
|
|
|
|
|
|
async def test_host_connectivity_disabled(
|
|
coresys: CoreSys,
|
|
network_manager_service: NetworkManagerService,
|
|
ha_ws_client: AsyncMock,
|
|
):
|
|
"""Test host connectivity check disabled."""
|
|
await coresys.host.network.load()
|
|
|
|
await coresys.core.set_state(CoreState.RUNNING)
|
|
await asyncio.sleep(0)
|
|
ha_ws_client.async_send_command.reset_mock()
|
|
|
|
assert "connectivity_check" not in coresys.resolution.unsupported
|
|
assert coresys.host.network.connectivity is True
|
|
|
|
network_manager_service.emit_properties_changed({"ConnectivityCheckEnabled": False})
|
|
await network_manager_service.ping()
|
|
assert coresys.host.network.connectivity is None
|
|
await asyncio.sleep(0)
|
|
ha_ws_client.async_send_command.assert_any_call(
|
|
{
|
|
"type": WSType.SUPERVISOR_EVENT,
|
|
"data": {
|
|
"event": WSEvent.SUPERVISOR_UPDATE,
|
|
"update_key": "network",
|
|
"data": {"host_internet": None},
|
|
},
|
|
}
|
|
)
|
|
assert "connectivity_check" in coresys.resolution.unsupported
|
|
|
|
ha_ws_client.async_send_command.reset_mock()
|
|
network_manager_service.emit_properties_changed({"ConnectivityCheckEnabled": True})
|
|
await network_manager_service.ping()
|
|
await network_manager_service.ping()
|
|
assert coresys.host.network.connectivity is True
|
|
await asyncio.sleep(0)
|
|
ha_ws_client.async_send_command.assert_any_call(
|
|
{
|
|
"type": WSType.SUPERVISOR_EVENT,
|
|
"data": {
|
|
"event": WSEvent.SUPERVISOR_UPDATE,
|
|
"update_key": "network",
|
|
"data": {"host_internet": True},
|
|
},
|
|
}
|
|
)
|
|
assert "connectivity_check" not in coresys.resolution.unsupported
|