diff options
Diffstat (limited to 'tools/test-networkmanager-service.py')
| -rwxr-xr-x | tools/test-networkmanager-service.py | 200 |
1 files changed, 159 insertions, 41 deletions
diff --git a/tools/test-networkmanager-service.py b/tools/test-networkmanager-service.py index b4268721..8a426332 100755 --- a/tools/test-networkmanager-service.py +++ b/tools/test-networkmanager-service.py @@ -429,6 +429,7 @@ IFACE_AGENT = "org.freedesktop.NetworkManager.SecretAgent" IFACE_WIRED = "org.freedesktop.NetworkManager.Device.Wired" IFACE_MODEM = "org.freedesktop.NetworkManager.Device.Modem" IFACE_VLAN = "org.freedesktop.NetworkManager.Device.Vlan" +IFACE_MACVLAN = "org.freedesktop.NetworkManager.Device.Macvlan" IFACE_WIFI_AP = "org.freedesktop.NetworkManager.AccessPoint" IFACE_ACTIVE_CONNECTION = "org.freedesktop.NetworkManager.Connection.Active" IFACE_VPN_CONNECTION = "org.freedesktop.NetworkManager.VPN.Connection" @@ -611,6 +612,7 @@ class NmUtil: t = s_con[NM.SETTING_CONNECTION_TYPE] if t not in [ NM.SETTING_GSM_SETTING_NAME, + NM.SETTING_MACVLAN_SETTING_NAME, NM.SETTING_VLAN_SETTING_NAME, NM.SETTING_VPN_SETTING_NAME, NM.SETTING_WIMAX_SETTING_NAME, @@ -845,13 +847,14 @@ PRP_DEVICE_DEVICE_TYPE = "DeviceType" PRP_DEVICE_AVAILABLE_CONNECTIONS = "AvailableConnections" PRP_DEVICE_LLDP_NEIGHBORS = "LldpNeighbors" PRP_DEVICE_INTERFACE_FLAGS = "InterfaceFlags" +PRP_DEVICE_HW_ADDRESS = "HwAddress" class Device(ExportedObj): path_counter_next = 1 path_prefix = "/org/freedesktop/NetworkManager/Devices/" - def __init__(self, iface, devtype, ident=None): + def __init__(self, iface, devtype, ident=None, hwaddr=None): if ident is None: ident = iface @@ -863,6 +866,7 @@ class Device(ExportedObj): self.dhcp4_config = None self.dhcp6_config = None self.activation_state_change_delay_ms = 50 + self.hwaddr = hwaddr is None if "" else hwaddr self.prp_state = NM.DeviceState.UNAVAILABLE @@ -890,6 +894,7 @@ class Device(ExportedObj): PRP_DEVICE_DEVICE_TYPE: dbus.UInt32(devtype), PRP_DEVICE_AVAILABLE_CONNECTIONS: ExportedObj.to_path_array([]), PRP_DEVICE_INTERFACE_FLAGS: dbus.UInt32(3), # up,lower-up + PRP_DEVICE_HW_ADDRESS: dbus.String(self.hwaddr), PRP_DEVICE_LLDP_NEIGHBORS: dbus.Array( [ dbus.Dictionary( @@ -1166,10 +1171,11 @@ PRP_WIRED_S390_SUBCHANNELS = "S390Subchannels" class WiredDevice(Device): def __init__(self, iface, mac=None, subchannels=None, ident=None): - Device.__init__(self, iface, NM.DeviceType.ETHERNET, ident) - if mac is None: - mac = Util.random_mac(self.ident) + mac = Util.random_mac(iface if ident is None else ident) + + Device.__init__(self, iface, NM.DeviceType.ETHERNET, ident, hwaddr=mac) + if subchannels is None: subchannels = dbus.Array(signature="s") @@ -1206,14 +1212,23 @@ class ModemDevice(Device): ############################################################################### + +class MacvlanDevice(Device): + def __init__(self, iface, mac=None): + Device.__init__(self, iface, NM.DeviceType.MACVLAN, hwaddr=mac) + self.dbus_interface_add(IFACE_MACVLAN, {}) + + +############################################################################### + PRP_VLAN_HW_ADDRESS = "HwAddress" PRP_VLAN_CARRIER = "Carrier" PRP_VLAN_VLAN_ID = "VlanId" class VlanDevice(Device): - def __init__(self, iface, ident=None): - Device.__init__(self, iface, NM.DeviceType.VLAN, ident) + def __init__(self, iface, mac=None, ident=None): + Device.__init__(self, iface, NM.DeviceType.VLAN, ident, hwaddr=mac) props = { PRP_VLAN_HW_ADDRESS: Util.random_mac(self.ident), @@ -1313,10 +1328,10 @@ PRP_WIFI_LAST_SCAN = "LastScan" class WifiDevice(Device): def __init__(self, iface, mac=None, ident=None): - Device.__init__(self, iface, NM.DeviceType.WIFI, ident) - if mac is None: - mac = Util.random_mac(self.ident) + mac = Util.random_mac(iface if ident is None else ident) + + Device.__init__(self, iface, NM.DeviceType.WIFI, ident, hwaddr=mac) self.aps = [] self.scan_cb_id = None @@ -1651,30 +1666,16 @@ class NetworkManager(ExportedObj): raise BusErr.UnknownConnectionException("Connection not found") con_hash = con_inst.con_hash - con_type = NmUtil.con_hash_get_type(con_hash) - device = self.find_device_first(path=devpath) - if not device: - if con_type == NM.SETTING_WIRED_SETTING_NAME: - device = self.find_device_first(dev_type=WiredDevice) - elif con_type == NM.SETTING_WIRELESS_SETTING_NAME: - device = self.find_device_first(dev_type=WifiDevice) - elif con_type == NM.SETTING_VLAN_SETTING_NAME: - ifname = con_hash[NM.SETTING_CONNECTION_SETTING_NAME]["interface-name"] - device = VlanDevice(ifname) + device = self.find_device(devpath, con_hash) + if device is None: + device = self.create_device(con_hash) + if device: + # Just created the device, it can start activating right away + device.activation_state_change_delay_ms = 0 self.add_device(device) - elif con_type == NM.SETTING_VPN_SETTING_NAME: - for ac in self.active_connections: - if ac.is_vpn: - continue - if ac.device: - device = ac.device - break - - if not device: - raise BusErr.UnknownDeviceException( - "No device found for the requested iface." - ) + if device is None: + raise BusErr.UnknownDeviceException("Device not found") # See if we need secrets. For the moment, we only support WPA if "802-11-wireless-security" in con_hash: @@ -1733,11 +1734,16 @@ class NetworkManager(ExportedObj): out_signature="ooa{sv}", ) def AddAndActivateConnection2(self, con_hash, devpath, specific_object, options): - device = self.find_device_first( - path=devpath, require=BusErr.UnknownDeviceException - ) - conpath = gl.settings.AddConnection(con_hash) - return (conpath, self.ActivateConnection(conpath, devpath, specific_object), []) + conpath = gl.settings.add_connection(con_hash) + try: + return ( + conpath, + self.ActivateConnection(conpath, devpath, specific_object), + [], + ) + except: + gl.settings.delete_connection(conpath) + raise @dbus.service.method(dbus_interface=IFACE_NM, in_signature="o", out_signature="") def DeactivateConnection(self, active_connection): @@ -1807,6 +1813,7 @@ class NetworkManager(ExportedObj): iface=_DEFAULT_ARG, ip_iface=_DEFAULT_ARG, dev_type=_DEFAULT_ARG, + hwaddr=_DEFAULT_ARG, ): r = None for d in self.devices: @@ -1826,6 +1833,9 @@ class NetworkManager(ExportedObj): if dev_type is not _DEFAULT_ARG: if not isinstance(d, dev_type): continue + if hwaddr is not _DEFAULT_ARG: + if d.hwaddr.lower() != hwaddr.lower(): + continue yield d def find_device_first( @@ -1835,11 +1845,17 @@ class NetworkManager(ExportedObj): iface=_DEFAULT_ARG, ip_iface=_DEFAULT_ARG, dev_type=_DEFAULT_ARG, + hwaddr=_DEFAULT_ARG, require=None, ): r = None for d in self.find_devices( - ident=ident, path=path, iface=iface, ip_iface=ip_iface, dev_type=dev_type + ident=ident, + path=path, + iface=iface, + ip_iface=ip_iface, + dev_type=dev_type, + hwaddr=hwaddr, ): r = d break @@ -1849,6 +1865,107 @@ class NetworkManager(ExportedObj): raise BusErr.UnknownDeviceException("Device not found") return r + def find_device(self, devpath, con_hash): + device = self.find_device_first(path=devpath) + if device: + return device + + con_type = NmUtil.con_hash_get_type(con_hash) + + if con_type == NM.SETTING_WIRED_SETTING_NAME: + return self.find_device_first(dev_type=WiredDevice) + + if con_type == NM.SETTING_WIRELESS_SETTING_NAME: + return self.find_device_first(dev_type=WifiDevice) + + if con_type == NM.SETTING_VPN_SETTING_NAME: + for ac in self.active_connections: + if ac.is_vpn: + continue + if ac.device: + return ac.device + + return None + + def create_device(self, con_hash): + con_type = NmUtil.con_hash_get_type(con_hash) + + if con_type == NM.SETTING_VLAN_SETTING_NAME: + iface = con_hash[NM.SETTING_CONNECTION_SETTING_NAME].get("interface-name") + parent_iface = con_hash[NM.SETTING_VLAN_SETTING_NAME].get("parent") + + if NM.SETTING_WIRED_SETTING_NAME in con_hash: + mac = con_hash[NM.SETTING_WIRED_SETTING_NAME].get("mac-address") + else: + mac = None + if mac is not None: + parent_hwaddr = "%02X:%02X:%02X:%02X:%02X:%02X" % ( + mac[0], + mac[1], + mac[2], + mac[3], + mac[4], + mac[5], + ) + else: + parent_hwaddr = _DEFAULT_ARG + + parent_ident = parent_iface if parent_iface is not None else _DEFAULT_ARG + parent_device = self.find_device_first( + dev_type=WiredDevice, hwaddr=parent_hwaddr, ident=parent_ident + ) + if parent_device is None: + parent_device = self.find_device_first( + dev_type=MacvlanDevice, hwaddr=parent_hwaddr, ident=parent_ident + ) + if parent_device is None: + raise BusErr.UnknownDeviceException("Parent device not found") + + if parent_hwaddr is None: + parent_hwaddr = parent_device.hwaddr + + if NM.SETTING_WIRED_SETTING_NAME in con_hash: + mac = con_hash[NM.SETTING_WIRED_SETTING_NAME].get("cloned-mac-address") + else: + mac = None + if mac is not None: + hwaddr = "%02X:%02X:%02X:%02X:%02X:%02X" % ( + mac[0], + mac[1], + mac[2], + mac[3], + mac[4], + mac[5], + ) + else: + hwaddr = None + + if parent_iface is None: + parent_iface = parent_device.ident + + if iface is None: + iface = "%s.%d" % ( + parent_iface, + con_hash[NM.SETTING_VLAN_SETTING_NAME]["id"], + ) + + return VlanDevice(iface, mac=hwaddr) + + if con_type == NM.SETTING_MACVLAN_SETTING_NAME: + ifname = con_hash[NM.SETTING_CONNECTION_SETTING_NAME]["interface-name"] + mac = con_hash[NM.SETTING_WIRED_SETTING_NAME]["cloned-mac-address"] + hwaddr = "%02X:%02X:%02X:%02X:%02X:%02X" % ( + mac[0], + mac[1], + mac[2], + mac[3], + mac[4], + mac[5], + ) + return MacvlanDevice(ifname, hwaddr) + + return None + def add_device(self, device): if self.find_device_first(ident=device.ident, path=device.path) is not None: raise TestError( @@ -2135,7 +2252,7 @@ class Connection(ExportedObj): dbus_interface=IFACE_CONNECTION, in_signature="", out_signature="" ) def Delete(self): - gl.settings.delete_connection(self) + gl.settings.delete_connection(self.path) @dbus.service.method( dbus_interface=IFACE_CONNECTION, in_signature="a{sa{sv}}", out_signature="" @@ -2267,7 +2384,7 @@ class Settings(ExportedObj): def cb(): if hasattr(con_inst, "_remove_next_connection_cb"): del con_inst._remove_next_connection_cb - self.delete_connection(con_inst) + self.delete_connection(con_inst.path) return False # We will delete the connection right away on an idle handler. However, @@ -2284,8 +2401,9 @@ class Settings(ExportedObj): raise BusErr.UnknownConnectionException("Connection not found") self.connections[path].update_connection(con_hash, do_verify_strict) - def delete_connection(self, con_inst): - del self.connections[con_inst.path] + def delete_connection(self, path): + con_inst = self.get_connection(path) + del self.connections[path] self._dbus_property_set( IFACE_SETTINGS, PRP_SETTINGS_CONNECTIONS, |