mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-11-13 21:10:29 +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>
434 lines
16 KiB
Python
434 lines
16 KiB
Python
"""Test network API."""
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from aiohttp.test_utils import TestClient
|
|
from dbus_fast import Variant
|
|
import pytest
|
|
|
|
from supervisor.const import DOCKER_IPV4_NETWORK_MASK, DOCKER_NETWORK
|
|
from supervisor.coresys import CoreSys
|
|
|
|
from tests.const import (
|
|
TEST_INTERFACE_ETH_MAC,
|
|
TEST_INTERFACE_ETH_NAME,
|
|
TEST_INTERFACE_WLAN_NAME,
|
|
)
|
|
from tests.dbus_service_mocks.base import DBusServiceMock
|
|
from tests.dbus_service_mocks.network_connection_settings import (
|
|
ConnectionSettings as ConnectionSettingsService,
|
|
)
|
|
from tests.dbus_service_mocks.network_manager import (
|
|
NetworkManager as NetworkManagerService,
|
|
)
|
|
from tests.dbus_service_mocks.network_settings import Settings as SettingsService
|
|
|
|
|
|
async def test_api_network_info(api_client: TestClient, coresys: CoreSys):
|
|
"""Test network manager api."""
|
|
resp = await api_client.get("/network/info")
|
|
result = await resp.json()
|
|
assert TEST_INTERFACE_ETH_NAME in (
|
|
inet["interface"] for inet in result["data"]["interfaces"]
|
|
)
|
|
assert TEST_INTERFACE_WLAN_NAME in (
|
|
inet["interface"] for inet in result["data"]["interfaces"]
|
|
)
|
|
|
|
for interface in result["data"]["interfaces"]:
|
|
if interface["interface"] == TEST_INTERFACE_ETH_NAME:
|
|
assert interface["primary"]
|
|
assert interface["ipv4"]["gateway"] == "192.168.2.1"
|
|
assert interface["mac"] == "AA:BB:CC:DD:EE:FF"
|
|
if interface["interface"] == TEST_INTERFACE_WLAN_NAME:
|
|
assert not interface["primary"]
|
|
assert interface["mac"] == "FF:EE:DD:CC:BB:AA"
|
|
assert interface["ipv4"] == {
|
|
"address": [],
|
|
"gateway": None,
|
|
"method": "disabled",
|
|
"nameservers": [],
|
|
"ready": False,
|
|
}
|
|
assert interface["ipv6"] == {
|
|
"addr_gen_mode": "default",
|
|
"address": [],
|
|
"gateway": None,
|
|
"ip6_privacy": "default",
|
|
"method": "disabled",
|
|
"nameservers": [],
|
|
"ready": False,
|
|
}
|
|
|
|
assert result["data"]["docker"]["interface"] == DOCKER_NETWORK
|
|
assert result["data"]["docker"]["address"] == str(DOCKER_IPV4_NETWORK_MASK)
|
|
assert result["data"]["docker"]["dns"] == str(coresys.docker.network.dns)
|
|
assert result["data"]["docker"]["gateway"] == str(coresys.docker.network.gateway)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"interface_id", [TEST_INTERFACE_ETH_NAME, TEST_INTERFACE_ETH_MAC]
|
|
)
|
|
async def test_api_network_interface_info(api_client: TestClient, interface_id: str):
|
|
"""Test network manager api."""
|
|
resp = await api_client.get(f"/network/interface/{interface_id}/info")
|
|
result = await resp.json()
|
|
assert result["data"]["ipv4"]["address"][-1] == "192.168.2.148/24"
|
|
assert result["data"]["ipv4"]["gateway"] == "192.168.2.1"
|
|
assert result["data"]["ipv4"]["nameservers"] == ["192.168.2.2"]
|
|
assert result["data"]["ipv4"]["ready"] is True
|
|
assert (
|
|
result["data"]["ipv6"]["address"][0] == "2a03:169:3df5:0:6be9:2588:b26a:a679/64"
|
|
)
|
|
assert result["data"]["ipv6"]["address"][1] == "2a03:169:3df5::2f1/128"
|
|
assert result["data"]["ipv6"]["gateway"] == "fe80::da58:d7ff:fe00:9c69"
|
|
assert result["data"]["ipv6"]["nameservers"] == [
|
|
"2001:1620:2777:1::10",
|
|
"2001:1620:2777:2::20",
|
|
]
|
|
assert result["data"]["ipv6"]["ready"] is True
|
|
assert result["data"]["interface"] == TEST_INTERFACE_ETH_NAME
|
|
|
|
|
|
async def test_api_network_interface_info_default(api_client: TestClient):
|
|
"""Test network manager default api."""
|
|
resp = await api_client.get("/network/interface/default/info")
|
|
result = await resp.json()
|
|
assert result["data"]["ipv4"]["address"][-1] == "192.168.2.148/24"
|
|
assert result["data"]["ipv4"]["gateway"] == "192.168.2.1"
|
|
assert result["data"]["ipv4"]["nameservers"] == ["192.168.2.2"]
|
|
assert result["data"]["ipv4"]["ready"] is True
|
|
assert (
|
|
result["data"]["ipv6"]["address"][0] == "2a03:169:3df5:0:6be9:2588:b26a:a679/64"
|
|
)
|
|
assert result["data"]["ipv6"]["address"][1] == "2a03:169:3df5::2f1/128"
|
|
assert result["data"]["ipv6"]["gateway"] == "fe80::da58:d7ff:fe00:9c69"
|
|
assert result["data"]["ipv6"]["nameservers"] == [
|
|
"2001:1620:2777:1::10",
|
|
"2001:1620:2777:2::20",
|
|
]
|
|
assert result["data"]["ipv6"]["ready"] is True
|
|
assert result["data"]["interface"] == TEST_INTERFACE_ETH_NAME
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"interface_id", [TEST_INTERFACE_ETH_NAME, TEST_INTERFACE_ETH_MAC]
|
|
)
|
|
async def test_api_network_interface_update_mac_or_name(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
network_manager_service: NetworkManagerService,
|
|
connection_settings_service: ConnectionSettingsService,
|
|
interface_id: str,
|
|
):
|
|
"""Test network manager API update with name or MAC address."""
|
|
network_manager_service.CheckConnectivity.calls.clear()
|
|
connection_settings_service.Update.calls.clear()
|
|
assert coresys.dbus.network.get(interface_id).settings.ipv4.method == "auto"
|
|
|
|
resp = await api_client.post(
|
|
f"/network/interface/{interface_id}/update",
|
|
json={
|
|
"ipv4": {
|
|
"method": "static",
|
|
"nameservers": ["1.1.1.1"],
|
|
"address": ["192.168.2.148/24"],
|
|
"gateway": "192.168.1.1",
|
|
}
|
|
},
|
|
)
|
|
result = await resp.json()
|
|
assert result["result"] == "ok"
|
|
assert network_manager_service.CheckConnectivity.calls == [()]
|
|
assert len(connection_settings_service.Update.calls) == 1
|
|
|
|
await connection_settings_service.ping()
|
|
assert (
|
|
coresys.dbus.network.get(TEST_INTERFACE_ETH_NAME).settings.ipv4.method
|
|
== "manual"
|
|
)
|
|
|
|
|
|
async def test_api_network_interface_update_ethernet(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
network_manager_service: NetworkManagerService,
|
|
connection_settings_service: ConnectionSettingsService,
|
|
):
|
|
"""Test network manager API update with name or MAC address."""
|
|
network_manager_service.CheckConnectivity.calls.clear()
|
|
connection_settings_service.Update.calls.clear()
|
|
|
|
# Full static configuration (represents frontend static config)
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update",
|
|
json={
|
|
"ipv4": {
|
|
"method": "static",
|
|
"nameservers": ["1.1.1.1"],
|
|
"address": ["192.168.2.148/24"],
|
|
"gateway": "192.168.2.1",
|
|
}
|
|
},
|
|
)
|
|
result = await resp.json()
|
|
assert result["result"] == "ok"
|
|
assert network_manager_service.CheckConnectivity.calls == [()]
|
|
assert len(connection_settings_service.Update.calls) == 1
|
|
settings = connection_settings_service.Update.calls[0][0]
|
|
|
|
assert "ipv4" in settings
|
|
assert settings["ipv4"]["method"] == Variant("s", "manual")
|
|
assert settings["ipv4"]["address-data"] == Variant(
|
|
"aa{sv}",
|
|
[{"address": Variant("s", "192.168.2.148"), "prefix": Variant("u", 24)}],
|
|
)
|
|
assert settings["ipv4"]["dns"] == Variant("au", [16843009])
|
|
assert settings["ipv4"]["gateway"] == Variant("s", "192.168.2.1")
|
|
|
|
# Partial static configuration, clears other settings (e.g. by CLI)
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update",
|
|
json={
|
|
"ipv4": {
|
|
"method": "static",
|
|
"address": ["192.168.2.149/24"],
|
|
}
|
|
},
|
|
)
|
|
result = await resp.json()
|
|
assert result["result"] == "ok"
|
|
assert len(connection_settings_service.Update.calls) == 2
|
|
settings = connection_settings_service.Update.calls[1][0]
|
|
|
|
assert "ipv4" in settings
|
|
assert settings["ipv4"]["method"] == Variant("s", "manual")
|
|
assert settings["ipv4"]["address-data"] == Variant(
|
|
"aa{sv}",
|
|
[{"address": Variant("s", "192.168.2.149"), "prefix": Variant("u", 24)}],
|
|
)
|
|
assert "dns" not in settings["ipv4"]
|
|
assert "gateway" not in settings["ipv4"]
|
|
|
|
# Auto configuration, clears all settings (represents frontend auto config)
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update",
|
|
json={
|
|
"ipv4": {
|
|
"method": "auto",
|
|
"nameservers": ["8.8.8.8"],
|
|
}
|
|
},
|
|
)
|
|
result = await resp.json()
|
|
assert result["result"] == "ok"
|
|
assert len(connection_settings_service.Update.calls) == 3
|
|
settings = connection_settings_service.Update.calls[2][0]
|
|
|
|
# Validate network update to auto clears address, DNS and gateway settings
|
|
assert "ipv4" in settings
|
|
assert settings["ipv4"]["method"] == Variant("s", "auto")
|
|
assert "address-data" not in settings["ipv4"]
|
|
assert "addresses" not in settings["ipv4"]
|
|
assert "dns-data" not in settings["ipv4"]
|
|
assert settings["ipv4"]["dns"] == Variant("au", [134744072])
|
|
assert "gateway" not in settings["ipv4"]
|
|
|
|
|
|
async def test_api_network_interface_update_wifi(api_client: TestClient):
|
|
"""Test network interface WiFi API."""
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_WLAN_NAME}/update",
|
|
json={
|
|
"enabled": True,
|
|
"ipv4": {
|
|
"method": "static",
|
|
"nameservers": ["1.1.1.1"],
|
|
"address": ["192.168.2.148/24"],
|
|
"gateway": "192.168.1.1",
|
|
},
|
|
"wifi": {"ssid": "MY_TEST", "auth": "wpa-psk", "psk": "myWifiPassword"},
|
|
},
|
|
)
|
|
result = await resp.json()
|
|
assert result["result"] == "ok"
|
|
|
|
|
|
async def test_api_network_interface_update_wifi_error(api_client: TestClient):
|
|
"""Test network interface WiFi API error handling."""
|
|
# Simulate frontend WiFi interface edit where the user did not select
|
|
# a WiFi SSID.
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_WLAN_NAME}/update",
|
|
json={
|
|
"enabled": True,
|
|
"ipv4": {
|
|
"method": "auto",
|
|
},
|
|
"ipv6": {
|
|
"method": "auto",
|
|
},
|
|
},
|
|
)
|
|
result = await resp.json()
|
|
assert result["result"] == "error"
|
|
assert (
|
|
result["message"]
|
|
== "Can't create config and activate wlan0: A 'wireless' setting with a valid SSID is required if no AP path was given."
|
|
)
|
|
|
|
|
|
async def test_api_network_interface_update_remove(api_client: TestClient):
|
|
"""Test network manager api."""
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update",
|
|
json={"enabled": False},
|
|
)
|
|
result = await resp.json()
|
|
assert result["result"] == "ok"
|
|
|
|
|
|
async def test_api_network_interface_info_invalid(api_client: TestClient):
|
|
"""Test network manager api."""
|
|
resp = await api_client.get("/network/interface/invalid/info")
|
|
result = await resp.json()
|
|
|
|
assert result["message"]
|
|
assert result["result"] == "error"
|
|
|
|
|
|
async def test_api_network_interface_update_invalid(api_client: TestClient):
|
|
"""Test network manager api."""
|
|
resp = await api_client.post("/network/interface/invalid/update", json={})
|
|
result = await resp.json()
|
|
assert result["message"] == "Interface invalid does not exist"
|
|
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update", json={}
|
|
)
|
|
result = await resp.json()
|
|
assert result["message"] == "You need to supply at least one option to update"
|
|
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update",
|
|
json={"ipv4": {"nameservers": "1.1.1.1"}},
|
|
)
|
|
result = await resp.json()
|
|
assert (
|
|
result["message"]
|
|
== "expected a list for dictionary value @ data['ipv4']['nameservers']. Got '1.1.1.1'"
|
|
)
|
|
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update",
|
|
json={"ipv4": {"gateway": "invalid"}},
|
|
)
|
|
result = await resp.json()
|
|
assert (
|
|
result["message"]
|
|
== "expected IPv4Address for dictionary value @ data['ipv4']['gateway']. Got 'invalid'"
|
|
)
|
|
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update",
|
|
json={"ipv6": {"gateway": "invalid"}},
|
|
)
|
|
result = await resp.json()
|
|
assert (
|
|
result["message"]
|
|
== "expected IPv6Address for dictionary value @ data['ipv6']['gateway']. Got 'invalid'"
|
|
)
|
|
|
|
|
|
async def test_api_network_wireless_scan(api_client: TestClient):
|
|
"""Test network manager api."""
|
|
with patch("asyncio.sleep", return_value=AsyncMock()):
|
|
resp = await api_client.get(
|
|
f"/network/interface/{TEST_INTERFACE_WLAN_NAME}/accesspoints"
|
|
)
|
|
result = await resp.json()
|
|
|
|
assert [ap["ssid"] for ap in result["data"]["accesspoints"]] == [
|
|
"UPC4814466",
|
|
"VQ@35(55720",
|
|
]
|
|
assert [ap["signal"] for ap in result["data"]["accesspoints"]] == [47, 63]
|
|
|
|
|
|
async def test_api_network_reload(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
network_manager_service: NetworkManagerService,
|
|
):
|
|
"""Test network manager reload api."""
|
|
network_manager_service.CheckConnectivity.calls.clear()
|
|
resp = await api_client.post("/network/reload")
|
|
result = await resp.json()
|
|
|
|
assert result["result"] == "ok"
|
|
# Check that we forced NM to do an immediate connectivity check
|
|
assert network_manager_service.CheckConnectivity.calls == [()]
|
|
|
|
|
|
async def test_api_network_vlan(
|
|
api_client: TestClient,
|
|
coresys: CoreSys,
|
|
network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
|
|
):
|
|
"""Test creating a vlan."""
|
|
settings_service: SettingsService = network_manager_services["network_settings"]
|
|
settings_service.AddConnection.calls.clear()
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/vlan/1",
|
|
json={"ipv4": {"method": "auto"}},
|
|
)
|
|
result = await resp.json()
|
|
assert result["result"] == "ok"
|
|
assert len(settings_service.AddConnection.calls) == 1
|
|
|
|
connection = settings_service.AddConnection.calls[0][0]
|
|
assert "uuid" in connection["connection"]
|
|
assert connection["connection"] == {
|
|
"id": Variant("s", "Supervisor eth0.1"),
|
|
"type": Variant("s", "vlan"),
|
|
"llmnr": Variant("i", 2),
|
|
"mdns": Variant("i", 2),
|
|
"autoconnect": Variant("b", True),
|
|
"uuid": connection["connection"]["uuid"],
|
|
}
|
|
assert connection["ipv4"] == {"method": Variant("s", "auto")}
|
|
assert connection["ipv6"] == {"method": Variant("s", "auto")}
|
|
assert connection["vlan"] == {
|
|
"id": Variant("u", 1),
|
|
"parent": Variant("s", "0c23631e-2118-355c-bbb0-8943229cb0d6"),
|
|
}
|
|
|
|
# Check if trying to recreate an existing VLAN raises an exception
|
|
result = await resp.json()
|
|
resp = await api_client.post(
|
|
f"/network/interface/{TEST_INTERFACE_ETH_NAME}/vlan/10",
|
|
json={"ipv4": {"method": "auto"}},
|
|
)
|
|
result = await resp.json()
|
|
assert result["result"] == "error"
|
|
assert len(settings_service.AddConnection.calls) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "url"),
|
|
[
|
|
("get", "/network/interface/bad/info"),
|
|
("post", "/network/interface/bad/update"),
|
|
("get", "/network/interface/bad/accesspoints"),
|
|
("post", "/network/interface/bad/vlan/1"),
|
|
],
|
|
)
|
|
async def test_network_interface_not_found(
|
|
api_client: TestClient, method: str, url: str
|
|
):
|
|
"""Test network interface not found error."""
|
|
resp = await api_client.request(method, url)
|
|
assert resp.status == 404
|
|
body = await resp.json()
|
|
assert body["message"] == "Interface bad does not exist"
|