Improve connection settings tests (#5278)

* Improve connection settings fixture

Make the connection settings fixture behave more closely to the actual
NetworkManager. The behavior has been tested with NetworkManager 1.42.4
(Debian 12) and 1.44.2 (HAOS 13.1). This likely behaves similar in older
versions too.

* Introduce separate skeleton and settings for wireless

Instead of having a combined network settings object which has
Ethernet and Wirless settings, create a separate settings object for
wireless.

* Handle addresses/address-data property like NetworkManager

* Address ruff check

* Improve network API test

Add a test which changes from "static" to "auto". Validate that settings
are updated accordingly. Specifically, today this does clear the DNS
setting (by not providing the property).

* ruff format

* ruff check

* Complete TEST_INTERFACE rename

* Add partial network update as test case

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 @@
 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"

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"

tests/dbus/network/setting/test_generate.py

@@ -9,12 +9,12 @@
 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)
 

tests/dbus/network/setting/test_init.py

@@ -16,6 +16,10 @@
 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,16 +102,22 @@ 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 "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" 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" not in settings
+        assert "802-11-wireless-security" not in settings
 
-    assert "802-11-wireless-security" not in settings
-    assert "vlan" 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):

tests/dbus/network/test_connection.py

@@ -7,7 +7,7 @@
 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