Home-Assistant/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-12-20 02:18:59 +00:00
Code Activity

Improve VLAN configuration (#6094)

Browse Source 
* Fix NetworkManager connection name for VLANs

The connection name for VLANs should include the parent interface name
for better identification. This was originally the intention, but the
interface object's name property was used which appears empty at that
point.

* Disallow creating multiple connections for the same VLAN id

Only allow a single connection per interface and VLAN id. The regular
network commands can be used to alter the configuration.

* Fix pytest

* Simply connection id name generation

Always rely on the Supervisor interface representation's name attribute
to generate the NetworkManager connection id. Make sure that the name
is correctly set when creating VLAN interfaces as well.

* Special case VLAN configuration

We can't use the match information when comparing Supervisor interface
representation with D-Bus representations. Special case VLAN and
compare using VLAN ID and parent interface.

Note that this currently compares connection UUID of the parent
interface.

* Fix pytest

* Separate VLAN creation logic from apply_changes

Apply changes is really all about updating the NetworkManager settings
of a particular network interface. The base in apply_changes() is
NetworkInterface class, which is the NetworkManager Device abstraction.
All physical interfaces have such a Device hence it is always present.

The only exception is when creating a VLAN: Since it is a virtual
device, there is no device when creating a VLAN.

This separate the two cases. This makes it much easier to reason if
a VLAN already exists or not, and to handle the case where a VLAN
needs to be created.

For all other network interfaces, the apply_changes() method can
now rely on the presence of the NetworkInterface Device abstraction.

* Add VLAN test interface and VLAN exists test

Add a test which checks that an error gets raised when a VLAN for a
particular interface/id combination already exists.

* Address pylint

* Fix test_ignore_veth_only_changes pytest

* Make VLAN interface disabled to avoid test issues

* Reference setting 38 in mocked connection

* Make sure interface type matches

Require a interface type match before doing any comparision.

* Add Supervisor host network configuration tests

* Fix device type checking

* Fix pytest

* Fix tests by taking VLAN interface into account

* Fix test_load_with_network_connection_issues

This seems like a hack, but it turns out that the additional active
connection caused coresys.host.network.update() to be called, which
implicitly "fake" activated the connection. Now it seems that our
mocking causes IPv4 gateway to be set.

So in a way, the test checked a particular mock behavior instead of
actual intention.

The crucial part of this test is that we make sure the settings remain
unchanged. This is done by ensuring that the the method is still auto.

* Fix test_check_network_interface_ipv4.py

Now that we have the VLAN interface active too it will raise an issue
as well.

* Apply suggestions from code review

Co-authored-by: Mike Degatano <michael.degatano@gmail.com>

* Fix ruff check issue

---------

Co-authored-by: Mike Degatano <michael.degatano@gmail.com>
This commit is contained in:
Stefan Agner
2025-08-22 11:09:39 +02:00
committed by GitHub
parent 3fc1abf661
commit 8a95113ebd
19 changed files with 551 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
supervisor/api/network.py
View File

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

4
supervisor/dbus/network/setting/__init__.py
View File

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

4
supervisor/dbus/network/setting/generate.py
View File

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

24
supervisor/host/configuration.py
View File

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

58
supervisor/host/network.py
View File

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

12
tests/api/test_network.py
View File

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

32
tests/conftest.py
View File

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

4
tests/dbus/network/setting/test_generate.py
View File

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

9
tests/dbus/network/setting/test_init.py
View File

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

9
tests/dbus/network/test_connection.py
View File

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

2
tests/dbus/network/test_network_manager.py
View File

@@ -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",
]
}
)

6
tests/dbus_service_mocks/network_active_connection.py
View File

@@ -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",
],
),
}

24
tests/dbus_service_mocks/network_connection_settings.py
View File

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

39
tests/dbus_service_mocks/network_device.py
View File

@@ -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",
),
}

50
tests/dbus_service_mocks/network_device_vlan.py Normal file
View File

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

7
tests/dbus_service_mocks/network_manager.py
View File

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

277
tests/host/test_configuration.py Normal file
View File

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

47
tests/host/test_network.py
View File

@@ -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):

15
tests/resolution/check/test_check_network_interface_ipv4.py
View File

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