diff --git a/tests/api/test_network.py b/tests/api/test_network.py index 5b5f60945..1eb8dd3d9 100644 --- a/tests/api/test_network.py +++ b/tests/api/test_network.py @@ -1,4 +1,4 @@ -"""Test NetwrokInterface API.""" +"""Test network API.""" from unittest.mock import AsyncMock, patch @@ -9,7 +9,11 @@ import pytest from supervisor.const import DOCKER_NETWORK, DOCKER_NETWORK_MASK from supervisor.coresys import CoreSys -from tests.const import TEST_INTERFACE, TEST_INTERFACE_WLAN +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, @@ -24,19 +28,19 @@ 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 in ( + assert TEST_INTERFACE_ETH_NAME in ( inet["interface"] for inet in result["data"]["interfaces"] ) - assert TEST_INTERFACE_WLAN in ( + 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: + 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: + if interface["interface"] == TEST_INTERFACE_WLAN_NAME: assert not interface["primary"] assert interface["mac"] == "FF:EE:DD:CC:BB:AA" assert interface["ipv4"] == { @@ -60,10 +64,12 @@ async def test_api_network_info(api_client: TestClient, coresys: CoreSys): assert result["data"]["docker"]["gateway"] == str(coresys.docker.network.gateway) -@pytest.mark.parametrize("intr_id", [TEST_INTERFACE, "AA:BB:CC:DD:EE:FF"]) -async def test_api_network_interface_info(api_client: TestClient, intr_id: str): +@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/{intr_id}/info") + 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" @@ -79,7 +85,7 @@ async def test_api_network_interface_info(api_client: TestClient, intr_id: str): "2001:1620:2777:2::20", ] assert result["data"]["ipv6"]["ready"] is True - assert result["data"]["interface"] == TEST_INTERFACE + assert result["data"]["interface"] == TEST_INTERFACE_ETH_NAME async def test_api_network_interface_info_default(api_client: TestClient): @@ -100,24 +106,26 @@ async def test_api_network_interface_info_default(api_client: TestClient): "2001:1620:2777:2::20", ] assert result["data"]["ipv6"]["ready"] is True - assert result["data"]["interface"] == TEST_INTERFACE + assert result["data"]["interface"] == TEST_INTERFACE_ETH_NAME -@pytest.mark.parametrize("intr_id", [TEST_INTERFACE, "AA:BB:CC:DD:EE:FF"]) -async def test_api_network_interface_update( +@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, - intr_id: str, + interface_id: str, ): - """Test network manager api.""" + """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(TEST_INTERFACE).settings.ipv4.method == "auto" + assert coresys.dbus.network.get(interface_id).settings.ipv4.method == "auto" resp = await api_client.post( - f"/network/interface/{intr_id}/update", + f"/network/interface/{interface_id}/update", json={ "ipv4": { "method": "static", @@ -133,14 +141,101 @@ async def test_api_network_interface_update( assert len(connection_settings_service.Update.calls) == 1 await connection_settings_service.ping() - await connection_settings_service.ping() - assert coresys.dbus.network.get(TEST_INTERFACE).settings.ipv4.method == "manual" + 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, updates only provided 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 settings["ipv4"]["dns"] == Variant("au", [16843009]) + assert settings["ipv4"]["gateway"] == Variant("s", "192.168.2.1") + + # 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", + } + }, + ) + 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 "dns" not in settings["ipv4"] + assert "gateway" not in settings["ipv4"] async def test_api_network_interface_update_wifi(api_client: TestClient): """Test network manager api.""" resp = await api_client.post( - f"/network/interface/{TEST_INTERFACE_WLAN}/update", + f"/network/interface/{TEST_INTERFACE_WLAN_NAME}/update", json={ "enabled": True, "ipv4": { @@ -159,7 +254,7 @@ async def test_api_network_interface_update_wifi(api_client: TestClient): 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}/update", + f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update", json={"enabled": False}, ) result = await resp.json() @@ -181,12 +276,14 @@ async def test_api_network_interface_update_invalid(api_client: TestClient): result = await resp.json() assert result["message"] == "Interface invalid does not exist" - resp = await api_client.post(f"/network/interface/{TEST_INTERFACE}/update", json={}) + 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}/update", + f"/network/interface/{TEST_INTERFACE_ETH_NAME}/update", json={"ipv4": {"nameservers": "1.1.1.1"}}, ) result = await resp.json() @@ -200,7 +297,7 @@ 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}/accesspoints" + f"/network/interface/{TEST_INTERFACE_WLAN_NAME}/accesspoints" ) result = await resp.json() @@ -235,7 +332,8 @@ async def test_api_network_vlan( settings_service: SettingsService = network_manager_services["network_settings"] settings_service.AddConnection.calls.clear() resp = await api_client.post( - f"/network/interface/{TEST_INTERFACE}/vlan/1", json={"ipv4": {"method": "auto"}} + f"/network/interface/{TEST_INTERFACE_ETH_NAME}/vlan/1", + json={"ipv4": {"method": "auto"}}, ) result = await resp.json() assert result["result"] == "ok" diff --git a/tests/const.py b/tests/const.py index 085d73759..1f0fe2d36 100644 --- a/tests/const.py +++ b/tests/const.py @@ -1,7 +1,8 @@ """Consts for tests.""" -TEST_INTERFACE = "eth0" -TEST_INTERFACE_WLAN = "wlan0" +TEST_INTERFACE_ETH_NAME = "eth0" +TEST_INTERFACE_ETH_MAC = "AA:BB:CC:DD:EE:FF" +TEST_INTERFACE_WLAN_NAME = "wlan0" TEST_WS_URL = "ws://test.org:3000" TEST_ADDON_SLUG = "local_ssh" diff --git a/tests/dbus/network/setting/test_generate.py b/tests/dbus/network/setting/test_generate.py index 58c7b32ad..bd6eb0181 100644 --- a/tests/dbus/network/setting/test_generate.py +++ b/tests/dbus/network/setting/test_generate.py @@ -9,12 +9,12 @@ from supervisor.host.configuration import IpConfig, IpSetting, VlanConfig from supervisor.host.const import InterfaceMethod, InterfaceType from supervisor.host.network import Interface -from tests.const import TEST_INTERFACE +from tests.const import TEST_INTERFACE_ETH_NAME async def test_get_connection_from_interface(network_manager: NetworkManager): """Test network interface.""" - dbus_interface = network_manager.get(TEST_INTERFACE) + dbus_interface = network_manager.get(TEST_INTERFACE_ETH_NAME) interface = Interface.from_dbus_interface(dbus_interface) connection_payload = get_connection_from_interface(interface, network_manager) @@ -33,7 +33,7 @@ async def test_get_connection_from_interface(network_manager: NetworkManager): async def test_get_connection_no_path(network_manager: NetworkManager): """Test network interface without a path.""" - dbus_interface = network_manager.get(TEST_INTERFACE) + dbus_interface = network_manager.get(TEST_INTERFACE_ETH_NAME) with patch.object(NetworkInterface, "path", new=PropertyMock(return_value=None)): interface = Interface.from_dbus_interface(dbus_interface) diff --git a/tests/dbus/network/setting/test_init.py b/tests/dbus/network/setting/test_init.py index 7066c7195..e48bda93e 100644 --- a/tests/dbus/network/setting/test_init.py +++ b/tests/dbus/network/setting/test_init.py @@ -16,6 +16,10 @@ 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_device import ( + ETHERNET_DEVICE_OBJECT_PATH, + WIRELESS_DEVICE_OBJECT_PATH, +) @pytest.fixture(name="connection_settings_service", autouse=True) @@ -27,14 +31,21 @@ async def fixture_connection_settings_service( @pytest.fixture(name="dbus_interface") -async def fixture_dbus_interface(dbus_session_bus: MessageBus) -> NetworkInterface: +async def fixture_dbus_interface( + dbus_session_bus: MessageBus, device_object_path: str = ETHERNET_DEVICE_OBJECT_PATH +) -> NetworkInterface: """Get connected dbus interface.""" - dbus_interface = NetworkInterface("/org/freedesktop/NetworkManager/Devices/1") + dbus_interface = NetworkInterface(device_object_path) await dbus_interface.connect(dbus_session_bus) yield dbus_interface -async def test_update( +@pytest.mark.parametrize( + "dbus_interface", + [ETHERNET_DEVICE_OBJECT_PATH, WIRELESS_DEVICE_OBJECT_PATH], + indirect=True, +) +async def test_ethernet_update( dbus_interface: NetworkInterface, connection_settings_service: ConnectionSettingsService, ): @@ -91,17 +102,23 @@ async def test_update( assert "proxy" in settings - assert "802-3-ethernet" in settings - assert settings["802-3-ethernet"]["auto-negotiate"] == Variant("b", False) - - assert "802-11-wireless" in settings - assert settings["802-11-wireless"]["ssid"] == Variant("ay", b"NETT") - assert "mode" not in settings["802-11-wireless"] - assert "powersave" not in settings["802-11-wireless"] - - assert "802-11-wireless-security" not in settings assert "vlan" not in settings + if settings["connection"]["type"] == "802-3-ethernet": + assert "802-3-ethernet" in settings + assert settings["802-3-ethernet"]["auto-negotiate"] == Variant("b", False) + + assert "802-11-wireless" not in settings + assert "802-11-wireless-security" not in settings + + if settings["connection"]["type"] == "802-11-wireless": + assert "802-11-wireless" in settings + assert settings["802-11-wireless"]["ssid"] == Variant("ay", b"NETT") + assert "mode" not in settings["802-11-wireless"] + assert "powersave" not in settings["802-11-wireless"] + + assert "802-11-wireless-security" not in settings + async def test_ipv6_disabled_is_link_local(dbus_interface: NetworkInterface): """Test disabled equals link local for ipv6.""" diff --git a/tests/dbus/network/test_connection.py b/tests/dbus/network/test_connection.py index 36e0370aa..24598ca52 100644 --- a/tests/dbus/network/test_connection.py +++ b/tests/dbus/network/test_connection.py @@ -7,7 +7,7 @@ from supervisor.dbus.const import ConnectionStateFlags from supervisor.dbus.network import NetworkManager from supervisor.dbus.network.connection import NetworkConnection -from tests.const import TEST_INTERFACE +from tests.const import TEST_INTERFACE_ETH_NAME from tests.dbus_service_mocks.base import DBusServiceMock from tests.dbus_service_mocks.network_active_connection import ( ActiveConnection as ActiveConnectionService, @@ -57,7 +57,7 @@ async def test_old_ipv4_disconnect( network_manager: NetworkManager, active_connection_service: ActiveConnectionService ): """Test old ipv4 disconnects on ipv4 change.""" - connection = network_manager.get(TEST_INTERFACE).connection + connection = network_manager.get(TEST_INTERFACE_ETH_NAME).connection ipv4 = connection.ipv4 assert ipv4.is_connected is True @@ -72,7 +72,7 @@ async def test_old_ipv6_disconnect( network_manager: NetworkManager, active_connection_service: ActiveConnectionService ): """Test old ipv6 disconnects on ipv6 change.""" - connection = network_manager.get(TEST_INTERFACE).connection + connection = network_manager.get(TEST_INTERFACE_ETH_NAME).connection ipv6 = connection.ipv6 assert ipv6.is_connected is True @@ -87,7 +87,7 @@ async def test_old_settings_disconnect( network_manager: NetworkManager, active_connection_service: ActiveConnectionService ): """Test old settings disconnects on settings change.""" - connection = network_manager.get(TEST_INTERFACE).connection + connection = network_manager.get(TEST_INTERFACE_ETH_NAME).connection settings = connection.settings assert settings.is_connected is True diff --git a/tests/dbus/network/test_interface.py b/tests/dbus/network/test_interface.py index 93074f95d..ef81e2f38 100644 --- a/tests/dbus/network/test_interface.py +++ b/tests/dbus/network/test_interface.py @@ -10,7 +10,7 @@ from supervisor.dbus.network import NetworkManager from supervisor.dbus.network.interface import NetworkInterface from tests.common import mock_dbus_services -from tests.const import TEST_INTERFACE, TEST_INTERFACE_WLAN +from tests.const import TEST_INTERFACE_ETH_NAME, TEST_INTERFACE_WLAN_NAME from tests.dbus_service_mocks.base import DBusServiceMock from tests.dbus_service_mocks.network_device import Device as DeviceService @@ -61,7 +61,7 @@ async def test_network_interface_ethernet( await interface.connect(dbus_session_bus) assert interface.sync_properties is True - assert interface.name == TEST_INTERFACE + assert interface.name == TEST_INTERFACE_ETH_NAME assert interface.type == DeviceType.ETHERNET assert interface.managed is True assert interface.wireless is None @@ -108,7 +108,7 @@ async def test_network_interface_wlan( await interface.connect(dbus_session_bus) assert interface.sync_properties is True - assert interface.name == TEST_INTERFACE_WLAN + assert interface.name == TEST_INTERFACE_WLAN_NAME assert interface.type == DeviceType.WIRELESS assert interface.wireless is not None assert interface.wireless.bitrate == 0 @@ -118,7 +118,7 @@ async def test_old_connection_disconnect( network_manager: NetworkManager, device_eth0_service: DeviceService ): """Test old connection disconnects on connection change.""" - interface = network_manager.get(TEST_INTERFACE) + interface = network_manager.get(TEST_INTERFACE_ETH_NAME) connection = interface.connection assert connection.is_connected is True @@ -133,7 +133,7 @@ async def test_old_wireless_disconnect( network_manager: NetworkManager, device_wlan0_service: DeviceService ): """Test old wireless disconnects on type change.""" - interface = network_manager.get(TEST_INTERFACE_WLAN) + interface = network_manager.get(TEST_INTERFACE_WLAN_NAME) wireless = interface.wireless assert wireless.is_connected is True @@ -167,9 +167,9 @@ async def test_interface_becomes_unmanaged( device_wlan0_service: DeviceService, ): """Test managed objects disconnect when interface becomes unmanaged.""" - eth0 = network_manager.get(TEST_INTERFACE) + eth0 = network_manager.get(TEST_INTERFACE_ETH_NAME) connection = eth0.connection - wlan0 = network_manager.get(TEST_INTERFACE_WLAN) + wlan0 = network_manager.get(TEST_INTERFACE_WLAN_NAME) wireless = wlan0.wireless assert connection.is_connected is True diff --git a/tests/dbus/network/test_network_manager.py b/tests/dbus/network/test_network_manager.py index 0ef1a30ac..4a0da3262 100644 --- a/tests/dbus/network/test_network_manager.py +++ b/tests/dbus/network/test_network_manager.py @@ -17,9 +17,9 @@ from supervisor.exceptions import ( ) from supervisor.utils.dbus import DBus -from tests.const import TEST_INTERFACE, TEST_INTERFACE_WLAN +from tests.const import 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 SETTINGS_FIXTURE +from tests.dbus_service_mocks.network_connection_settings import SETTINGS_1_FIXTURE from tests.dbus_service_mocks.network_manager import ( NetworkManager as NetworkManagerService, ) @@ -43,7 +43,7 @@ async def test_network_manager( await network_manager.connect(dbus_session_bus) - assert TEST_INTERFACE in network_manager + assert TEST_INTERFACE_ETH_NAME in network_manager assert network_manager.connectivity_enabled is True network_manager_service.emit_properties_changed({"ConnectivityCheckEnabled": False}) @@ -113,7 +113,7 @@ async def test_add_and_activate_connection( network_manager_service.AddAndActivateConnection.calls.clear() settings, connection = await network_manager.add_and_activate_connection( - SETTINGS_FIXTURE, "/org/freedesktop/NetworkManager/Devices/1" + SETTINGS_1_FIXTURE, "/org/freedesktop/NetworkManager/Devices/1" ) assert settings.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6" assert settings.ipv4.method == "auto" @@ -122,7 +122,7 @@ async def test_add_and_activate_connection( connection.settings.object_path == "/org/freedesktop/NetworkManager/Settings/1" ) assert network_manager_service.AddAndActivateConnection.calls == [ - (SETTINGS_FIXTURE, "/org/freedesktop/NetworkManager/Devices/1", "/") + (SETTINGS_1_FIXTURE, "/org/freedesktop/NetworkManager/Devices/1", "/") ] @@ -130,13 +130,13 @@ async def test_removed_devices_disconnect( network_manager_service: NetworkManagerService, network_manager: NetworkManager ): """Test removed devices are disconnected.""" - wlan = network_manager.get(TEST_INTERFACE_WLAN) + wlan = network_manager.get(TEST_INTERFACE_WLAN_NAME) assert wlan.is_connected is True network_manager_service.emit_properties_changed({"Devices": []}) await network_manager_service.ping() - assert TEST_INTERFACE_WLAN not in network_manager + assert TEST_INTERFACE_WLAN_NAME not in network_manager assert wlan.is_connected is False diff --git a/tests/dbus/network/test_settings.py b/tests/dbus/network/test_settings.py index 51b7fae55..e0fa62550 100644 --- a/tests/dbus/network/test_settings.py +++ b/tests/dbus/network/test_settings.py @@ -7,7 +7,7 @@ from supervisor.dbus.network.settings import NetworkManagerSettings from supervisor.exceptions import DBusNotConnectedError from tests.common import mock_dbus_services -from tests.dbus_service_mocks.network_connection_settings import SETTINGS_FIXTURE +from tests.dbus_service_mocks.network_connection_settings import SETTINGS_1_FIXTURE from tests.dbus_service_mocks.network_settings import Settings as SettingsService @@ -30,15 +30,15 @@ async def test_add_connection( settings = NetworkManagerSettings() with pytest.raises(DBusNotConnectedError): - await settings.add_connection(SETTINGS_FIXTURE) + await settings.add_connection(SETTINGS_1_FIXTURE) await settings.connect(dbus_session_bus) - connection_settings = await settings.add_connection(SETTINGS_FIXTURE) + connection_settings = await settings.add_connection(SETTINGS_1_FIXTURE) assert connection_settings.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6" assert connection_settings.ipv4.method == "auto" - assert settings_service.AddConnection.calls == [(SETTINGS_FIXTURE,)] + assert settings_service.AddConnection.calls == [(SETTINGS_1_FIXTURE,)] async def test_reload_connections( diff --git a/tests/dbus/network/test_wireless.py b/tests/dbus/network/test_wireless.py index 3fe1ddd21..0d9d93499 100644 --- a/tests/dbus/network/test_wireless.py +++ b/tests/dbus/network/test_wireless.py @@ -6,7 +6,7 @@ import pytest from supervisor.dbus.network import NetworkManager from supervisor.dbus.network.wireless import NetworkWireless -from tests.const import TEST_INTERFACE_WLAN +from tests.const import TEST_INTERFACE_WLAN_NAME from tests.dbus_service_mocks.base import DBusServiceMock from tests.dbus_service_mocks.network_device_wireless import ( DeviceWireless as DeviceWirelessService, @@ -56,7 +56,8 @@ async def test_request_scan( """Test request scan.""" device_wireless_service.RequestScan.calls.clear() assert ( - await network_manager.get(TEST_INTERFACE_WLAN).wireless.request_scan() is None + await network_manager.get(TEST_INTERFACE_WLAN_NAME).wireless.request_scan() + is None ) assert device_wireless_service.RequestScan.calls == [({},)] @@ -64,7 +65,7 @@ async def test_request_scan( async def test_get_all_access_points(network_manager: NetworkManager): """Test get all access points.""" accesspoints = await network_manager.get( - TEST_INTERFACE_WLAN + TEST_INTERFACE_WLAN_NAME ).wireless.get_all_accesspoints() assert len(accesspoints) == 2 assert accesspoints[0].mac == "E4:57:40:A9:D7:DE" @@ -75,7 +76,7 @@ async def test_get_all_access_points(network_manager: NetworkManager): async def test_old_active_ap_disconnects(network_manager: NetworkManager): """Test old access point disconnects on active ap change.""" - wireless = network_manager.get(TEST_INTERFACE_WLAN).wireless + wireless = network_manager.get(TEST_INTERFACE_WLAN_NAME).wireless await wireless.update( {"ActiveAccessPoint": "/org/freedesktop/NetworkManager/AccessPoint/43099"} diff --git a/tests/dbus_service_mocks/network_active_connection.py b/tests/dbus_service_mocks/network_active_connection.py index c7263968d..8b1b61fa1 100644 --- a/tests/dbus_service_mocks/network_active_connection.py +++ b/tests/dbus_service_mocks/network_active_connection.py @@ -48,6 +48,12 @@ FIXTURES: dict[str, ActiveConnectionFixture] = { "/org/freedesktop/NetworkManager/Devices/5", ], ), + "/org/freedesktop/NetworkManager/ActiveConnection/3": ActiveConnectionFixture( + connection="/org/freedesktop/NetworkManager/Settings/3", + devices=[ + "/org/freedesktop/NetworkManager/Devices/3", + ], + ), } diff --git a/tests/dbus_service_mocks/network_connection_settings.py b/tests/dbus_service_mocks/network_connection_settings.py index 4ee601f18..7d7efb814 100644 --- a/tests/dbus_service_mocks/network_connection_settings.py +++ b/tests/dbus_service_mocks/network_connection_settings.py @@ -1,6 +1,10 @@ """Mock of Network Manager Connection Settings service.""" -from dbus_fast import Variant +from copy import deepcopy +from ipaddress import IPv4Address, IPv6Address +import socket + +from dbus_fast import DBusError, Variant from dbus_fast.service import PropertyAccess, dbus_property, signal from .base import DBusServiceMock, dbus_method @@ -8,44 +12,20 @@ from .base import DBusServiceMock, dbus_method BUS_NAME = "org.freedesktop.NetworkManager" DEFAULT_OBJECT_PATH = "/org/freedesktop/NetworkManager/Settings/1" -SETTINGS_FIXTURE: dict[str, dict[str, Variant]] = { - "connection": { - "id": Variant("s", "Wired connection 1"), - "interface-name": Variant("s", "eth0"), - "llmnr": Variant("i", 2), - "mdns": Variant("i", 2), - "permissions": Variant("as", []), - "timestamp": Variant("t", 1598125548), - "type": Variant("s", "802-3-ethernet"), - "uuid": Variant("s", "0c23631e-2118-355c-bbb0-8943229cb0d6"), - }, +# NetworkManager Connection settings skeleton which gets generated automatically +# Created with 1.42.4, using: +# nmcli con add type ethernet con-name "Test" +# busctl call org.freedesktop.NetworkManager /org/freedesktop/NetworkManager/Settings/5 org.freedesktop.NetworkManager.Settings.Connection GetSettings --json=pretty +# Note that "id" and "type" seem to be the bare minimum an update call, so they can be +# ommitted here. +MINIMAL_SETTINGS_FIXTURE = { "ipv4": { - "address-data": Variant( - "aa{sv}", - [ - { - "address": Variant("s", "192.168.2.148"), - "prefix": Variant("u", 24), - } - ], - ), - "addresses": Variant("aau", [[2483202240, 24, 16951488]]), - "dns": Variant("au", [16951488]), - "dns-data": Variant("as", ["192.168.2.1"]), + "address-data": Variant("aa{sv}", []), + "addresses": Variant("aau", []), "dns-search": Variant("as", []), - "gateway": Variant("s", "192.168.2.1"), "method": Variant("s", "auto"), - "route-data": Variant( - "aa{sv}", - [ - { - "dest": Variant("s", "192.168.122.0"), - "prefix": Variant("u", 24), - "next-hop": Variant("s", "10.10.10.1"), - } - ], - ), - "routes": Variant("aau", [[8038592, 24, 17435146, 0]]), + "route-data": Variant("aa{sv}", []), + "routes": Variant("aau", []), }, "ipv6": { "address-data": Variant("aa{sv}", []), @@ -54,32 +34,118 @@ SETTINGS_FIXTURE: dict[str, dict[str, Variant]] = { "method": Variant("s", "auto"), "route-data": Variant("aa{sv}", []), "routes": Variant("a(ayuayu)", []), - "addr-gen-mode": Variant("i", 0), }, "proxy": {}, +} + +MINIMAL_ETHERNET_SETTINGS_FIXTURE = MINIMAL_SETTINGS_FIXTURE | { + "connection": { + "permissions": Variant("as", []), + "uuid": Variant("s", "ee736ea0-e2cc-4cc5-9c35-d6df94a56b47"), + }, "802-3-ethernet": { - "assigned-mac-address": Variant("s", "preserve"), "auto-negotiate": Variant("b", False), "mac-address-blacklist": Variant("as", []), "s390-options": Variant("a{ss}", {}), }, - "802-11-wireless": {"ssid": Variant("ay", b"NETT")}, } -SETINGS_FIXTURES: dict[str, dict[str, dict[str, Variant]]] = { - "/org/freedesktop/NetworkManager/Settings/1": SETTINGS_FIXTURE, - "/org/freedesktop/NetworkManager/Settings/2": { + +MINIMAL_WIRELESS_SETTINGS_FIXTURE = MINIMAL_SETTINGS_FIXTURE | { + "connection": { + "permissions": Variant("as", []), + "uuid": Variant("s", "bf9f098a-23f5-41b0-873b-b449c58df499"), + }, + "802-11-wireless": { + "mac-address-blacklist": Variant("as", []), + "seen-bssids": Variant("as", []), + "ssid": Variant("ay", b"TestSSID"), + }, +} + + +def settings_update(minimal_setting, new_settings): + """Update Connection settings with minimal skeleton in mind.""" + settings = deepcopy(minimal_setting) + for k, v in new_settings.items(): + if k in settings: + settings[k].update(v) + else: + settings[k] = v + return settings + + +SETTINGS_1_FIXTURE: dict[str, dict[str, Variant]] = settings_update( + MINIMAL_ETHERNET_SETTINGS_FIXTURE, + { + "connection": { + "id": Variant("s", "Wired connection 1"), + "interface-name": Variant("s", "eth0"), + "llmnr": Variant("i", 2), + "mdns": Variant("i", 2), + "timestamp": Variant("t", 1598125548), + "type": Variant("s", "802-3-ethernet"), + "uuid": Variant("s", "0c23631e-2118-355c-bbb0-8943229cb0d6"), + }, + "ipv4": { + "address-data": Variant( + "aa{sv}", + [ + { + "address": Variant("s", "192.168.2.148"), + "prefix": Variant("u", 24), + } + ], + ), + "addresses": Variant("aau", [[2483202240, 24, 0]]), + "dns": Variant("au", [16951488]), + "dns-data": Variant("as", ["192.168.2.1"]), + "gateway": Variant("s", "192.168.2.1"), + "method": Variant("s", "auto"), + "route-data": Variant( + "aa{sv}", + [ + { + "dest": Variant("s", "192.168.122.0"), + "prefix": Variant("u", 24), + "next-hop": Variant("s", "10.10.10.1"), + } + ], + ), + "routes": Variant("aau", [[8038592, 24, 17435146, 0]]), + }, + "ipv6": { + "method": Variant("s", "auto"), + "dns": Variant("aay", [IPv6Address("2001:4860:4860::8888").packed]), + "dns-data": Variant("as", ["2001:4860:4860::8888"]), + "addr-gen-mode": Variant("i", 0), + }, + "802-3-ethernet": { + "assigned-mac-address": Variant("s", "preserve"), + }, + }, +) + +SETTINGS_2_FIXTURE = settings_update( + MINIMAL_ETHERNET_SETTINGS_FIXTURE, + { "connection": { k: v - for k, v in SETTINGS_FIXTURE["connection"].items() + for k, v in SETTINGS_1_FIXTURE["connection"].items() if k != "interface-name" }, - "ipv4": SETTINGS_FIXTURE["ipv4"], - "ipv6": SETTINGS_FIXTURE["ipv6"], - "proxy": {}, - "802-3-ethernet": SETTINGS_FIXTURE["802-3-ethernet"], - "802-11-wireless": SETTINGS_FIXTURE["802-11-wireless"], + "ipv4": SETTINGS_1_FIXTURE["ipv4"], + "ipv6": SETTINGS_1_FIXTURE["ipv6"], + "802-3-ethernet": SETTINGS_1_FIXTURE["802-3-ethernet"], "match": {"path": Variant("as", ["platform-ff3f0000.ethernet"])}, }, +) + +SETTINGS_3_FIXTURE = deepcopy(MINIMAL_WIRELESS_SETTINGS_FIXTURE) + +SETINGS_FIXTURES: dict[str, dict[str, dict[str, Variant]]] = { + "/org/freedesktop/NetworkManager/Settings/1": SETTINGS_1_FIXTURE, + "/org/freedesktop/NetworkManager/Settings/2": SETTINGS_2_FIXTURE, + "/org/freedesktop/NetworkManager/Settings/3": SETTINGS_3_FIXTURE, } @@ -100,7 +166,7 @@ class ConnectionSettings(DBusServiceMock): """Initialize object.""" super().__init__() self.object_path = object_path - self.settings = SETINGS_FIXTURES[object_path] + self.settings = deepcopy(SETINGS_FIXTURES[object_path]) @dbus_property(access=PropertyAccess.READ) def Unsaved(self) -> "b": @@ -128,7 +194,64 @@ class ConnectionSettings(DBusServiceMock): @dbus_method() def Update(self, properties: "a{sa{sv}}") -> None: """Do Update method.""" - self.settings = properties + if "connection" not in properties: + raise DBusError( + "org.freedesktop.NetworkManager.Settings.Connection.MissingProperty", + "connection.type: property is missing", + ) + for required_prop in ("type", "id"): + if required_prop not in properties["connection"]: + raise DBusError( + "org.freedesktop.NetworkManager.Settings.Connection.MissingProperty", + f"connection.{required_prop}: property is missing", + ) + if properties["connection"]["type"] == "802-11-wireless": + self.settings = settings_update( + MINIMAL_WIRELESS_SETTINGS_FIXTURE, properties + ) + elif properties["connection"]["type"] == "802-3-ethernet": + self.settings = settings_update( + MINIMAL_ETHERNET_SETTINGS_FIXTURE, properties + ) + else: + self.settings = settings_update(MINIMAL_SETTINGS_FIXTURE, properties) + # Post process addresses/address-data and dns/dns-data + # If both "address" and "address-data" are provided the former wins + # If both "dns" and "dns-data" are provided the former wins + if "ipv4" in properties: + ipv4 = properties["ipv4"] + if "address-data" in ipv4: + addresses = Variant("aau", []) + for entry in ipv4["address-data"].value: + addresses.value.append( + [ + socket.htonl(int(IPv4Address(entry["address"].value))), + entry["prefix"].value, + 0, + ] + ) + self.settings["ipv4"]["addresses"] = addresses + if "addresses" in ipv4: + address_data = Variant("aa{sv}", []) + for entry in ipv4["addresses"].value: + ipv4address = IPv4Address(socket.ntohl(entry[0])) + address_data.value.append( + { + "address": Variant("s", str(ipv4address)), + "prefix": Variant("u", int(entry[1])), + } + ) + self.settings["ipv4"]["address-data"] = address_data + if "dns-data" in ipv4: + dns = Variant("au", []) + for entry in ipv4["dns-data"].value: + dns.value.append(socket.htonl(int(IPv4Address(entry)))) + self.settings["ipv4"]["dns"] = dns + if "dns" in ipv4: + dns_data = Variant("as", []) + for entry in ipv4["dns"].value: + dns_data.value.append(str(IPv4Address(socket.ntohl(entry)))) + self.settings["ipv4"]["dns-data"] = dns_data self.Updated() @dbus_method() diff --git a/tests/dbus_service_mocks/network_device.py b/tests/dbus_service_mocks/network_device.py index 61cd2b1f1..51a0efe91 100644 --- a/tests/dbus_service_mocks/network_device.py +++ b/tests/dbus_service_mocks/network_device.py @@ -9,7 +9,9 @@ from dbus_fast.service import PropertyAccess, dbus_property, signal from .base import DBusServiceMock, dbus_method BUS_NAME = "org.freedesktop.NetworkManager" -DEFAULT_OBJECT_PATH = "/org/freedesktop/NetworkManager/Devices/1" +ETHERNET_DEVICE_OBJECT_PATH = "/org/freedesktop/NetworkManager/Devices/1" +WIRELESS_DEVICE_OBJECT_PATH = "/org/freedesktop/NetworkManager/Devices/3" +DEFAULT_OBJECT_PATH = ETHERNET_DEVICE_OBJECT_PATH def setup(object_path: str | None = None) -> DBusServiceMock: @@ -112,7 +114,7 @@ FIXTURES: dict[str, DeviceFixture] = { FirmwareMissing=False, NmPluginMissing=False, DeviceType=2, - AvailableConnections=[], + AvailableConnections=["/org/freedesktop/NetworkManager/Settings/3"], PhysicalPortId="", Mtu=1500, Metered=0, diff --git a/tests/host/test_network.py b/tests/host/test_network.py index 4bb1a33b5..83eb21a8c 100644 --- a/tests/host/test_network.py +++ b/tests/host/test_network.py @@ -19,7 +19,7 @@ from tests.dbus_service_mocks.network_active_connection import ( ActiveConnection as ActiveConnectionService, ) from tests.dbus_service_mocks.network_connection_settings import ( - SETTINGS_FIXTURE, + SETTINGS_1_FIXTURE, ConnectionSettings as ConnectionSettingsService, ) from tests.dbus_service_mocks.network_device_wireless import ( @@ -83,11 +83,15 @@ async def test_load( assert name_dict["wlan0"].enabled is False assert connection_settings_service.settings["ipv4"]["method"].value == "auto" - assert "address-data" not in connection_settings_service.settings["ipv4"] + assert connection_settings_service.settings["ipv4"]["address-data"] == Variant( + "aa{sv}", [] + ) assert "gateway" not in connection_settings_service.settings["ipv4"] assert "dns" not in connection_settings_service.settings["ipv4"] assert connection_settings_service.settings["ipv6"]["method"].value == "auto" - assert "address-data" not in connection_settings_service.settings["ipv6"] + assert connection_settings_service.settings["ipv6"]["address-data"] == Variant( + "aa{sv}", [] + ) assert "gateway" not in connection_settings_service.settings["ipv6"] assert "dns" not in connection_settings_service.settings["ipv6"] @@ -110,7 +114,7 @@ async def test_load_with_disabled_methods( network_manager_service.ActivateConnection.calls.clear() disabled = {"method": Variant("s", "disabled")} - connection_settings_service.settings = SETTINGS_FIXTURE | { + connection_settings_service.settings = SETTINGS_1_FIXTURE | { "ipv4": disabled, "ipv6": disabled, } @@ -119,15 +123,6 @@ async def test_load_with_disabled_methods( await coresys.host.network.load() assert network_manager_service.ActivateConnection.calls == [] - assert connection_settings_service.settings["ipv4"]["method"].value == "disabled" - assert "address-data" not in connection_settings_service.settings["ipv4"] - assert "gateway" not in connection_settings_service.settings["ipv4"] - assert "dns" not in connection_settings_service.settings["ipv4"] - assert connection_settings_service.settings["ipv6"]["method"].value == "disabled" - assert "address-data" not in connection_settings_service.settings["ipv6"] - assert "gateway" not in connection_settings_service.settings["ipv6"] - assert "dns" not in connection_settings_service.settings["ipv6"] - async def test_load_with_network_connection_issues( coresys: CoreSys,