clients/tests: add python test script for nmcli tests

Add a test which runs nmcli against our stub NetworkManager
service and compares the output.

The output formats of nmcli are complicated and not easily understood.
For example how --mode tabular|multiline interacts with selecting
output-fields (--fields) and output modes ([default]|--terse|--pretty).
Also, there are things like `nmcli connection show --order $FIELD_SPEC`.

We need unit tests to ensure that we don't change the output
accidentally.

1 files changed, 178 insertions, 71 deletions

diff --git a/tools/test-networkmanager-service.py b/tools/test-networkmanager-service.py
index 17bd7e28c0..ed435e67f8 100755
--- a/tools/test-networkmanager-service.py
+++ b/tools/test-networkmanager-service.py
@@ -21,6 +21,40 @@ import dbus.mainloop.glib
 import random
 import collections
 import uuid
+import hashlib
+
+#########################################################
+
+class TestError(AssertionError):
+    def __init__(self, message = 'Unspecified error', errors = None):
+        AssertionError.__init__(self, message)
+        self.errors = errors
+
+def pseudorandom_stream(seed, length = None):
+    seed = str(seed)
+    v = None
+    i = 0
+    while length is None or length > 0:
+        if not v:
+            s = seed + str(i)
+            s = s.encode('utf8')
+            v = hashlib.sha256(s).hexdigest()
+            i += 1
+        yield int(v[0:2], 16)
+        v = v[2:]
+        if length is not None:
+            length -= 1
+
+def pseudorandom_num(seed, v_end, v_start = 0):
+    n = 0
+    span = v_end - v_start
+    for r in pseudorandom_stream(seed):
+        n = n * 256 + r
+        if n > span:
+            break
+    return v_start + (n % span)
+
+#########################################################
 
 mainloop = GLib.MainLoop()
 
@@ -75,6 +109,7 @@ NM_ACTIVE_CONNECTION_STATE_DEACTIVATING = 3
 NM_ACTIVE_CONNECTION_STATE_DEACTIVATED  = 4
 
 #########################################################
+
 IFACE_DBUS = 'org.freedesktop.DBus'
 
 class UnknownInterfaceException(dbus.DBusException):
@@ -98,9 +133,21 @@ class ExportedObj(dbus.service.Object):
 
     DBusInterface = collections.namedtuple('DBusInterface', ['dbus_iface', 'get_props_func', 'prop_changed_func'])
 
-    def __init__(self, bus, object_path):
+    def __init__(self, bus, object_path, ident = None):
         dbus.service.Object.__init__(self, bus, object_path)
         self._bus = bus
+
+        # ident is an optional (unique) identifier for the instance.
+        # The test driver may set it to reference to the object by
+        # this identifier. For NetworkManager, the real ID of an
+        # object on D-Bus is the object_path. But that is generated
+        # by the stub server only after the test user created the
+        # object. The ident parameter may be specified by the user
+        # and thus can be hard-coded in the test.
+        if ident is None:
+            ident = object_path
+        self.ident = ident
+
         self.path = object_path
         self.__ensure_dbus_ifaces()
         object_manager.add_object(self)
@@ -176,7 +223,11 @@ PD_AVAILABLE_CONNECTIONS = "AvailableConnections"
 class Device(ExportedObj):
     counter = 1
 
-    def __init__(self, bus, iface, devtype):
+    def __init__(self, bus, iface, devtype, ident = None):
+
+        if ident is None:
+            ident = iface
+
         object_path = "/org/freedesktop/NetworkManager/Devices/%d" % Device.counter
         Device.counter = Device.counter + 1
 
@@ -192,7 +243,7 @@ class Device(ExportedObj):
         self.available_connections = []
 
         self.add_dbus_interface(IFACE_DEVICE, self.__get_props, Device.PropertiesChanged)
-        ExportedObj.__init__(self, bus, object_path)
+        ExportedObj.__init__(self, bus, object_path, ident)
 
     # Properties interface
     def __get_props(self):
@@ -236,11 +287,12 @@ class Device(ExportedObj):
 
 ###################################################################
 
-def random_mac():
-    return '%02X:%02X:%02X:%02X:%02X:%02X' % (
-        random.randint(0, 255), random.randint(0, 255), random.randint(0, 255),
-        random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)
-      )
+def random_mac(seed = None):
+    if seed is None:
+        r = tuple([random.randint(0, 255) for x in range(6)])
+    else:
+        r = tuple(pseudorandom_stream(seed, 6))
+    return '%02X:%02X:%02X:%02X:%02X:%02X' % r
 
 ###################################################################
 IFACE_WIRED = 'org.freedesktop.NetworkManager.Device.Wired'
@@ -252,17 +304,17 @@ PE_CARRIER = "Carrier"
 PE_S390_SUBCHANNELS = "S390Subchannels"
 
 class WiredDevice(Device):
-    def __init__(self, bus, iface, mac, subchannels):
-
+    def __init__(self, bus, iface, mac = None, subchannels = None, ident = None):
         if mac is None:
-            self.mac = random_mac()
-        else:
-            self.mac = mac
+            mac = random_mac(iface if ident is None else ident)
+        if subchannels is None:
+            subchannels = dbus.Array(signature = 's')
+        self.mac = mac
         self.carrier = False
         self.s390_subchannels = subchannels
 
         self.add_dbus_interface(IFACE_WIRED, self.__get_props, WiredDevice.PropertiesChanged)
-        Device.__init__(self, bus, iface, NM_DEVICE_TYPE_ETHERNET)
+        Device.__init__(self, bus, iface, NM_DEVICE_TYPE_ETHERNET, ident)
 
     # Properties interface
     def __get_props(self):
@@ -289,13 +341,13 @@ PV_CARRIER = "Carrier"
 PV_VLAN_ID = "VlanId"
 
 class VlanDevice(Device):
-    def __init__(self, bus, iface):
-        self.mac = random_mac()
+    def __init__(self, bus, iface, ident = None):
+        self.mac = random_mac(iface if ident is None else ident)
         self.carrier = False
         self.vlan_id = 1
 
         self.add_dbus_interface(IFACE_VLAN, self.__get_props, VlanDevice.PropertiesChanged)
-        Device.__init__(self, bus, iface, NM_DEVICE_TYPE_VLAN)
+        Device.__init__(self, bus, iface, NM_DEVICE_TYPE_VLAN, ident)
 
     # Properties interface
     def __get_props(self):
@@ -325,24 +377,35 @@ PP_STRENGTH = "Strength"
 class WifiAp(ExportedObj):
     counter = 0
 
-    def __init__(self, bus, ssid, mac, flags, wpaf, rsnf, freq):
+    def __init__(self, bus, ssid, bssid = None, flags = None, wpaf = None, rsnf = None, freq = None, strength = None, ident = None):
         path = "/org/freedesktop/NetworkManager/AccessPoint/%d" % WifiAp.counter
         WifiAp.counter = WifiAp.counter + 1
 
+        if flags is None:
+            flags = 0x1
+        if wpaf is None:
+            wpaf = 0x1cc
+        if rsnf is None:
+            rsnf = 0x1cc
+        if freq is None:
+            freq = 2412
+        if bssid is None:
+            bssid = random_mac(path)
+        if strength is None:
+            strength = pseudorandom_num(path, 100)
+
         self.ssid = ssid
-        if mac:
-            self.bssid = mac
-        else:
-            self.bssid = random_mac()
+        self.bssid = bssid
         self.flags = flags
         self.wpaf = wpaf
         self.rsnf = rsnf
         self.freq = freq
-        self.strength = random.randint(0, 100)
+        self.strength = strength
+        self.strength_counter = 0
         self.strength_id = GLib.timeout_add_seconds(10, self.strength_cb, None)
 
         self.add_dbus_interface(IFACE_WIFI_AP, self.__get_props, WifiAp.PropertiesChanged)
-        ExportedObj.__init__(self, bus, path)
+        ExportedObj.__init__(self, bus, path, ident)
 
     def __del__(self):
         if self.strength_id > 0:
@@ -350,7 +413,8 @@ class WifiAp(ExportedObj):
         self.strength_id = 0
 
     def strength_cb(self, ignored):
-        self.strength = random.randint(0, 100)
+        self.strength_counter += 1
+        self.strength = pseudorandom_num(self.path + str(self.strength_counter), 100)
         self.__notify(PP_STRENGTH)
         return True
 
@@ -390,13 +454,15 @@ PW_ACTIVE_ACCESS_POINT = "ActiveAccessPoint"
 PW_WIRELESS_CAPABILITIES = "WirelessCapabilities"
 
 class WifiDevice(Device):
-    def __init__(self, bus, iface):
-        self.mac = random_mac()
+    def __init__(self, bus, iface, mac = None, ident = None):
+        if mac is None:
+            mac = random_mac(iface if ident is None else ident)
+        self.mac = mac
         self.aps = []
         self.active_ap = None
 
         self.add_dbus_interface(IFACE_WIFI, self.__get_props, WifiDevice.PropertiesChanged)
-        Device.__init__(self, bus, iface, NM_DEVICE_TYPE_WIFI)
+        Device.__init__(self, bus, iface, NM_DEVICE_TYPE_WIFI, ident)
 
     # methods
     @dbus.service.method(dbus_interface=IFACE_WIFI, in_signature='', out_signature='ao')
@@ -421,6 +487,7 @@ class WifiDevice(Device):
         self.aps.append(ap)
         self.__notify(PW_ACCESS_POINTS)
         self.AccessPointAdded(to_path(ap))
+        return ap
 
     @dbus.service.signal(IFACE_WIFI, signature='o')
     def AccessPointRemoved(self, ap_path):
@@ -450,12 +517,6 @@ class WifiDevice(Device):
     def PropertiesChanged(self, changed):
         pass
 
-    # test functions
-    def add_test_ap(self, ssid, mac):
-        ap = WifiAp(self._bus, ssid, mac, 0x1, 0x1cc, 0x1cc, 2412)
-        self.add_ap(ap)
-        return ap
-
     def remove_ap_by_path(self, path):
         for ap in self.aps:
             if ap.path == path:
@@ -526,14 +587,14 @@ PX_BSID = "Bsid"
 PX_ACTIVE_NSP = "ActiveNsp"
 
 class WimaxDevice(Device):
-    def __init__(self, bus, iface):
-        self.mac = random_mac()
-        self.bsid = random_mac()
+    def __init__(self, bus, iface, ident = None):
+        self.mac = random_mac(iface if ident is None else ident)
+        self.bsid = random_mac(iface if ident is None else ident)
         self.nsps = []
         self.active_nsp = None
 
         self.add_dbus_interface(IFACE_WIMAX, self.__get_props, WimaxDevice.PropertiesChanged)
-        Device.__init__(self, bus, iface, NM_DEVICE_TYPE_WIMAX)
+        Device.__init__(self, bus, iface, NM_DEVICE_TYPE_WIMAX, ident)
 
     # methods
     @dbus.service.method(dbus_interface=IFACE_WIMAX, in_signature='', out_signature='ao')
@@ -841,11 +902,20 @@ class NetworkManager(ExportedObj):
     def DeviceAdded(self, devpath):
         pass
 
+    def find_device(self, ident):
+        for d in self.devices:
+            if d.ident == ident:
+                return d
+
     def add_device(self, device):
+        d = self.find_device(device.ident)
+        if d:
+            raise TestError("Device with ident=%s already added (%s)" % (device.ident, d.path))
         self.devices.append(device)
         self.__notify(PM_DEVICES)
         self.__notify(PM_ALL_DEVICES)
         self.DeviceAdded(to_path(device))
+        return device
 
     @dbus.service.signal(IFACE_NM, signature='o')
     def DeviceRemoved(self, devpath):
@@ -890,32 +960,43 @@ class NetworkManager(ExportedObj):
     def Quit(self):
         mainloop.quit()
 
+    @dbus.service.method(IFACE_TEST, in_signature='a{ss}', out_signature='a(sss)')
+    def FindConnections(self, args):
+        return [(c.path, c.get_uuid(), c.get_id()) for c in settings.find_connections(**args)]
+
+    @dbus.service.method(IFACE_TEST, in_signature='sa{sv}', out_signature='o')
+    def AddObj(self, class_name, args):
+        if class_name in ['WiredDevice', 'WifiDevice']:
+            py_class = globals()[class_name]
+            d = py_class(self._bus, **args)
+            return to_path(self.add_device(d))
+        elif class_name in ['WifiAp']:
+            if 'device' not in args:
+                raise TestError('missing "device" paramter')
+            d = self.find_device(args['device'])
+            if not d:
+                raise TestError('no device "%s" found' % args['device'])
+            del args['device']
+            if 'ssid' not in args:
+                args['ssid'] = d.ident + '-ap-' + str(WifiAp.counter + 1)
+            ap = WifiAp(self._bus, **args)
+            return to_path(d.add_ap(ap))
+        raise TestError("Invalid python type \"%s\"" % (class_name))
+
     @dbus.service.method(IFACE_TEST, in_signature='ssas', out_signature='o')
     def AddWiredDevice(self, ifname, mac, subchannels):
-        for d in self.devices:
-            if d.iface == ifname:
-                raise PermissionDeniedException("Device already added")
         dev = WiredDevice(self._bus, ifname, mac, subchannels)
-        self.add_device(dev)
-        return to_path(dev)
+        return to_path(self.add_device(dev))
 
     @dbus.service.method(IFACE_TEST, in_signature='s', out_signature='o')
     def AddWifiDevice(self, ifname):
-        for d in self.devices:
-            if d.iface == ifname:
-                raise PermissionDeniedException("Device already added")
         dev = WifiDevice(self._bus, ifname)
-        self.add_device(dev)
-        return to_path(dev)
+        return to_path(self.add_device(dev))
 
     @dbus.service.method(IFACE_TEST, in_signature='s', out_signature='o')
     def AddWimaxDevice(self, ifname):
-        for d in self.devices:
-            if d.iface == ifname:
-                raise PermissionDeniedException("Device already added")
         dev = WimaxDevice(self._bus, ifname)
-        self.add_device(dev)
-        return to_path(dev)
+        return to_path(self.add_device(dev))
 
     @dbus.service.method(IFACE_TEST, in_signature='o', out_signature='')
     def RemoveDevice(self, path):
@@ -926,10 +1007,11 @@ class NetworkManager(ExportedObj):
         raise UnknownDeviceException("Device not found")
 
     @dbus.service.method(IFACE_TEST, in_signature='sss', out_signature='o')
-    def AddWifiAp(self, ifname, ssid, mac):
-        for d in self.devices:
-            if d.iface == ifname:
-                return to_path(d.add_test_ap(ssid, mac))
+    def AddWifiAp(self, ifname, ssid, bssid):
+        d = self.find_device(ifname)
+        if d:
+            ap = WifiAp(self._bus, ssid, bssid)
+            return to_path(d.add_ap(ap))
         raise UnknownDeviceException("Device not found")
 
     @dbus.service.method(IFACE_TEST, in_signature='so', out_signature='')
@@ -989,15 +1071,19 @@ class MissingSettingException(dbus.DBusException):
     _dbus_error_name = IFACE_CONNECTION + '.MissingSetting'
 
 class Connection(ExportedObj):
-    def __init__(self, bus, object_path, settings, remove_func, verify_connection=True):
+    def __init__(self, bus, path_counter, settings, remove_func, verify_connection=True):
+
+        path = "/org/freedesktop/NetworkManager/Settings/Connection/%s" % (path_counter)
 
+        if 'connection' not in settings:
+            settings['connection'] = { }
+        if self.get_id(settings) is None:
+            settings['connection']['id'] = 'connection-%s' % (path_counter)
         if self.get_uuid(settings) is None:
-            if 'connection' not in settings:
-                settings['connection'] = { }
-            settings['connection']['uuid'] = uuid.uuid4()
+            settings['connection']['uuid'] = str(uuid.uuid3(uuid.NAMESPACE_URL, path))
         self.verify(settings, verify_strict=verify_connection)
 
-        self.path = object_path
+        self.path = path
         self.settings = settings
         self.remove_func = remove_func
         self.visible = True
@@ -1005,7 +1091,16 @@ class Connection(ExportedObj):
         self.props['Unsaved'] = False
 
         self.add_dbus_interface(IFACE_CONNECTION, self.__get_props, None)
-        ExportedObj.__init__(self, bus, object_path)
+        ExportedObj.__init__(self, bus, path)
+
+    def get_id(self, settings=None):
+        if settings is None:
+            settings = self.settings
+        if 'connection' in settings:
+            s_con = settings['connection']
+            if 'id' in s_con:
+                return s_con['id']
+        return None
 
     def get_uuid(self, settings=None):
         if settings is None:
@@ -1092,7 +1187,7 @@ class Settings(ExportedObj):
     def __init__(self, bus, object_path):
         self.connections = {}
         self.bus = bus
-        self.counter = 1
+        self.counter = 0
         self.remove_next_connection = False
         self.props = {}
         self.props['Hostname'] = "foobar.baz"
@@ -1108,6 +1203,19 @@ class Settings(ExportedObj):
     def get_connection(self, path):
         return self.connections[path]
 
+    def find_connections(self, path = None, con_id = None, con_uuid = None):
+        for c in self.connections.values():
+            if path is not None:
+                if c.path != path:
+                    continue
+            if con_id is not None:
+                if c.get_id() != con_id:
+                    continue
+            if con_uuid is not None:
+                if c.get_uuid() != con_uuid:
+                    continue
+            yield c
+
     @dbus.service.method(dbus_interface=IFACE_SETTINGS, in_signature='', out_signature='ao')
     def ListConnections(self):
         return self.connections.keys()
@@ -1117,24 +1225,23 @@ class Settings(ExportedObj):
         return self.add_connection(settings)
 
     def add_connection(self, settings, verify_connection=True):
-        path = "/org/freedesktop/NetworkManager/Settings/Connection/{0}".format(self.counter)
-        con = Connection(self.bus, path, settings, self.delete_connection, verify_connection)
+        self.counter += 1
+        con = Connection(self.bus, self.counter, settings, self.delete_connection, verify_connection)
 
         uuid = con.get_uuid()
         if uuid in [c.get_uuid() for c in self.connections.values()]:
             raise InvalidSettingException('cannot add duplicate connection with uuid %s' % (uuid))
 
-        self.counter = self.counter + 1
-        self.connections[path] = con
+        self.connections[con.path] = con
         self.props['Connections'] = dbus.Array(self.connections.keys(), 'o')
-        self.NewConnection(path)
+        self.NewConnection(con.path)
         self.__notify('Connections')
 
         if self.remove_next_connection:
             self.remove_next_connection = False
-            self.connections[path].Delete()
+            self.connections[con.path].Delete()
 
-        return path
+        return con.path
 
     def update_connection(self, connection, path=None, verify_connection=True):
         if path is None: