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(