tests: refactor finding device in NetworkManager stub

- add find_devices() and find_device_first() functions,
  to not re-implement iterating the device list.

- for test functions, accept the device's "ident", instead
  of ifname. The "ident" must b unique, contrary to the "ifname".
This commit is contained in:
Thomas Haller
2018-05-09 12:46:34 +02:00
parent d4093a3a2c
commit 6d34d5e41b

View File

@@ -23,6 +23,8 @@ import collections
import uuid import uuid
import hashlib import hashlib
_DEFAULT_ARG = object()
######################################################### #########################################################
class TestError(AssertionError): class TestError(AssertionError):
@@ -786,11 +788,8 @@ class NetworkManager(ExportedObj):
@dbus.service.method(dbus_interface=IFACE_NM, in_signature='s', out_signature='o') @dbus.service.method(dbus_interface=IFACE_NM, in_signature='s', out_signature='o')
def GetDeviceByIpIface(self, ip_iface): def GetDeviceByIpIface(self, ip_iface):
for d in self.devices: d = self.find_device_first(ip_iface = ip_iface, require = UnknownDeviceException)
# ignore iface/ip_iface distinction for now
if d.iface == ip_iface:
return to_path(d) return to_path(d)
raise UnknownDeviceException("No device found for the requested iface.")
@dbus.service.method(dbus_interface=IFACE_NM, in_signature='ooo', out_signature='o') @dbus.service.method(dbus_interface=IFACE_NM, in_signature='ooo', out_signature='o')
def ActivateConnection(self, conpath, devpath, specific_object): def ActivateConnection(self, conpath, devpath, specific_object):
@@ -802,11 +801,7 @@ class NetworkManager(ExportedObj):
hash = connection.GetSettings() hash = connection.GetSettings()
s_con = hash['connection'] s_con = hash['connection']
device = None device = self.find_device_first(path = devpath)
for d in self.devices:
if d.path == devpath:
device = d
break
if not device and s_con['type'] == 'vlan': if not device and s_con['type'] == 'vlan':
ifname = s_con['interface-name'] ifname = s_con['interface-name']
device = VlanDevice(self._bus, ifname) device = VlanDevice(self._bus, ifname)
@@ -842,14 +837,7 @@ class NetworkManager(ExportedObj):
@dbus.service.method(dbus_interface=IFACE_NM, in_signature='a{sa{sv}}oo', out_signature='oo') @dbus.service.method(dbus_interface=IFACE_NM, in_signature='a{sa{sv}}oo', out_signature='oo')
def AddAndActivateConnection(self, connection, devpath, specific_object): def AddAndActivateConnection(self, connection, devpath, specific_object):
device = None device = self.find_device_first(path = devpath, require = UnknownDeviceException)
for d in self.devices:
if d.path == devpath:
device = d
break
if not device:
raise UnknownDeviceException("No device found for the requested iface.")
conpath = settings.AddConnection(connection) conpath = settings.AddConnection(connection)
return (conpath, self.ActivateConnection(conpath, devpath, specific_object)) return (conpath, self.ActivateConnection(conpath, devpath, specific_object))
@@ -902,15 +890,38 @@ class NetworkManager(ExportedObj):
def DeviceAdded(self, devpath): def DeviceAdded(self, devpath):
pass pass
def find_device(self, ident): def find_devices(self, ident = _DEFAULT_ARG, path = _DEFAULT_ARG, iface = _DEFAULT_ARG, ip_iface = _DEFAULT_ARG):
r = None
for d in self.devices: for d in self.devices:
if d.ident == ident: if ident is not _DEFAULT_ARG:
return d if d.ident != ident:
continue
if path is not _DEFAULT_ARG:
if d.path != path:
continue
if iface is not _DEFAULT_ARG:
if d.iface != iface:
continue
if ip_iface is not _DEFAULT_ARG:
# ignore iface/ip_iface distinction for now
if d.iface != ip_iface:
continue
yield d
def find_device_first(self, ident = _DEFAULT_ARG, path = _DEFAULT_ARG, iface = _DEFAULT_ARG, ip_iface = _DEFAULT_ARG, require = None):
r = None
for d in self.find_devices(ident = ident, path = path, iface = iface, ip_iface = ip_iface):
r = d
break
if r is None and require:
if require is TestError:
raise TestError('Device not found')
raise UnknownDeviceException('Device not found')
return r
def add_device(self, device): def add_device(self, device):
d = self.find_device(device.ident) if self.find_device_first(ident = device.ident, path = device.path) is not None:
if d: raise TestError("Duplicate device ident=%s / path=%s" % (device.ident, device.path))
raise TestError("Device with ident=%s already added (%s)" % (device.ident, d.path))
self.devices.append(device) self.devices.append(device)
self.__notify(PM_DEVICES) self.__notify(PM_DEVICES)
self.__notify(PM_ALL_DEVICES) self.__notify(PM_ALL_DEVICES)
@@ -973,9 +984,7 @@ class NetworkManager(ExportedObj):
elif class_name in ['WifiAp']: elif class_name in ['WifiAp']:
if 'device' not in args: if 'device' not in args:
raise TestError('missing "device" paramter') raise TestError('missing "device" paramter')
d = self.find_device(args['device']) d = self.find_device_first(ident = args['device'], require = TestError)
if not d:
raise TestError('no device "%s" found' % args['device'])
del args['device'] del args['device']
if 'ssid' not in args: if 'ssid' not in args:
args['ssid'] = d.ident + '-ap-' + str(WifiAp.counter + 1) args['ssid'] = d.ident + '-ap-' + str(WifiAp.counter + 1)
@@ -1000,42 +1009,29 @@ class NetworkManager(ExportedObj):
@dbus.service.method(IFACE_TEST, in_signature='o', out_signature='') @dbus.service.method(IFACE_TEST, in_signature='o', out_signature='')
def RemoveDevice(self, path): def RemoveDevice(self, path):
for d in self.devices: d = self.find_device_first(path = path, require = TestError)
if d.path == path:
self.remove_device(d) self.remove_device(d)
return
raise UnknownDeviceException("Device not found")
@dbus.service.method(IFACE_TEST, in_signature='sss', out_signature='o') @dbus.service.method(IFACE_TEST, in_signature='sss', out_signature='o')
def AddWifiAp(self, ifname, ssid, bssid): def AddWifiAp(self, ident, ssid, bssid):
d = self.find_device(ifname) d = self.find_device_first(ident = ident, require = TestError)
if d:
ap = WifiAp(self._bus, ssid, bssid) ap = WifiAp(self._bus, ssid, bssid)
return to_path(d.add_ap(ap)) return to_path(d.add_ap(ap))
raise UnknownDeviceException("Device not found")
@dbus.service.method(IFACE_TEST, in_signature='so', out_signature='') @dbus.service.method(IFACE_TEST, in_signature='so', out_signature='')
def RemoveWifiAp(self, ifname, ap_path): def RemoveWifiAp(self, ident, ap_path):
for d in self.devices: d = self.find_device_first(ident = ident, require = TestError)
if d.iface == ifname:
d.remove_ap_by_path(ap_path) d.remove_ap_by_path(ap_path)
return
raise UnknownDeviceException("Device not found")
@dbus.service.method(IFACE_TEST, in_signature='ss', out_signature='o') @dbus.service.method(IFACE_TEST, in_signature='ss', out_signature='o')
def AddWimaxNsp(self, ifname, name): def AddWimaxNsp(self, ident, name):
for d in self.devices: d = self.find_device_first(ident = ident, require = TestError)
if d.iface == ifname:
return to_path(d.add_test_nsp(name)) return to_path(d.add_test_nsp(name))
raise UnknownDeviceException("Device not found")
@dbus.service.method(IFACE_TEST, in_signature='so', out_signature='') @dbus.service.method(IFACE_TEST, in_signature='so', out_signature='')
def RemoveWimaxNsp(self, ifname, nsp_path): def RemoveWimaxNsp(self, ident, nsp_path):
for d in self.devices: d = self.find_device_first(ident = ident, require = TestError)
if d.iface == ifname:
d.remove_nsp_by_path(nsp_path) d.remove_nsp_by_path(nsp_path)
return
raise UnknownDeviceException("Device not found")
@dbus.service.method(IFACE_TEST, in_signature='', out_signature='') @dbus.service.method(IFACE_TEST, in_signature='', out_signature='')
def AutoRemoveNextConnection(self): def AutoRemoveNextConnection(self):