From 8a95113ebd18bf4a5a3cae557732f91bf60be13e Mon Sep 17 00:00:00 2001 From: Stefan Agner Date: Fri, 22 Aug 2025 11:09:39 +0200 Subject: [PATCH] Improve VLAN configuration (#6094) * 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 * Fix ruff check issue --------- Co-authored-by: Mike Degatano --- supervisor/api/network.py | 4 +- supervisor/dbus/network/setting/__init__.py | 4 +- supervisor/dbus/network/setting/generate.py | 4 +- supervisor/host/configuration.py | 24 ++ supervisor/host/network.py | 58 ++-- tests/api/test_network.py | 12 +- tests/conftest.py | 32 +- tests/dbus/network/setting/test_generate.py | 4 +- tests/dbus/network/setting/test_init.py | 9 +- tests/dbus/network/test_connection.py | 9 +- tests/dbus/network/test_network_manager.py | 2 + .../network_active_connection.py | 6 + .../network_connection_settings.py | 24 +- tests/dbus_service_mocks/network_device.py | 39 +++ .../dbus_service_mocks/network_device_vlan.py | 50 ++++ tests/dbus_service_mocks/network_manager.py | 7 +- tests/host/test_configuration.py | 277 ++++++++++++++++++ tests/host/test_network.py | 47 +-- .../test_check_network_interface_ipv4.py | 15 +- 19 files changed, 551 insertions(+), 76 deletions(-) create mode 100644 tests/dbus_service_mocks/network_device_vlan.py create mode 100644 tests/host/test_configuration.py diff --git a/supervisor/api/network.py b/supervisor/api/network.py index c851f71b3..a3aa1bded 100644 --- a/supervisor/api/network.py +++ b/supervisor/api/network.py @@ -325,7 +325,7 @@ class APINetwork(CoreSysAttributes): ) vlan_interface = Interface( - "", + f"{interface.name}.{vlan}", "", "", True, @@ -339,4 +339,4 @@ class APINetwork(CoreSysAttributes): None, vlan_config, ) - await asyncio.shield(self.sys_host.network.apply_changes(vlan_interface)) + await asyncio.shield(self.sys_host.network.create_vlan(vlan_interface)) diff --git a/supervisor/dbus/network/setting/__init__.py b/supervisor/dbus/network/setting/__init__.py index 90a58bb80..7ef6d0538 100644 --- a/supervisor/dbus/network/setting/__init__.py +++ b/supervisor/dbus/network/setting/__init__.py @@ -273,8 +273,8 @@ class NetworkSetting(DBusInterface): if CONF_ATTR_VLAN in data: if CONF_ATTR_VLAN_ID in data[CONF_ATTR_VLAN]: self._vlan = VlanProperties( - data[CONF_ATTR_VLAN][CONF_ATTR_VLAN_ID], - data[CONF_ATTR_VLAN].get(CONF_ATTR_VLAN_PARENT), + id=data[CONF_ATTR_VLAN][CONF_ATTR_VLAN_ID], + parent=data[CONF_ATTR_VLAN].get(CONF_ATTR_VLAN_PARENT), ) else: self._vlan = None diff --git a/supervisor/dbus/network/setting/generate.py b/supervisor/dbus/network/setting/generate.py index b15a29e1e..88e642456 100644 --- a/supervisor/dbus/network/setting/generate.py +++ b/supervisor/dbus/network/setting/generate.py @@ -177,8 +177,6 @@ def get_connection_from_interface( # Generate/Update ID/name if not name or not name.startswith("Supervisor"): name = f"Supervisor {interface.name}" - if interface.type == InterfaceType.VLAN: - name = f"{name}.{cast(VlanConfig, interface.vlan).id}" if interface.type == InterfaceType.ETHERNET: iftype = "802-3-ethernet" @@ -220,7 +218,7 @@ def get_connection_from_interface( conn[CONF_ATTR_802_ETHERNET] = { CONF_ATTR_802_ETHERNET_ASSIGNED_MAC: Variant("s", "preserve") } - elif interface.type == "vlan": + elif interface.type == InterfaceType.VLAN: parent = cast(VlanConfig, interface.vlan).interface if ( parent diff --git a/supervisor/host/configuration.py b/supervisor/host/configuration.py index 729f4a25e..46a535450 100644 --- a/supervisor/host/configuration.py +++ b/supervisor/host/configuration.py @@ -82,6 +82,11 @@ class VlanConfig: """Represent a vlan configuration.""" id: int + # Note: On VLAN creation, parent is the interface name, but in the NetworkManager + # config the parent is set to the connection UUID in get_connection_from_interface(). + # On network update (which we call in apply_changes() on VLAN creation), the + # parent is then set to that connection UUID in _map_nm_vlan(), hence we always + # operate with a connection UUID as interface! interface: str | None @@ -108,6 +113,25 @@ class Interface: if not inet.settings: return False + # Special handling for VLAN interfaces + if self.type == InterfaceType.VLAN and inet.type == DeviceType.VLAN: + if not self.vlan: + raise RuntimeError("VLAN information missing") + + if inet.settings.vlan: + # For VLANs, compare by VLAN id and parent interface + return ( + inet.settings.vlan.id == self.vlan.id + and inet.settings.vlan.parent == self.vlan.interface + ) + return False + + if (self.type, inet.type) not in [ + (InterfaceType.ETHERNET, DeviceType.ETHERNET), + (InterfaceType.WIRELESS, DeviceType.WIRELESS), + ]: + return False + if inet.settings.match and inet.settings.match.path: return inet.settings.match.path == [self.path] diff --git a/supervisor/host/network.py b/supervisor/host/network.py index 5b7ba5b05..43b0c7179 100644 --- a/supervisor/host/network.py +++ b/supervisor/host/network.py @@ -5,6 +5,8 @@ from contextlib import suppress import logging from typing import Any +from supervisor.utils.sentry import async_capture_exception + from ..const import ATTR_HOST_INTERNET from ..coresys import CoreSys, CoreSysAttributes from ..dbus.const import ( @@ -20,7 +22,6 @@ from ..dbus.const import ( WirelessMethodType, ) from ..dbus.network.connection import NetworkConnection -from ..dbus.network.interface import NetworkInterface from ..dbus.network.setting.generate import get_connection_from_interface from ..exceptions import ( DBusError, @@ -209,20 +210,53 @@ class NetworkManager(CoreSysAttributes): await self.check_connectivity(force=force_connectivity_check) + async def create_vlan(self, interface: Interface) -> None: + """Create a VLAN interface.""" + if interface.vlan is None: + raise RuntimeError("VLAN information is missing") + # For VLAN interfaces, check if one already exists with same ID on same parent + try: + self.sys_dbus.network.get(interface.name) + except NetworkInterfaceNotFound: + _LOGGER.debug( + "VLAN interface %s does not exist, creating it", interface.name + ) + else: + raise HostNetworkError( + f"VLAN {interface.vlan.id} already exists on interface {interface.vlan.interface}", + _LOGGER.error, + ) + + settings = get_connection_from_interface(interface, self.sys_dbus.network) + + try: + await self.sys_dbus.network.settings.add_connection(settings) + except DBusError as err: + raise HostNetworkError( + f"Can't create new interface: {err}", _LOGGER.error + ) from err + + await self.update(force_connectivity_check=True) + async def apply_changes( self, interface: Interface, *, update_only: bool = False ) -> None: """Apply Interface changes to host.""" - inet: NetworkInterface | None = None - with suppress(NetworkInterfaceNotFound): + try: inet = self.sys_dbus.network.get(interface.name) + except NetworkInterfaceNotFound as err: + # The API layer (or anybody else) should not pass any updates for + # non-existing interfaces. + await async_capture_exception(err) + raise HostNetworkError( + "Requested Network interface update is not possible", _LOGGER.warning + ) from err con: NetworkConnection | None = None # Update exist configuration if ( - inet - and inet.settings + inet.settings and inet.settings.connection and interface.equals_dbus_interface(inet) and interface.enabled @@ -257,7 +291,7 @@ class NetworkManager(CoreSysAttributes): ) # Create new configuration and activate interface - elif inet and interface.enabled: + elif interface.enabled: _LOGGER.debug("Create new configuration for %s", interface.name) settings = get_connection_from_interface(interface, self.sys_dbus.network) @@ -280,7 +314,7 @@ class NetworkManager(CoreSysAttributes): ) from err # Remove config from interface - elif inet and not interface.enabled: + elif not interface.enabled: if not inet.settings: _LOGGER.debug("Interface %s is already disabled.", interface.name) return @@ -291,16 +325,6 @@ class NetworkManager(CoreSysAttributes): f"Can't disable interface {interface.name}: {err}", _LOGGER.error ) from err - # Create new interface (like vlan) - elif not inet: - settings = get_connection_from_interface(interface, self.sys_dbus.network) - - try: - await self.sys_dbus.network.settings.add_connection(settings) - except DBusError as err: - raise HostNetworkError( - f"Can't create new interface: {err}", _LOGGER.error - ) from err else: raise HostNetworkError( "Requested Network interface update is not possible", _LOGGER.warning diff --git a/tests/api/test_network.py b/tests/api/test_network.py index 29ccc5c0e..050156727 100644 --- a/tests/api/test_network.py +++ b/tests/api/test_network.py @@ -389,7 +389,7 @@ async def test_api_network_vlan( connection = settings_service.AddConnection.calls[0][0] assert "uuid" in connection["connection"] assert connection["connection"] == { - "id": Variant("s", "Supervisor .1"), + "id": Variant("s", "Supervisor eth0.1"), "type": Variant("s", "vlan"), "llmnr": Variant("i", 2), "mdns": Variant("i", 2), @@ -403,6 +403,16 @@ async def test_api_network_vlan( "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"), diff --git a/tests/conftest.py b/tests/conftest.py index da1976151..f14fbf09e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -64,11 +64,17 @@ from .common import ( from .const import TEST_ADDON_SLUG from .dbus_service_mocks.base import DBusServiceMock from .dbus_service_mocks.network_connection_settings import ( + DEFAULT_OBJECT_PATH as DEFAULT_CONNECTION_SETTINGS_OBJECT_PATH, ConnectionSettings as ConnectionSettingsService, ) from .dbus_service_mocks.network_dns_manager import DnsManager as DnsManagerService from .dbus_service_mocks.network_manager import NetworkManager as NetworkManagerService +from tests.dbus_service_mocks.network_active_connection import ( + DEFAULT_OBJECT_PATH as DEFAULT_ACTIVE_CONNECTION_OBJECT_PATH, + ActiveConnection as ActiveConnectionService, +) + # pylint: disable=redefined-outer-name, protected-access @@ -191,13 +197,21 @@ async def fixture_network_manager_services( "/org/freedesktop/NetworkManager/AccessPoint/43099", "/org/freedesktop/NetworkManager/AccessPoint/43100", ], - "network_active_connection": None, - "network_connection_settings": None, + "network_active_connection": [ + "/org/freedesktop/NetworkManager/ActiveConnection/1", + "/org/freedesktop/NetworkManager/ActiveConnection/38", + ], + "network_connection_settings": [ + "/org/freedesktop/NetworkManager/Settings/1", + "/org/freedesktop/NetworkManager/Settings/38", + ], "network_device_wireless": None, "network_device": [ "/org/freedesktop/NetworkManager/Devices/1", "/org/freedesktop/NetworkManager/Devices/3", + "/org/freedesktop/NetworkManager/Devices/38", ], + "network_device_vlan": None, "network_dns_manager": None, "network_ip4config": None, "network_ip6config": None, @@ -235,12 +249,24 @@ async def dns_manager_service( yield network_manager_services["network_dns_manager"] +@pytest.fixture(name="active_connection_service") +async def fixture_active_connection_service( + network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]], +) -> ActiveConnectionService: + """Return mock active connection service.""" + yield network_manager_services["network_active_connection"][ + DEFAULT_ACTIVE_CONNECTION_OBJECT_PATH + ] + + @pytest.fixture(name="connection_settings_service") async def fixture_connection_settings_service( network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]], ) -> ConnectionSettingsService: """Return mock connection settings service.""" - yield network_manager_services["network_connection_settings"] + yield network_manager_services["network_connection_settings"][ + DEFAULT_CONNECTION_SETTINGS_OBJECT_PATH + ] @pytest.fixture(name="udisks2_services") diff --git a/tests/dbus/network/setting/test_generate.py b/tests/dbus/network/setting/test_generate.py index df391c959..9132afd08 100644 --- a/tests/dbus/network/setting/test_generate.py +++ b/tests/dbus/network/setting/test_generate.py @@ -48,7 +48,7 @@ async def test_get_connection_no_path(network_manager: NetworkManager): async def test_generate_from_vlan(network_manager: NetworkManager): """Test generate from a vlan interface.""" vlan_interface = Interface( - name="", + name="eth0.1", mac="", path="", enabled=True, @@ -64,7 +64,7 @@ async def test_generate_from_vlan(network_manager: NetworkManager): ) connection_payload = get_connection_from_interface(vlan_interface, network_manager) - assert connection_payload["connection"]["id"].value == "Supervisor .1" + assert connection_payload["connection"]["id"].value == "Supervisor eth0.1" assert connection_payload["connection"]["type"].value == "vlan" assert "uuid" in connection_payload["connection"] assert "match" not in connection_payload["connection"] diff --git a/tests/dbus/network/setting/test_init.py b/tests/dbus/network/setting/test_init.py index f1197d993..8bdcbd66e 100644 --- a/tests/dbus/network/setting/test_init.py +++ b/tests/dbus/network/setting/test_init.py @@ -15,7 +15,6 @@ from supervisor.host.configuration import Ip6Setting from supervisor.host.const import InterfaceMethod from supervisor.host.network import Interface -from tests.dbus_service_mocks.base import DBusServiceMock from tests.dbus_service_mocks.network_connection_settings import ( ConnectionSettings as ConnectionSettingsService, ) @@ -24,13 +23,7 @@ from tests.dbus_service_mocks.network_device import ( WIRELESS_DEVICE_OBJECT_PATH, ) - -@pytest.fixture(name="connection_settings_service", autouse=True) -async def fixture_connection_settings_service( - network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]], -) -> ConnectionSettingsService: - """Mock Connection Settings service.""" - yield network_manager_services["network_connection_settings"] +pytestmark = pytest.mark.usefixtures("connection_settings_service") @pytest.fixture(name="dbus_interface") diff --git a/tests/dbus/network/test_connection.py b/tests/dbus/network/test_connection.py index 3fd89baf1..0a1fa1977 100644 --- a/tests/dbus/network/test_connection.py +++ b/tests/dbus/network/test_connection.py @@ -8,18 +8,11 @@ from supervisor.dbus.network import NetworkManager from supervisor.dbus.network.connection import NetworkConnection 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, ) - -@pytest.fixture(name="active_connection_service", autouse=True) -async def fixture_active_connection_service( - network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]], -) -> ActiveConnectionService: - """Mock Active Connection service.""" - yield network_manager_services["network_active_connection"] +pytestmark = pytest.mark.usefixtures("active_connection_service") async def test_active_connection( diff --git a/tests/dbus/network/test_network_manager.py b/tests/dbus/network/test_network_manager.py index 894c64ced..55b61d9c2 100644 --- a/tests/dbus/network/test_network_manager.py +++ b/tests/dbus/network/test_network_manager.py @@ -196,6 +196,7 @@ async def test_ignore_veth_only_changes( assert network_manager.properties["Devices"] == [ "/org/freedesktop/NetworkManager/Devices/1", "/org/freedesktop/NetworkManager/Devices/3", + "/org/freedesktop/NetworkManager/Devices/38", ] with patch.object(NetworkInterface, "connect") as connect: network_manager_service.emit_properties_changed( @@ -204,6 +205,7 @@ async def test_ignore_veth_only_changes( "/org/freedesktop/NetworkManager/Devices/1", "/org/freedesktop/NetworkManager/Devices/3", "/org/freedesktop/NetworkManager/Devices/35", + "/org/freedesktop/NetworkManager/Devices/38", ] } ) diff --git a/tests/dbus_service_mocks/network_active_connection.py b/tests/dbus_service_mocks/network_active_connection.py index 8b1b61fa1..4b47f5847 100644 --- a/tests/dbus_service_mocks/network_active_connection.py +++ b/tests/dbus_service_mocks/network_active_connection.py @@ -54,6 +54,12 @@ FIXTURES: dict[str, ActiveConnectionFixture] = { "/org/freedesktop/NetworkManager/Devices/3", ], ), + "/org/freedesktop/NetworkManager/ActiveConnection/38": ActiveConnectionFixture( + connection="/org/freedesktop/NetworkManager/Settings/38", + devices=[ + "/org/freedesktop/NetworkManager/Devices/38", + ], + ), } diff --git a/tests/dbus_service_mocks/network_connection_settings.py b/tests/dbus_service_mocks/network_connection_settings.py index 7d7efb814..fc9170968 100644 --- a/tests/dbus_service_mocks/network_connection_settings.py +++ b/tests/dbus_service_mocks/network_connection_settings.py @@ -142,10 +142,30 @@ SETTINGS_2_FIXTURE = settings_update( SETTINGS_3_FIXTURE = deepcopy(MINIMAL_WIRELESS_SETTINGS_FIXTURE) -SETINGS_FIXTURES: dict[str, dict[str, dict[str, Variant]]] = { +# VLAN settings fixture for eth0.10 (VLAN ID 10 on eth0) +SETTINGS_38_FIXTURE = settings_update( + MINIMAL_SETTINGS_FIXTURE, + { + "connection": { + "id": Variant("s", "Supervisor eth0.10"), + "type": Variant("s", "vlan"), + "uuid": Variant("s", "31ac31ac-31ac-31ac-31ac-31ac31ac31ac"), + "interface-name": Variant("s", "eth0.10"), + }, + "vlan": { + "id": Variant("u", 10), + "parent": Variant( + "s", "0c23631e-2118-355c-bbb0-8943229cb0d6" + ), # eth0's UUID + }, + }, +) + +SETTINGS_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, + "/org/freedesktop/NetworkManager/Settings/38": SETTINGS_38_FIXTURE, } @@ -166,7 +186,7 @@ class ConnectionSettings(DBusServiceMock): """Initialize object.""" super().__init__() self.object_path = object_path - self.settings = deepcopy(SETINGS_FIXTURES[object_path]) + self.settings = deepcopy(SETTINGS_FIXTURES[object_path]) @dbus_property(access=PropertyAccess.READ) def Unsaved(self) -> "b": diff --git a/tests/dbus_service_mocks/network_device.py b/tests/dbus_service_mocks/network_device.py index 51a0efe91..b4203c50a 100644 --- a/tests/dbus_service_mocks/network_device.py +++ b/tests/dbus_service_mocks/network_device.py @@ -55,6 +55,9 @@ class DeviceFixture: InterfaceFlags: c_uint32 HwAddress: str Ports: list[str] + # VLAN specific properties + VlanId: c_uint32 | None = None + Parent: str | None = None FIXTURES: dict[str, DeviceFixture] = { @@ -228,6 +231,42 @@ FIXTURES: dict[str, DeviceFixture] = { HwAddress="9A:4B:E3:9A:F8:D3", Ports=[], ), + "/org/freedesktop/NetworkManager/Devices/38": DeviceFixture( + Udi="/sys/devices/virtual/net/eth0.10", + Path="", + Interface="eth0.10", + IpInterface="eth0.10", + Driver="vlan", + DriverVersion="1.8", + FirmwareVersion="N/A", + Capabilities=7, + Ip4Address=0, + State=100, + StateReason=[100, 0], + ActiveConnection="/org/freedesktop/NetworkManager/ActiveConnection/38", + Ip4Config="/org/freedesktop/NetworkManager/IP4Config/1", + Dhcp4Config="/org/freedesktop/NetworkManager/DHCP4Config/1", + Ip6Config="/org/freedesktop/NetworkManager/IP6Config/1", + Dhcp6Config="/", + Managed=True, + Autoconnect=True, + FirmwareMissing=False, + NmPluginMissing=False, + DeviceType=11, + AvailableConnections=["/org/freedesktop/NetworkManager/Settings/38"], + PhysicalPortId="", + Mtu=1500, + Metered=4, + LldpNeighbors=[], + Real=True, + Ip4Connectivity=1, + Ip6Connectivity=3, + InterfaceFlags=65539, + HwAddress="52:54:00:2B:36:80", + Ports=[], + VlanId=10, + Parent="/org/freedesktop/NetworkManager/Devices/1", + ), } diff --git a/tests/dbus_service_mocks/network_device_vlan.py b/tests/dbus_service_mocks/network_device_vlan.py new file mode 100644 index 000000000..5ea95ac58 --- /dev/null +++ b/tests/dbus_service_mocks/network_device_vlan.py @@ -0,0 +1,50 @@ +"""Mock of Network Manager Device VLAN service.""" + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock +from .network_device import FIXTURES + +BUS_NAME = "org.freedesktop.NetworkManager" +VLAN_DEVICE_OBJECT_PATH = "/org/freedesktop/NetworkManager/Devices/38" +DEFAULT_OBJECT_PATH = VLAN_DEVICE_OBJECT_PATH + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return DeviceVlan(object_path if object_path else DEFAULT_OBJECT_PATH) + + +class DeviceVlan(DBusServiceMock): + """Device VLAN mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/Devices/38 + """ + + interface = "org.freedesktop.NetworkManager.Device.Vlan" + + def __init__(self, object_path: str): + """Initialize object.""" + super().__init__() + self.object_path = object_path + self.fixture = FIXTURES[object_path] + + @dbus_property(access=PropertyAccess.READ) + def HwAddress(self) -> "s": + """Get HwAddress.""" + return self.fixture.HwAddress + + @dbus_property(access=PropertyAccess.READ) + def Carrier(self) -> "b": + """Get Carrier.""" + return True + + @dbus_property(access=PropertyAccess.READ) + def Parent(self) -> "o": + """Get Parent.""" + return self.fixture.Parent + + @dbus_property(access=PropertyAccess.READ) + def VlanId(self) -> "u": + """Get VlanId.""" + return self.fixture.VlanId diff --git a/tests/dbus_service_mocks/network_manager.py b/tests/dbus_service_mocks/network_manager.py index cc224dd04..9254c599f 100644 --- a/tests/dbus_service_mocks/network_manager.py +++ b/tests/dbus_service_mocks/network_manager.py @@ -27,6 +27,7 @@ class NetworkManager(DBusServiceMock): devices = [ "/org/freedesktop/NetworkManager/Devices/1", "/org/freedesktop/NetworkManager/Devices/3", + "/org/freedesktop/NetworkManager/Devices/38", ] @dbus_property(access=PropertyAccess.READ) @@ -41,6 +42,7 @@ class NetworkManager(DBusServiceMock): "/org/freedesktop/NetworkManager/Devices/1", "/org/freedesktop/NetworkManager/Devices/2", "/org/freedesktop/NetworkManager/Devices/3", + "/org/freedesktop/NetworkManager/Devices/38", ] @dbus_property(access=PropertyAccess.READ) @@ -101,7 +103,10 @@ class NetworkManager(DBusServiceMock): @dbus_property(access=PropertyAccess.READ) def ActiveConnections(self) -> "ao": """Get ActiveConnections.""" - return ["/org/freedesktop/NetworkManager/ActiveConnection/1"] + return [ + "/org/freedesktop/NetworkManager/ActiveConnection/1", + "/org/freedesktop/NetworkManager/ActiveConnection/38", + ] @dbus_property(access=PropertyAccess.READ) def PrimaryConnection(self) -> "o": diff --git a/tests/host/test_configuration.py b/tests/host/test_configuration.py new file mode 100644 index 000000000..69654eed9 --- /dev/null +++ b/tests/host/test_configuration.py @@ -0,0 +1,277 @@ +"""Test host configuration interface.""" + +from unittest.mock import Mock + +from dbus_fast import Variant +import pytest + +from supervisor.coresys import CoreSys +from supervisor.dbus.const import DeviceType +from supervisor.host.configuration import Interface, VlanConfig +from supervisor.host.const import InterfaceType + +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 Device as DeviceService + + +async def test_equals_dbus_interface_no_settings(coresys: CoreSys): + """Test returns False when NetworkInterface has no settings.""" + await coresys.host.network.load() + + # Create test interface + test_interface = Interface( + name="eth0", + enabled=True, + connected=True, + primary=False, + type=InterfaceType.ETHERNET, + ipv4=None, + ipv4setting=None, + ipv6=None, + ipv6setting=None, + wifi=None, + vlan=None, + path="platform-ff3f0000.ethernet", + mac="AA:BB:CC:DD:EE:FF", + ) + + # Get network interface and remove its connection to simulate no settings + network_interface = coresys.dbus.network.get("eth0") + network_interface._connection = None + + assert test_interface.equals_dbus_interface(network_interface) is False + + +async def test_equals_dbus_interface_connection_name_match(coresys: CoreSys): + """Test interface comparison returns True when connection interface name matches.""" + await coresys.host.network.load() + + # Create test interface + test_interface = Interface( + name="eth0", + enabled=True, + connected=True, + primary=False, + type=InterfaceType.ETHERNET, + ipv4=None, + ipv4setting=None, + ipv6=None, + ipv6setting=None, + wifi=None, + vlan=None, + path="platform-ff3f0000.ethernet", + mac="AA:BB:CC:DD:EE:FF", + ) + + # Get the network interface - this should have connection settings with interface-name = "eth0" + network_interface = coresys.dbus.network.get("eth0") + + assert test_interface.equals_dbus_interface(network_interface) is True + + +def test_equals_dbus_interface_connection_name_no_match(): + """Test interface comparison returns False when connection interface name differs.""" + + # Create test interface + test_interface = Interface( + name="eth0", + enabled=True, + connected=True, + primary=False, + type=InterfaceType.ETHERNET, + ipv4=None, + ipv4setting=None, + ipv6=None, + ipv6setting=None, + wifi=None, + vlan=None, + path="platform-ff3f0000.ethernet", + mac="AA:BB:CC:DD:EE:FF", + ) + + # Mock network interface with different connection name + mock_network_interface = Mock() + mock_network_interface.type = DeviceType.ETHERNET + mock_network_interface.settings = Mock() + mock_network_interface.settings.match = None + mock_network_interface.settings.connection = Mock() + mock_network_interface.settings.connection.interface_name = "eth1" # Different name + + assert test_interface.equals_dbus_interface(mock_network_interface) is False + + +async def test_equals_dbus_interface_path_match( + coresys: CoreSys, + connection_settings_service: ConnectionSettingsService, +): + """Test interface comparison returns True when path matches.""" + await coresys.host.network.load() + + # Create test interface + test_interface = Interface( + name="eth0", + enabled=True, + connected=True, + primary=False, + type=InterfaceType.ETHERNET, + ipv4=None, + ipv4setting=None, + ipv6=None, + ipv6setting=None, + wifi=None, + vlan=None, + path="platform-ff3f0000.ethernet", + mac="AA:BB:CC:DD:EE:FF", + ) + + # Add match settings with path and remove interface name to force path matching + connection_settings_service.settings["match"] = { + "path": Variant("as", ["platform-ff3f0000.ethernet"]) + } + connection_settings_service.settings["connection"].pop("interface-name", None) + + network_interface = coresys.dbus.network.get("eth0") + + assert test_interface.equals_dbus_interface(network_interface) is True + + +def test_equals_dbus_interface_vlan_type_mismatch(): + """Test VLAN interface returns False when NetworkInterface type doesn't match.""" + + # Create VLAN test interface + test_vlan_interface = Interface( + name="eth0.10", + enabled=True, + connected=True, + primary=False, + type=InterfaceType.VLAN, + ipv4=None, + ipv4setting=None, + ipv6=None, + ipv6setting=None, + wifi=None, + vlan=VlanConfig(id=10, interface="0c23631e-2118-355c-bbb0-8943229cb0d6"), + path="", + mac="52:54:00:2B:36:80", + ) + + # Mock non-VLAN NetworkInterface - should return False immediately + mock_network_interface = Mock() + mock_network_interface.type = DeviceType.ETHERNET # Not VLAN type + mock_network_interface.settings = Mock() + + # Should return False immediately since types don't match + assert test_vlan_interface.equals_dbus_interface(mock_network_interface) is False + + +def test_equals_dbus_interface_vlan_missing_info(): + """Test VLAN interface raises RuntimeError when VLAN info is missing.""" + + # Create VLAN test interface without VLAN config + test_vlan_interface = Interface( + name="eth0.10", + enabled=True, + connected=True, + primary=False, + type=InterfaceType.VLAN, + ipv4=None, + ipv4setting=None, + ipv6=None, + ipv6setting=None, + wifi=None, + vlan=None, # Missing VLAN config! + path="", + mac="52:54:00:2B:36:80", + ) + + # Mock VLAN NetworkInterface + mock_network_interface = Mock() + mock_network_interface.type = DeviceType.VLAN + mock_network_interface.settings = Mock() + + # Should raise RuntimeError + try: + test_vlan_interface.equals_dbus_interface(mock_network_interface) + assert False, "Expected RuntimeError" + except RuntimeError as e: + assert str(e) == "VLAN information missing" + + +def test_equals_dbus_interface_vlan_no_vlan_settings(): + """Test VLAN interface returns False when NetworkInterface has no VLAN settings.""" + + # Create VLAN test interface + test_vlan_interface = Interface( + name="eth0.10", + enabled=True, + connected=True, + primary=False, + type=InterfaceType.VLAN, + ipv4=None, + ipv4setting=None, + ipv6=None, + ipv6setting=None, + wifi=None, + vlan=VlanConfig(id=10, interface="0c23631e-2118-355c-bbb0-8943229cb0d6"), + path="", + mac="52:54:00:2B:36:80", + ) + + # Mock VLAN NetworkInterface without VLAN settings + mock_network_interface = Mock() + mock_network_interface.type = DeviceType.VLAN + mock_network_interface.settings = Mock() + mock_network_interface.settings.vlan = None # No VLAN settings + + assert test_vlan_interface.equals_dbus_interface(mock_network_interface) is False + + +@pytest.fixture(name="device_eth0_10_service") +async def fixture_device_eth0_10_service( + network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]], +) -> DeviceService: + """Mock Device eth0.10 service.""" + yield network_manager_services["network_device"][ + "/org/freedesktop/NetworkManager/Devices/38" + ] + + +async def test_equals_dbus_interface_eth0_10_real( + coresys: CoreSys, device_eth0_10_service: DeviceService +): + """Test eth0.10 interface with real D-Bus interface.""" + await coresys.host.network.load() + + # Get the eth0.10 interface + network_interface = coresys.dbus.network.get("eth0.10") + + # Check if it has the expected VLAN type + assert network_interface.type == DeviceType.VLAN + assert network_interface.settings is not None + assert network_interface.settings.vlan is not None + + # Create matching test interface with correct parent UUID + test_vlan_interface = Interface( + name="eth0.10", + enabled=True, + connected=True, + primary=False, + type=InterfaceType.VLAN, + ipv4=None, + ipv4setting=None, + ipv6=None, + ipv6setting=None, + wifi=None, + vlan=VlanConfig( + id=network_interface.settings.vlan.id, + interface=network_interface.settings.vlan.parent, + ), + path="", + mac="52:54:00:2B:36:80", + ) + + # Test should pass with matching VLAN config + assert test_vlan_interface.equals_dbus_interface(network_interface) is True diff --git a/tests/host/test_network.py b/tests/host/test_network.py index f311b4bc6..22759f839 100644 --- a/tests/host/test_network.py +++ b/tests/host/test_network.py @@ -30,14 +30,6 @@ from tests.dbus_service_mocks.network_manager import ( ) -@pytest.fixture(name="active_connection_service") -async def fixture_active_connection_service( - network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]], -) -> ActiveConnectionService: - """Return mock active connection service.""" - yield network_manager_services["network_active_connection"] - - @pytest.fixture(name="wireless_service") async def fixture_wireless_service( network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]], @@ -62,7 +54,7 @@ async def test_load( 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) == 2 + 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" @@ -100,14 +92,20 @@ async def test_load( 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 network_manager_service.ActivateConnection.calls == [ - ( - "/org/freedesktop/NetworkManager/Settings/1", - "/org/freedesktop/NetworkManager/Devices/1", - "/", - ) - ] + 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 == [] @@ -127,7 +125,11 @@ async def test_load_with_disabled_methods( await coresys.dbus.network.get("eth0").settings.reload() await coresys.host.network.load() - assert network_manager_service.ActivateConnection.calls == [] + 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( @@ -144,16 +146,19 @@ async def test_load_with_network_connection_issues( await active_connection_service.ping() await coresys.host.network.load() + await coresys.host.network.update() - assert network_manager_service.ActivateConnection.calls == [] - assert len(coresys.host.network.interfaces) == 2 + 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"].ipv4.gateway is None assert name_dict["eth0"].ipv6setting.method == InterfaceMethod.AUTO - assert name_dict["eth0"].ipv6.gateway == IPv6Address("fe80::da58:d7ff:fe00:9c69") async def test_scan_wifi(coresys: CoreSys): diff --git a/tests/resolution/check/test_check_network_interface_ipv4.py b/tests/resolution/check/test_check_network_interface_ipv4.py index d0f93cca6..d5b50f20c 100644 --- a/tests/resolution/check/test_check_network_interface_ipv4.py +++ b/tests/resolution/check/test_check_network_interface_ipv4.py @@ -13,7 +13,10 @@ from supervisor.resolution.checks.network_interface_ipv4 import ( from supervisor.resolution.const import ContextType, IssueType from supervisor.resolution.data import Issue -TEST_ISSUE = Issue(IssueType.IPV4_CONNECTION_PROBLEM, ContextType.SYSTEM, "eth0") +TEST_ISSUES = { + Issue(IssueType.IPV4_CONNECTION_PROBLEM, ContextType.SYSTEM, "eth0"), + Issue(IssueType.IPV4_CONNECTION_PROBLEM, ContextType.SYSTEM, "eth0.10"), +} async def test_base(coresys: CoreSys): @@ -26,13 +29,13 @@ async def test_base(coresys: CoreSys): @pytest.mark.parametrize( "state_flags,issues", [ - ({ConnectionStateFlags.IP4_READY}, []), - ({ConnectionStateFlags.IP6_READY}, [TEST_ISSUE]), - ({ConnectionStateFlags.NONE}, [TEST_ISSUE]), + ({ConnectionStateFlags.IP4_READY}, set()), + ({ConnectionStateFlags.IP6_READY}, TEST_ISSUES), + ({ConnectionStateFlags.NONE}, TEST_ISSUES), ], ) async def test_check( - coresys: CoreSys, state_flags: set[ConnectionStateFlags], issues: list[Issue] + coresys: CoreSys, state_flags: set[ConnectionStateFlags], issues: set[Issue] ): """Test check.""" network_interface = CheckNetworkInterfaceIPV4(coresys) @@ -49,7 +52,7 @@ async def test_check( ): await network_interface.run_check() - assert coresys.resolution.issues == issues + assert set(coresys.resolution.issues) == issues @pytest.mark.parametrize(