Fix typing issues in NetworkManager D-Bus integration (#6385)

* Fix typing for IPv6 addr-gen-mode and ip6-privacy settings

* Fix ConnectionStateType typing

* Rename ConnectionStateType to ConnectionState

The extra type suffix is unnecessary.

* Apply suggestions from code review

Co-authored-by: Jan Čermák <sairon@users.noreply.github.com>

---------

Co-authored-by: Jan Čermák <sairon@users.noreply.github.com>
This commit is contained in:
Stefan Agner 2025-12-03 16:28:43 +01:00 committed by GitHub
parent aeb8e59da4
commit fea8159ccf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 75 additions and 32 deletions

View File

@ -250,7 +250,7 @@ class ConnectionType(StrEnum):
WIRELESS = "802-11-wireless" WIRELESS = "802-11-wireless"
class ConnectionStateType(IntEnum): class ConnectionState(IntEnum):
"""Connection states. """Connection states.
https://networkmanager.dev/docs/api/latest/nm-dbus-types.html#NMActiveConnectionState https://networkmanager.dev/docs/api/latest/nm-dbus-types.html#NMActiveConnectionState

View File

@ -90,8 +90,8 @@ class Ip4Properties(IpProperties):
class Ip6Properties(IpProperties): class Ip6Properties(IpProperties):
"""IPv6 properties object for Network Manager.""" """IPv6 properties object for Network Manager."""
addr_gen_mode: int addr_gen_mode: int | None
ip6_privacy: int ip6_privacy: int | None
dns: list[bytes] | None dns: list[bytes] | None

View File

@ -16,8 +16,8 @@ from ..const import (
DBUS_IFACE_CONNECTION_ACTIVE, DBUS_IFACE_CONNECTION_ACTIVE,
DBUS_NAME_NM, DBUS_NAME_NM,
DBUS_OBJECT_BASE, DBUS_OBJECT_BASE,
ConnectionState,
ConnectionStateFlags, ConnectionStateFlags,
ConnectionStateType,
) )
from ..interface import DBusInterfaceProxy, dbus_property from ..interface import DBusInterfaceProxy, dbus_property
from ..utils import dbus_connected from ..utils import dbus_connected
@ -67,9 +67,9 @@ class NetworkConnection(DBusInterfaceProxy):
@property @property
@dbus_property @dbus_property
def state(self) -> ConnectionStateType: def state(self) -> ConnectionState:
"""Return the state of the connection.""" """Return the state of the connection."""
return ConnectionStateType(self.properties[DBUS_ATTR_STATE]) return ConnectionState(self.properties[DBUS_ATTR_STATE])
@property @property
def state_flags(self) -> set[ConnectionStateFlags]: def state_flags(self) -> set[ConnectionStateFlags]:

View File

@ -16,7 +16,11 @@ from ....host.const import (
InterfaceType, InterfaceType,
MulticastDnsMode, MulticastDnsMode,
) )
from ...const import MulticastDnsValue from ...const import (
InterfaceAddrGenMode as NMInterfaceAddrGenMode,
InterfaceIp6Privacy as NMInterfaceIp6Privacy,
MulticastDnsValue,
)
from .. import NetworkManager from .. import NetworkManager
from . import ( from . import (
CONF_ATTR_802_ETHERNET, CONF_ATTR_802_ETHERNET,
@ -118,24 +122,41 @@ def _get_ipv6_connection_settings(
ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "auto") ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "auto")
if ipv6setting: if ipv6setting:
if ipv6setting.addr_gen_mode == InterfaceAddrGenMode.EUI64: if ipv6setting.addr_gen_mode == InterfaceAddrGenMode.EUI64:
ipv6[CONF_ATTR_IPV6_ADDR_GEN_MODE] = Variant("i", 0) ipv6[CONF_ATTR_IPV6_ADDR_GEN_MODE] = Variant(
"i", NMInterfaceAddrGenMode.EUI64.value
)
elif ( elif (
not support_addr_gen_mode_defaults not support_addr_gen_mode_defaults
or ipv6setting.addr_gen_mode == InterfaceAddrGenMode.STABLE_PRIVACY or ipv6setting.addr_gen_mode == InterfaceAddrGenMode.STABLE_PRIVACY
): ):
ipv6[CONF_ATTR_IPV6_ADDR_GEN_MODE] = Variant("i", 1) ipv6[CONF_ATTR_IPV6_ADDR_GEN_MODE] = Variant(
"i", NMInterfaceAddrGenMode.STABLE_PRIVACY.value
)
elif ipv6setting.addr_gen_mode == InterfaceAddrGenMode.DEFAULT_OR_EUI64: elif ipv6setting.addr_gen_mode == InterfaceAddrGenMode.DEFAULT_OR_EUI64:
ipv6[CONF_ATTR_IPV6_ADDR_GEN_MODE] = Variant("i", 2) ipv6[CONF_ATTR_IPV6_ADDR_GEN_MODE] = Variant(
"i", NMInterfaceAddrGenMode.DEFAULT_OR_EUI64.value
)
else: else:
ipv6[CONF_ATTR_IPV6_ADDR_GEN_MODE] = Variant("i", 3) ipv6[CONF_ATTR_IPV6_ADDR_GEN_MODE] = Variant(
"i", NMInterfaceAddrGenMode.DEFAULT.value
)
if ipv6setting.ip6_privacy == InterfaceIp6Privacy.DISABLED: if ipv6setting.ip6_privacy == InterfaceIp6Privacy.DISABLED:
ipv6[CONF_ATTR_IPV6_PRIVACY] = Variant("i", 0) ipv6[CONF_ATTR_IPV6_PRIVACY] = Variant(
"i", NMInterfaceIp6Privacy.DISABLED.value
)
elif ipv6setting.ip6_privacy == InterfaceIp6Privacy.ENABLED_PREFER_PUBLIC: elif ipv6setting.ip6_privacy == InterfaceIp6Privacy.ENABLED_PREFER_PUBLIC:
ipv6[CONF_ATTR_IPV6_PRIVACY] = Variant("i", 1) ipv6[CONF_ATTR_IPV6_PRIVACY] = Variant(
"i", NMInterfaceIp6Privacy.ENABLED_PREFER_PUBLIC.value
)
elif ipv6setting.ip6_privacy == InterfaceIp6Privacy.ENABLED: elif ipv6setting.ip6_privacy == InterfaceIp6Privacy.ENABLED:
ipv6[CONF_ATTR_IPV6_PRIVACY] = Variant("i", 2) ipv6[CONF_ATTR_IPV6_PRIVACY] = Variant(
"i", NMInterfaceIp6Privacy.ENABLED.value
)
else: else:
ipv6[CONF_ATTR_IPV6_PRIVACY] = Variant("i", -1) ipv6[CONF_ATTR_IPV6_PRIVACY] = Variant(
"i", NMInterfaceIp6Privacy.DEFAULT.value
)
elif ipv6setting.method == InterfaceMethod.DISABLED: elif ipv6setting.method == InterfaceMethod.DISABLED:
ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "link-local") ipv6[CONF_ATTR_IPV6_METHOD] = Variant("s", "link-local")
elif ipv6setting.method == InterfaceMethod.STATIC: elif ipv6setting.method == InterfaceMethod.STATIC:

View File

@ -6,8 +6,8 @@ import logging
import socket import socket
from ..dbus.const import ( from ..dbus.const import (
ConnectionState,
ConnectionStateFlags, ConnectionStateFlags,
ConnectionStateType,
DeviceType, DeviceType,
InterfaceAddrGenMode as NMInterfaceAddrGenMode, InterfaceAddrGenMode as NMInterfaceAddrGenMode,
InterfaceIp6Privacy as NMInterfaceIp6Privacy, InterfaceIp6Privacy as NMInterfaceIp6Privacy,
@ -267,25 +267,47 @@ class Interface:
return InterfaceMethod.DISABLED return InterfaceMethod.DISABLED
@staticmethod @staticmethod
def _map_nm_addr_gen_mode(addr_gen_mode: int) -> InterfaceAddrGenMode: def _map_nm_addr_gen_mode(addr_gen_mode: int | None) -> InterfaceAddrGenMode:
"""Map IPv6 interface addr_gen_mode.""" """Map IPv6 interface addr_gen_mode.
NetworkManager omits the addr_gen_mode property when set to DEFAULT, so we
treat None as DEFAULT here.
"""
mapping = { mapping = {
NMInterfaceAddrGenMode.EUI64.value: InterfaceAddrGenMode.EUI64, NMInterfaceAddrGenMode.EUI64.value: InterfaceAddrGenMode.EUI64,
NMInterfaceAddrGenMode.STABLE_PRIVACY.value: InterfaceAddrGenMode.STABLE_PRIVACY, NMInterfaceAddrGenMode.STABLE_PRIVACY.value: InterfaceAddrGenMode.STABLE_PRIVACY,
NMInterfaceAddrGenMode.DEFAULT_OR_EUI64.value: InterfaceAddrGenMode.DEFAULT_OR_EUI64, NMInterfaceAddrGenMode.DEFAULT_OR_EUI64.value: InterfaceAddrGenMode.DEFAULT_OR_EUI64,
NMInterfaceAddrGenMode.DEFAULT.value: InterfaceAddrGenMode.DEFAULT,
None: InterfaceAddrGenMode.DEFAULT,
} }
if addr_gen_mode not in mapping:
_LOGGER.warning(
"Unknown addr_gen_mode value from NetworkManager: %s", addr_gen_mode
)
return mapping.get(addr_gen_mode, InterfaceAddrGenMode.DEFAULT) return mapping.get(addr_gen_mode, InterfaceAddrGenMode.DEFAULT)
@staticmethod @staticmethod
def _map_nm_ip6_privacy(ip6_privacy: int) -> InterfaceIp6Privacy: def _map_nm_ip6_privacy(ip6_privacy: int | None) -> InterfaceIp6Privacy:
"""Map IPv6 interface ip6_privacy.""" """Map IPv6 interface ip6_privacy.
NetworkManager omits the ip6_privacy property when set to DEFAULT, so we
treat None as DEFAULT here.
"""
mapping = { mapping = {
NMInterfaceIp6Privacy.DISABLED.value: InterfaceIp6Privacy.DISABLED, NMInterfaceIp6Privacy.DISABLED.value: InterfaceIp6Privacy.DISABLED,
NMInterfaceIp6Privacy.ENABLED_PREFER_PUBLIC.value: InterfaceIp6Privacy.ENABLED_PREFER_PUBLIC, NMInterfaceIp6Privacy.ENABLED_PREFER_PUBLIC.value: InterfaceIp6Privacy.ENABLED_PREFER_PUBLIC,
NMInterfaceIp6Privacy.ENABLED.value: InterfaceIp6Privacy.ENABLED, NMInterfaceIp6Privacy.ENABLED.value: InterfaceIp6Privacy.ENABLED,
NMInterfaceIp6Privacy.DEFAULT.value: InterfaceIp6Privacy.DEFAULT,
None: InterfaceIp6Privacy.DEFAULT,
} }
if ip6_privacy not in mapping:
_LOGGER.warning(
"Unknown ip6_privacy value from NetworkManager: %s", ip6_privacy
)
return mapping.get(ip6_privacy, InterfaceIp6Privacy.DEFAULT) return mapping.get(ip6_privacy, InterfaceIp6Privacy.DEFAULT)
@staticmethod @staticmethod
@ -295,8 +317,8 @@ class Interface:
return False return False
return connection.state in ( return connection.state in (
ConnectionStateType.ACTIVATED, ConnectionState.ACTIVATED,
ConnectionStateType.ACTIVATING, ConnectionState.ACTIVATING,
) )
@staticmethod @staticmethod

View File

@ -16,7 +16,7 @@ from ..dbus.const import (
DBUS_IFACE_DNS, DBUS_IFACE_DNS,
DBUS_IFACE_NM, DBUS_IFACE_NM,
DBUS_SIGNAL_NM_CONNECTION_ACTIVE_CHANGED, DBUS_SIGNAL_NM_CONNECTION_ACTIVE_CHANGED,
ConnectionStateType, ConnectionState,
ConnectivityState, ConnectivityState,
DeviceType, DeviceType,
WirelessMethodType, WirelessMethodType,
@ -338,16 +338,16 @@ class NetworkManager(CoreSysAttributes):
# the state change before this point. Get the state currently to # the state change before this point. Get the state currently to
# avoid any race condition. # avoid any race condition.
await con.update() await con.update()
state: ConnectionStateType = con.state state: ConnectionState = con.state
while state != ConnectionStateType.ACTIVATED: while state != ConnectionState.ACTIVATED:
if state == ConnectionStateType.DEACTIVATED: if state == ConnectionState.DEACTIVATED:
raise HostNetworkError( raise HostNetworkError(
"Activating connection failed, check connection settings." "Activating connection failed, check connection settings."
) )
msg = await signal.wait_for_signal() msg = await signal.wait_for_signal()
state = msg[0] state = ConnectionState(msg[0])
_LOGGER.debug("Active connection state changed to %s", state) _LOGGER.debug("Active connection state changed to %s", state)
# update_only means not done by user so don't force a check afterwards # update_only means not done by user so don't force a check afterwards

View File

@ -2,7 +2,7 @@
from ...const import CoreState from ...const import CoreState
from ...coresys import CoreSys from ...coresys import CoreSys
from ...dbus.const import ConnectionStateFlags, ConnectionStateType from ...dbus.const import ConnectionState, ConnectionStateFlags
from ...dbus.network.interface import NetworkInterface from ...dbus.network.interface import NetworkInterface
from ...exceptions import NetworkInterfaceNotFound from ...exceptions import NetworkInterfaceNotFound
from ..const import ContextType, IssueType from ..const import ContextType, IssueType
@ -47,7 +47,7 @@ class CheckNetworkInterfaceIPV4(CheckBase):
return not ( return not (
interface.connection.state interface.connection.state
in [ConnectionStateType.ACTIVATED, ConnectionStateType.ACTIVATING] in [ConnectionState.ACTIVATED, ConnectionState.ACTIVATING]
and ConnectionStateFlags.IP4_READY in interface.connection.state_flags and ConnectionStateFlags.IP4_READY in interface.connection.state_flags
) )

View File

@ -6,7 +6,7 @@ from unittest.mock import Mock, PropertyMock, patch
from dbus_fast.aio.message_bus import MessageBus from dbus_fast.aio.message_bus import MessageBus
import pytest import pytest
from supervisor.dbus.const import ConnectionStateType from supervisor.dbus.const import ConnectionState
from supervisor.dbus.network import NetworkManager from supervisor.dbus.network import NetworkManager
from supervisor.dbus.network.interface import NetworkInterface from supervisor.dbus.network.interface import NetworkInterface
from supervisor.exceptions import ( from supervisor.exceptions import (
@ -93,7 +93,7 @@ async def test_activate_connection(
"/org/freedesktop/NetworkManager/Settings/1", "/org/freedesktop/NetworkManager/Settings/1",
"/org/freedesktop/NetworkManager/Devices/1", "/org/freedesktop/NetworkManager/Devices/1",
) )
assert connection.state == ConnectionStateType.ACTIVATED assert connection.state == ConnectionState.ACTIVATED
assert ( assert (
connection.settings.object_path == "/org/freedesktop/NetworkManager/Settings/1" connection.settings.object_path == "/org/freedesktop/NetworkManager/Settings/1"
) )
@ -117,7 +117,7 @@ async def test_add_and_activate_connection(
) )
assert settings.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6" assert settings.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6"
assert settings.ipv4.method == "auto" assert settings.ipv4.method == "auto"
assert connection.state == ConnectionStateType.ACTIVATED assert connection.state == ConnectionState.ACTIVATED
assert ( assert (
connection.settings.object_path == "/org/freedesktop/NetworkManager/Settings/1" connection.settings.object_path == "/org/freedesktop/NetworkManager/Settings/1"
) )