Include interface name in match-device settings (#4444)

This commit is contained in:
Mike Degatano 2023-07-19 04:15:59 -04:00 committed by GitHub
parent b19dcef5b7
commit f1a72ee418
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 225 additions and 34 deletions

View File

@ -69,7 +69,7 @@ class NetworkManager(DBusInterfaceProxy):
@property
def interfaces(self) -> set[NetworkInterface]:
"""Return a dictionary of active interfaces."""
"""Return a set of active interfaces."""
return set(self._interfaces.values())
@property

View File

@ -60,7 +60,9 @@ def get_connection_from_interface(
if interface.type != InterfaceType.VLAN:
conn[CONF_ATTR_DEVICE] = {
ATTR_MATCH_DEVICE: Variant("s", f"mac:{interface.mac}")
ATTR_MATCH_DEVICE: Variant(
"s", f"mac:{interface.mac},interface-name:{interface.name}"
)
}
ipv4 = {}

View File

@ -75,8 +75,12 @@ class Interface:
if not inet.settings:
return False
if inet.settings.device:
return inet.settings.device.match_device == f"mac:{self.mac}"
if inet.settings.device and inet.settings.device.match_device:
matchers = inet.settings.device.match_device.split(",", 1)
return (
f"mac:{self.mac}" in matchers
or f"interface-name:{self.name}" in matchers
)
return inet.settings.connection.interface_name == self.name

View File

@ -1,7 +1,5 @@
"""Test settings generation from interface."""
import pytest
from supervisor.dbus.network import NetworkManager
from supervisor.dbus.network.setting.generate import get_connection_from_interface
from supervisor.host.network import Interface
@ -9,7 +7,6 @@ from supervisor.host.network import Interface
from tests.const import TEST_INTERFACE
@pytest.mark.asyncio
async def test_get_connection_from_interface(network_manager: NetworkManager):
"""Test network interface."""
dbus_interface = network_manager.get(TEST_INTERFACE)
@ -20,7 +17,10 @@ async def test_get_connection_from_interface(network_manager: NetworkManager):
assert "interface-name" not in connection_payload["connection"]
assert connection_payload["connection"]["type"].value == "802-3-ethernet"
assert connection_payload["device"]["match-device"].value == "mac:AA:BB:CC:DD:EE:FF"
assert (
connection_payload["device"]["match-device"].value
== "mac:AA:BB:CC:DD:EE:FF,interface-name:eth0"
)
assert connection_payload["ipv4"]["method"].value == "auto"
assert "address-data" not in connection_payload["ipv4"]

View File

@ -58,7 +58,9 @@ async def test_update(
)
assert settings["connection"]["autoconnect"] == Variant("b", True)
assert settings["device"] == {"match-device": Variant("s", "mac:AA:BB:CC:DD:EE:FF")}
assert settings["device"] == {
"match-device": Variant("s", "mac:AA:BB:CC:DD:EE:FF,interface-name:eth0")
}
assert "ipv4" in settings
assert settings["ipv4"]["method"] == Variant("s", "auto")

View File

@ -1,20 +1,59 @@
"""Mock of Network Manager Active Connection service."""
from dataclasses import dataclass, field
from dbus_fast.service import PropertyAccess, dbus_property, signal
from .base import DBusServiceMock
BUS_NAME = "org.freedesktop.NetworkManager"
DEFAULT_OBJECT_PATH = "/org/freedesktop/NetworkManager/ActiveConnection/1"
def setup(object_path: str | None = None) -> DBusServiceMock:
"""Create dbus mock object."""
return ActiveConnection()
return ActiveConnection(object_path if object_path else DEFAULT_OBJECT_PATH)
# pylint: disable=invalid-name
@dataclass(slots=True)
class ActiveConnectionFixture:
"""Active Connection fixture."""
connection: str = "/org/freedesktop/NetworkManager/Settings/1"
specific_object: str = "/"
id: str = "Wired connection 1"
uuid: str = "0c23631e-2118-355c-bbb0-8943229cb0d6"
type: str = "802-3-ethernet"
devices: list[str] = field(
default_factory=lambda: ["/org/freedesktop/NetworkManager/Devices/1"]
)
state: int = 2
state_flags: int = 92
default: bool = True
ip4_config: str = "/org/freedesktop/NetworkManager/IP4Config/1"
dhcp4_config: str = "/org/freedesktop/NetworkManager/DHCP4Config/1"
default6: bool = False
ip6_config: str = "/org/freedesktop/NetworkManager/IP6Config/1"
dhcp6_config: str = "/"
vpn: bool = False
master: str = "/"
FIXTURES: dict[str, ActiveConnectionFixture] = {
DEFAULT_OBJECT_PATH: ActiveConnectionFixture(),
"/org/freedesktop/NetworkManager/ActiveConnection/2": ActiveConnectionFixture(
connection="/org/freedesktop/NetworkManager/Settings/2",
devices=[
"/org/freedesktop/NetworkManager/Devices/4",
"/org/freedesktop/NetworkManager/Devices/5",
],
),
}
class ActiveConnection(DBusServiceMock):
"""Active Connection mock.
@ -24,85 +63,91 @@ class ActiveConnection(DBusServiceMock):
interface = "org.freedesktop.NetworkManager.Connection.Active"
object_path = "/org/freedesktop/NetworkManager/ActiveConnection/1"
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 Connection(self) -> "o":
"""Get Connection."""
return "/org/freedesktop/NetworkManager/Settings/1"
return self.fixture.connection
@dbus_property(access=PropertyAccess.READ)
def SpecificObject(self) -> "o":
"""Get SpecificObject."""
return "/"
return self.fixture.specific_object
@dbus_property(access=PropertyAccess.READ)
def Id(self) -> "s":
"""Get Id."""
return "Wired connection 1"
return self.fixture.id
@dbus_property(access=PropertyAccess.READ)
def Uuid(self) -> "s":
"""Get Uuid."""
return "0c23631e-2118-355c-bbb0-8943229cb0d6"
return self.fixture.uuid
@dbus_property(access=PropertyAccess.READ)
def Type(self) -> "s":
"""Get Type."""
return "802-3-ethernet"
return self.fixture.type
@dbus_property(access=PropertyAccess.READ)
def Devices(self) -> "ao":
"""Get Devices."""
return ["/org/freedesktop/NetworkManager/Devices/1"]
return self.fixture.devices
@dbus_property(access=PropertyAccess.READ)
def State(self) -> "u":
"""Get State."""
return 2
return self.fixture.state
@dbus_property(access=PropertyAccess.READ)
def StateFlags(self) -> "u":
"""Get StateFlags."""
return 92
return self.fixture.state_flags
@dbus_property(access=PropertyAccess.READ)
def Default(self) -> "b":
"""Get Default."""
return True
return self.fixture.default
@dbus_property(access=PropertyAccess.READ)
def Ip4Config(self) -> "o":
"""Get Ip4Config."""
return "/org/freedesktop/NetworkManager/IP4Config/1"
return self.fixture.ip4_config
@dbus_property(access=PropertyAccess.READ)
def Dhcp4Config(self) -> "o":
"""Get Dhcp4Config."""
return "/org/freedesktop/NetworkManager/DHCP4Config/1"
return self.fixture.dhcp4_config
@dbus_property(access=PropertyAccess.READ)
def Default6(self) -> "b":
"""Get Default6."""
return False
return self.fixture.default6
@dbus_property(access=PropertyAccess.READ)
def Ip6Config(self) -> "o":
"""Get Ip6Config."""
return "/org/freedesktop/NetworkManager/IP6Config/1"
return self.fixture.ip6_config
@dbus_property(access=PropertyAccess.READ)
def Dhcp6Config(self) -> "o":
"""Get Dhcp6Config."""
return "/"
return self.fixture.dhcp6_config
@dbus_property(access=PropertyAccess.READ)
def Vpn(self) -> "b":
"""Get Vpn."""
return False
return self.fixture.vpn
@dbus_property(access=PropertyAccess.READ)
def Master(self) -> "o":
"""Get Master."""
return "/"
return self.fixture.master
@signal()
def StateChanged(self) -> "uu":

View File

@ -6,7 +6,9 @@ from dbus_fast.service import PropertyAccess, dbus_property, signal
from .base import DBusServiceMock, dbus_method
BUS_NAME = "org.freedesktop.NetworkManager"
SETTINGS_FIXTURE = {
DEFAULT_OBJECT_PATH = "/org/freedesktop/NetworkManager/Settings/1"
SETTINGS_FIXTURE: dict[str, dict[str, Variant]] = {
"connection": {
"id": Variant("s", "Wired connection 1"),
"interface-name": Variant("s", "eth0"),
@ -62,11 +64,29 @@ SETTINGS_FIXTURE = {
},
"802-11-wireless": {"ssid": Variant("ay", b"NETT")},
}
SETINGS_FIXTURES: dict[str, dict[str, dict[str, Variant]]] = {
"/org/freedesktop/NetworkManager/Settings/1": SETTINGS_FIXTURE,
"/org/freedesktop/NetworkManager/Settings/2": {
"connection": {
k: v
for k, v in SETTINGS_FIXTURE["connection"].items()
if k != "interface-name"
},
"ipv4": SETTINGS_FIXTURE["ipv4"],
"ipv6": SETTINGS_FIXTURE["ipv6"],
"proxy": {},
"802-3-ethernet": SETTINGS_FIXTURE["802-3-ethernet"],
"802-11-wireless": SETTINGS_FIXTURE["802-11-wireless"],
"device": {
"match-device": Variant("s", "mac:AA:BB:CC:DD:EE:FF,interface-name:eth0"),
},
},
}
def setup(object_path: str | None = None) -> DBusServiceMock:
"""Create dbus mock object."""
return ConnectionSettings()
return ConnectionSettings(object_path if object_path else DEFAULT_OBJECT_PATH)
# pylint: disable=invalid-name
@ -79,8 +99,12 @@ class ConnectionSettings(DBusServiceMock):
"""
interface = "org.freedesktop.NetworkManager.Settings.Connection"
object_path = "/org/freedesktop/NetworkManager/Settings/1"
settings = SETTINGS_FIXTURE
def __init__(self, object_path: str):
"""Initialize object."""
super().__init__()
self.object_path = object_path
self.settings = SETINGS_FIXTURES[object_path]
@dbus_property(access=PropertyAccess.READ)
def Unsaved(self) -> "b":

View File

@ -127,6 +127,74 @@ FIXTURES: dict[str, DeviceFixture] = {
HwAddress="FF:EE:DD:CC:BB:AA",
Ports=[],
),
"/org/freedesktop/NetworkManager/Devices/4": DeviceFixture(
Udi="/sys/devices/pci0000:00/0000:00:1f.6/net/eth0",
Path="platform-ff3f0000.ethernet",
Interface="eth0",
IpInterface="eth0",
Driver="e1000e",
DriverVersion="3.2.6-k",
FirmwareVersion="0.7-4",
Capabilities=3,
Ip4Address=2499979456,
State=100,
StateReason=[100, 0],
ActiveConnection="/org/freedesktop/NetworkManager/ActiveConnection/2",
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=1,
AvailableConnections=["/org/freedesktop/NetworkManager/Settings/2"],
PhysicalPortId="",
Mtu=1500,
Metered=4,
LldpNeighbors=[],
Real=True,
Ip4Connectivity=4,
Ip6Connectivity=3,
InterfaceFlags=65539,
HwAddress="A1:B2:C3:D4:E5:F6",
Ports=[],
),
"/org/freedesktop/NetworkManager/Devices/5": DeviceFixture(
Udi="/sys/devices/pci0000:00/0000:00:1f.6/net/en0",
Path="platform-ff3f0000.ethernet",
Interface="en0",
IpInterface="en0",
Driver="e1000e",
DriverVersion="3.2.6-k",
FirmwareVersion="0.7-4",
Capabilities=3,
Ip4Address=2499979456,
State=100,
StateReason=[100, 0],
ActiveConnection="/org/freedesktop/NetworkManager/ActiveConnection/2",
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=1,
AvailableConnections=["/org/freedesktop/NetworkManager/Settings/2"],
PhysicalPortId="",
Mtu=1500,
Metered=4,
LldpNeighbors=[],
Real=True,
Ip4Connectivity=4,
Ip6Connectivity=3,
InterfaceFlags=65539,
HwAddress="AA:BB:CC:DD:EE:FF",
Ports=[],
),
"/org/freedesktop/NetworkManager/Devices/35": DeviceFixture(
Udi="/sys/devices/virtual/net/veth87bd238'",
Path="",

View File

@ -25,14 +25,15 @@ class NetworkManager(DBusServiceMock):
object_path = "/org/freedesktop/NetworkManager"
version = "1.22.10"
connectivity = 4
devices = [
"/org/freedesktop/NetworkManager/Devices/1",
"/org/freedesktop/NetworkManager/Devices/3",
]
@dbus_property(access=PropertyAccess.READ)
def Devices(self) -> "ao":
"""Get Devices."""
return [
"/org/freedesktop/NetworkManager/Devices/1",
"/org/freedesktop/NetworkManager/Devices/3",
]
return self.devices
@dbus_property(access=PropertyAccess.READ)
def AllDevices(self) -> "ao":

View File

@ -13,6 +13,7 @@ from supervisor.exceptions import HostNotSupportedError
from supervisor.homeassistant.const import WSEvent, WSType
from supervisor.host.const import WifiMode
from tests.common import mock_dbus_services
from tests.dbus_service_mocks.base import DBusServiceMock
from tests.dbus_service_mocks.network_active_connection import (
ActiveConnection as ActiveConnectionService,
@ -252,3 +253,47 @@ async def test_host_connectivity_disabled(
}
)
assert "connectivity_check" not in coresys.resolution.unsupported
@pytest.mark.parametrize(
"interface_obj_path",
[
"/org/freedesktop/NetworkManager/Devices/4",
"/org/freedesktop/NetworkManager/Devices/5",
],
)
async def test_load_with_mac_or_name_change(
coresys: CoreSys,
network_manager_service: NetworkManagerService,
interface_obj_path: str,
):
"""Test load fixes match-device settings if mac address or interface name has changed."""
await mock_dbus_services(
{
"network_active_connection": "/org/freedesktop/NetworkManager/ActiveConnection/2",
"network_connection_settings": "/org/freedesktop/NetworkManager/Settings/2",
"network_device": interface_obj_path,
},
coresys.dbus.bus,
)
await coresys.dbus.network.update({"Devices": [interface_obj_path]})
network_manager_service.ActivateConnection.calls.clear()
assert len(coresys.dbus.network.interfaces) == 1
interface = next(iter(coresys.dbus.network.interfaces))
assert interface.object_path == interface_obj_path
expected_match_device = (
f"mac:{interface.hw_address},interface-name:{interface.name}"
)
assert interface.settings.device.match_device != expected_match_device
await coresys.host.network.load()
assert network_manager_service.ActivateConnection.calls == [
(
"/org/freedesktop/NetworkManager/Settings/2",
interface_obj_path,
"/",
)
]
assert interface.settings.device.match_device == expected_match_device