diff options
Diffstat (limited to 'clients/tests/test-client.py')
| -rwxr-xr-x | clients/tests/test-client.py | 856 |
1 files changed, 856 insertions, 0 deletions
diff --git a/clients/tests/test-client.py b/clients/tests/test-client.py new file mode 100755 index 00000000..a8e107a0 --- /dev/null +++ b/clients/tests/test-client.py @@ -0,0 +1,856 @@ +#!/usr/bin/env python + +from __future__ import print_function + +import sys + +try: + import gi + from gi.repository import GLib + + gi.require_version('NM', '1.0') + from gi.repository import NM +except Exception as e: + GLib = None + NM = None + +import os +import errno +import unittest +import socket +import itertools +import subprocess +import shlex +import re +import dbus +import time +import dbus.service +import dbus.mainloop.glib + +# The test can be configured via the following environment variables: +ENV_NM_TEST_CLIENT_BUILDDIR = 'NM_TEST_CLIENT_BUILDDIR' +ENV_NM_TEST_CLIENT_NMCLI_PATH = 'NM_TEST_CLIENT_NMCLI_PATH' +ENV_NM_TEST_CLIENT_CHECK_L10N = 'NM_TEST_CLIENT_CHECK_L10N' +ENV_NM_TEST_REGENERATE = 'NM_TEST_REGENERATE' + +############################################################################### + +class PathConfiguration: + + @staticmethod + def srcdir(): + # this is the directory where the test script itself lies. + # Based on this directory, we find other parts that we expect + # in the source repository. + return os.path.dirname(os.path.abspath(__file__)) + + @staticmethod + def top_srcdir(): + return os.path.abspath(PathConfiguration.srcdir() + "/../..") + + @staticmethod + def test_networkmanager_service_path(): + v = os.path.abspath(PathConfiguration.top_srcdir() + "/tools/test-networkmanager-service.py") + assert os.path.exists(v), ("Cannot find test server at \"%s\"" % (v)) + return v + +############################################################################### + +os.sys.path.append(os.path.abspath(PathConfiguration.top_srcdir() + '/examples/python')) +import nmex + +dbus_session_inited = False + +_DEFAULT_ARG = object() + +############################################################################### + +class Util: + + @staticmethod + def python_has_version(major, minor = 0): + return sys.version_info[0] > major \ + or ( sys.version_info[0] == major \ + and sys.version_info[1] >= minor) + + @staticmethod + def is_string(s): + if Util.python_has_version(3): + t = str + else: + t = basestring + return isinstance(s, t) + + _find_unsafe = re.compile(r'[^\w@%+=:,./-]', + re.ASCII if sys.version_info[0] >= 3 else 0).search + + @staticmethod + def quote(s): + if Util.python_has_version(3, 3): + return shlex.quote(s) + if not s: + return "''" + if Util._find_unsafe(s) is None: + return s + return "'" + s.replace("'", "'\"'\"'") + "'" + + @staticmethod + def popen_wait(p, timeout = None): + # wait() has a timeout argument only since 3.3 + if Util.python_has_version(3, 3): + return p.wait(timeout) + if timeout is None: + return p.wait() + start = nmex.nm_boot_time_ns() + while True: + if p.poll() is not None: + return p.returncode + if start + (timeout * 1000000000) < nmex.nm_boot_time_ns(): + raise Exception("timeout expired") + time.sleep(0.05) + + @staticmethod + def iter_single(itr, min_num = 1, max_num = 1): + itr = list(itr) + n = 0 + v = None + for c in itr: + n += 1 + if n > 1: + break + v = c + if n < min_num: + raise AssertionError("Expected at least %s elements, but %s found" % (min_num, n)) + if n > max_num: + raise AssertionError("Expected at most %s elements, but %s found" % (max_num, n)) + return v + + @staticmethod + def file_read(filename): + try: + with open(filename, 'rb') as f: + return f.read() + except: + return None + + @staticmethod + def replace_text(text, replace_arr): + if not replace_arr: + return text + text = [text] + for replace in replace_arr: + try: + v_search = replace[0]() + except TypeError: + v_search = replace[0] + assert v_search is None or Util.is_string(v_search) + if not v_search: + continue + v_replace = replace[1] + v_search = v_search.encode('utf-8') + v_replace = v_replace.encode('utf-8') + text2 = [] + for t in text: + if isinstance(t, tuple): + text2.append(t) + continue + t2 = t.split(v_search) + text2.append(t2[0]) + for t3 in t2[1:]: + text2.append( (v_replace,) ) + text2.append(t3) + text = text2 + return b''.join([(t[0] if isinstance(t, tuple) else t) for t in text]) + +############################################################################### + +class Configuration: + + def __init__(self): + self._values = {} + + def get(self, name): + v = self._values.get(name, None) + if name in self._values: + return v + if name == ENV_NM_TEST_CLIENT_BUILDDIR: + v = os.environ.get(ENV_NM_TEST_CLIENT_BUILDDIR, PathConfiguration.top_srcdir()) + if not os.path.isdir(v): + raise Exception("Missing builddir. Set NM_TEST_CLIENT_BUILDDIR?") + elif name == ENV_NM_TEST_CLIENT_NMCLI_PATH: + v = os.environ.get(ENV_NM_TEST_CLIENT_NMCLI_PATH, None) + if v is None: + try: + v = os.path.abspath(self.get(ENV_NM_TEST_CLIENT_BUILDDIR) + "/clients/cli/nmcli") + except: + pass + if not os.path.exists(v): + raise Exception("Missing nmcli binary. Set NM_TEST_CLIENT_NMCLI_PATH?") + elif name == ENV_NM_TEST_CLIENT_CHECK_L10N: + # if we test locales other than 'C', the output of nmcli depends on whether + # nmcli can load the translations. Unfortunately, I cannot find a way to + # make gettext use the po/*.gmo files from the build-dir. + # + # hence, such tests only work, if you also issue `make-install` + # + # Only by setting NM_TEST_CLIENT_CHECK_L10N=1, these tests are included + # as well. + v = (os.environ.get(ENV_NM_TEST_CLIENT_CHECK_L10N, '0') == '1') + elif name == ENV_NM_TEST_REGENERATE: + # in the "regenerate" mode, the tests will rewrite the files on disk against + # which we assert. That is useful, if there are intentional changes and + # we want to regenerate the expected output. + v = (os.environ.get(ENV_NM_TEST_REGENERATE, '0') == '1') + else: + raise Exception() + self._values[name] = v + return v + +conf = Configuration() + +############################################################################### + +class NMStubServer: + + @staticmethod + def _conn_get_main_object(conn): + try: + return conn.get_object('org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager') + except: + return None + + def __init__(self): + service_path = PathConfiguration.test_networkmanager_service_path() + self._conn = dbus.SessionBus() + p = subprocess.Popen([sys.executable, service_path], + stdin = subprocess.PIPE) + + start = nmex.nm_boot_time_ns() + while True: + if p.poll() is not None: + p.stdin.close() + if p.returncode == 77: + raise unittest.SkipTest('the stub service %s exited with status 77' % (service_path)) + raise Exception('the stub service %s exited unexpectedly' % (service_path)) + nmobj = self._conn_get_main_object(self._conn) + if nmobj is not None: + break + if (nmex.nm_boot_time_ns() - start) / 1000000 >= 2000: + p.stdin.close() + p.kill() + Util.popen_wait(p, 1000) + raise Exception("after starting stub service the D-Bus name was not claimed in time") + + self._nmobj = nmobj + self._nmiface = dbus.Interface(nmobj, "org.freedesktop.NetworkManager.LibnmGlibTest") + self._p = p + + def shutdown(self): + self._nmobj = None + self._nmiface = None + self._conn = None + self._p.stdin.close() + self._p.kill() + Util.popen_wait(self._p, 1000) + self._p = None + if self._conn_get_main_object(self._conn) is not None: + raise Exception("Stub service is not still here although it should shut down") + + class _MethodProxy: + def __init__(self, parent, method_name): + self._parent = parent + self._method_name = method_name + def __call__(self, *args, **kwargs): + dbus_iface = kwargs.pop('dbus_iface', None) + if dbus_iface is None: + dbus_iface = self._parent._nmiface + method = dbus_iface.get_dbus_method(self._method_name) + if kwargs: + # for convenience, we allow the caller to specify arguments + # as kwargs. In this case, we construct a a{sv} array as last argument. + kwargs2 = {} + args = list(args) + args.append(kwargs2) + for k in kwargs.keys(): + kwargs2[k] = kwargs[k] + return method(*args) + + def __getattr__(self, member): + if not member.startswith("op_"): + raise AttributeError(member) + return self._MethodProxy(self, member[3:]) + + def addConnection(self, connection, verify_connection = True): + return self.op_AddConnection(connection, verify_connection) + + def findConnectionUuid(self, con_id): + try: + u = Util.iter_single(self.op_FindConnections(con_id = con_id))[1] + assert u, ("Invalid uuid %s" % (u)) + except Exception as e: + raise AssertionError("Unexpectedly not found connection %s: %s" % (con_id, str(e))) + return u + + def setProperty(self, path, propname, value, iface_name = None): + if iface_name is None: + iface_name = '' + self.op_SetProperties([ + (path, [ + (iface_name, [ + (propname, value), + ]), + ]), + ]) + +############################################################################### + +class AsyncProcess(): + + def __init__(self, + args, + env, + complete_cb): + self._args = args + self._env = env + self._complete_cb = complete_cb + + def start(self): + if not hasattr(self, '_p'): + self._p = subprocess.Popen(self._args, + stdout = subprocess.PIPE, + stderr = subprocess.PIPE, + env = self._env) + + def wait(self): + + self.start() + + Util.popen_wait(self._p, 2000) + + (returncode, stdout, stderr) = (self._p.returncode, self._p.stdout.read(), self._p.stderr.read()) + + self._p.stdout.close() + self._p.stderr.close() + self._p = None + + self._complete_cb(self, returncode, stdout, stderr) + +############################################################################### + +file_list = [] + +class NmTestBase(unittest.TestCase): + pass + +class TestNmcli(NmTestBase): + + def call_nmcli_l(self, + args, + check_on_disk = _DEFAULT_ARG, + expected_returncode = _DEFAULT_ARG, + expected_stdout = _DEFAULT_ARG, + expected_stderr = _DEFAULT_ARG, + replace_stdout = None, + replace_stderr = None, + sort_lines_stdout = False, + extra_env = None, + sync_barrier = False): + frame = sys._getframe(1) + for lang in [ 'C', 'pl' ]: + self._call_nmcli(args, + lang, + check_on_disk, + expected_returncode, + expected_stdout, + expected_stderr, + replace_stdout, + replace_stderr, + sort_lines_stdout, + extra_env, + sync_barrier, + frame) + + + def call_nmcli(self, + args, + langs = None, + lang = None, + check_on_disk = _DEFAULT_ARG, + expected_returncode = _DEFAULT_ARG, + expected_stdout = _DEFAULT_ARG, + expected_stderr = _DEFAULT_ARG, + replace_stdout = None, + replace_stderr = None, + sort_lines_stdout = False, + extra_env = None, + sync_barrier = None): + + frame = sys._getframe(1) + + if langs is not None: + assert lang is None + else: + if lang is None: + lang = 'C' + langs = [lang] + + if sync_barrier is None: + sync_barrier = (len(langs) == 1) + + for lang in langs: + self._call_nmcli(args, + lang, + check_on_disk, + expected_returncode, + expected_stdout, + expected_stderr, + replace_stdout, + replace_stderr, + sort_lines_stdout, + extra_env, + sync_barrier, + frame) + + def _call_nmcli(self, + args, + lang, + check_on_disk, + expected_returncode, + expected_stdout, + expected_stderr, + replace_stdout, + replace_stderr, + sort_lines_stdout, + extra_env, + sync_barrier, + frame): + + if sync_barrier: + self.async_wait() + + calling_fcn = frame.f_code.co_name + calling_num = self._calling_num.get(calling_fcn, 0) + 1 + self._calling_num[calling_fcn] = calling_num + + test_name = '%s-%03d' % (calling_fcn, calling_num) + + # we cannot use frame.f_code.co_filename directly, because it might be different depending + # on where the file lies and which is CWD. We still want to give the location of + # the file, so that the user can easier find the source (when looking at the .expected files) + script_filename = 'clients/tests/test-client.py' + self.assertTrue(os.path.abspath(frame.f_code.co_filename).endswith(script_filename)) + + calling_location = '%s:%d:%s()/%d' % (script_filename, frame.f_lineno, frame.f_code.co_name, calling_num) + + if lang is None or lang == 'C': + lang = 'C' + language = '' + elif lang is 'de': + lang = 'de_DE.utf8' + language = 'de' + elif lang is 'pl': + lang = 'pl_PL.UTF-8' + language = 'pl' + else: + self.fail('invalid language %s' % (lang)) + + env = {} + if extra_env is not None: + for k, v in extra_env.items(): + env[k] = v + for k in ['LD_LIBRARY_PATH', + 'DBUS_SESSION_BUS_ADDRESS']: + val = os.environ.get(k, None) + if val is not None: + env[k] = val + env['LANG'] = lang + env['LANGUAGE'] = language + env['LIBNM_USE_SESSION_BUS'] = '1' + env['LIBNM_USE_NO_UDEV'] = '1' + env['TERM'] = 'linux' + + args = [conf.get(ENV_NM_TEST_CLIENT_NMCLI_PATH)] + list(args) + + if replace_stdout is not None: + replace_stdout = list(replace_stdout) + if replace_stderr is not None: + replace_stderr = list(replace_stderr) + + if check_on_disk is _DEFAULT_ARG: + check_on_disk = ( expected_returncode is _DEFAULT_ARG + and expected_stdout is _DEFAULT_ARG + and expected_stderr is _DEFAULT_ARG) + if expected_returncode is _DEFAULT_ARG: + expected_returncode = None + if expected_stdout is _DEFAULT_ARG: + expected_stdout = None + if expected_stderr is _DEFAULT_ARG: + expected_stderr = None + + def complete_cb(async_job, + returncode, + stdout, + stderr): + + stdout = Util.replace_text(stdout, replace_stdout) + stderr = Util.replace_text(stderr, replace_stderr) + + if sort_lines_stdout: + stdout = b'\n'.join(sorted(stdout.split(b'\n'))) + + ignore_l10n_diff = ( lang != 'C' + and not conf.get(ENV_NM_TEST_CLIENT_CHECK_L10N)) + + if expected_stderr is not None: + if expected_stderr != stderr: + if ignore_l10n_diff: + self._skip_test_for_l10n_diff.append(test_name) + else: + self.assertEqual(expected_stderr, stderr) + if expected_stdout is not None: + if expected_stdout != stdout: + if ignore_l10n_diff: + self._skip_test_for_l10n_diff.append(test_name) + else: + self.assertEqual(expected_stdout, stdout) + if expected_returncode is not None: + self.assertEqual(expected_returncode, returncode) + + dirname = PathConfiguration.srcdir() + '/test-client.check-on-disk' + basename = test_name + '.expected' + filename = os.path.abspath(dirname + '/' + basename) + + if not check_on_disk: + if os.path.exists(filename): + self.fail("The file '%s' exists, although we expect it not to." % (filename)) + return + + file_list.append(basename) + + content_old = Util.file_read(filename) + + content_new = ('location: %s\n' % (calling_location)).encode('utf8') + \ + ('cmd: $NMCLI %s\n' % (' '.join([Util.quote(a) for a in args[1:]]))).encode('utf8') + \ + ('lang: %s\n' % (lang)).encode('utf8') + \ + ('returncode: %d\n' % (returncode)).encode('utf8') + \ + ('stdout: %d bytes\n>>>\n' % (len(stdout))).encode('utf8') + \ + stdout + \ + ('\n<<<\nstderr: %d bytes\n>>>\n' % (len(stderr))).encode('utf8') + \ + stderr + \ + '\n<<<\n'.encode('utf8') + + w = conf.get(ENV_NM_TEST_REGENERATE) + + if content_old is not None: + if content_old == content_new: + return + + if not w: + if ignore_l10n_diff: + self._skip_test_for_l10n_diff.append(test_name) + return + print("\n\n\nThe file '%s' does not have the expected content:" % (filename)) + print("ACTUAL OUTPUT:\n[[%s]]\n" % (content_new)) + print("EXPECT OUTPUT:\n[[%s]]\n" % (content_old)) + print("Let the test write the file by rerunning with NM_TEST_REGENERATE=1\n\n") + raise AssertionError("Unexpected output of command, expected %s. Rerun test with NM_TEST_REGENERATE=1 to regenerate files" % (filename)) + else: + if not w: + self.fail("The file '%s' does not exist. Let the test write the file by rerunning with NM_TEST_REGENERATE=1" % (filename)) + + try: + if not os.path.exists(dirname): + os.makedirs(dirname) + with open(filename, 'wb') as content_file: + content_file.write(content_new) + except Exception as e: + self.fail("Failure to write '%s': %s" % (filename, e)) + + async_job = AsyncProcess(args = args, + env = env, + complete_cb = complete_cb) + + self._async_jobs.append(async_job) + + if sync_barrier: + self.async_wait() + else: + self.async_start() + + def async_start(self): + # limit number parallel running jobs + for async_job in self._async_jobs[0:10]: + async_job.start() + + def async_wait(self): + while self._async_jobs: + self.async_start() + self._async_jobs.pop(0).wait() + + def setUp(self): + if not dbus_session_inited: + self.skipTest("Own D-Bus session for testing is not initialized. Do you have dbus-run-session available?") + if NM is None: + self.skipTest("gi.NM is not available. Did you build with introspection?") + self.srv = NMStubServer() + self._calling_num = {} + self._skip_test_for_l10n_diff = [] + self._async_jobs = [] + + def tearDown(self): + self.async_wait() + self.srv.shutdown() + self.srv = None + self._calling_num = None + if self._skip_test_for_l10n_diff: + # nmcli loads translations from the installation path. This failure commonly + # happens because you did not install the binary in the --prefix, before + # running the test. Hence, translations are not available or differ. + msg = "Skipped asserting for localized tests %s. Set NM_TEST_CLIENT_CHECK_L10N=1 to force fail." % (','.join(self._skip_test_for_l10n_diff)) + if Util.python_has_version(3): + # python2 does not suppot skipping the test during tearDown() + self.skipTest(msg) + print(msg + "\n") + self._skip_test_for_l10n_diff = None + + def init_001(self): + self.srv.op_AddObj('WiredDevice', + iface = 'eth0') + self.srv.op_AddObj('WiredDevice', + iface = 'eth1') + self.srv.op_AddObj('WifiDevice', + iface = 'wlan0') + self.srv.op_AddObj('WifiDevice', + iface = 'wlan1') + + # add another device with an identical ifname. The D-Bus API itself + # does not enforce the ifnames are unique. + self.srv.op_AddObj('WifiDevice', + ident = 'wlan1/x', + iface = 'wlan1') + + self.srv.op_AddObj('WifiAp', + device = 'wlan0') + self.srv.op_AddObj('WifiAp', + device = 'wlan0') + self.srv.op_AddObj('WifiAp', + device = 'wlan0') + + self.srv.op_AddObj('WifiAp', + device = 'wlan1') + + self.srv.addConnection( { + 'connection': { + 'type': '802-3-ethernet', + 'id': 'con-1', + }, + }) + + def test_001(self): + + self.call_nmcli_l([]) + + self.call_nmcli_l(['-f', 'AP', '-mode', 'multiline', '-p', 'd', 'show', 'wlan0']) + + self.call_nmcli_l(['c', 's']) + + self.call_nmcli_l(['bogus', 's']) + + for mode in [[], + ['--mode', 'tabular'], + ['--mode', 'multiline']]: + for fmt in [[], + ['--pretty'], + ['--terse']]: + self.call_nmcli_l(mode + fmt + ['general', 'permissions']) + + def test_002(self): + self.init_001() + + self.call_nmcli_l(['d']) + + self.call_nmcli_l(['-f', 'all', 'd']) + + self.call_nmcli_l([]) + + self.call_nmcli_l(['-f', 'AP', '-mode', 'multiline', 'd', 'show', 'wlan0']) + self.call_nmcli_l(['-f', 'AP', '-mode', 'multiline', '-p', 'd', 'show', 'wlan0']) + self.call_nmcli_l(['-f', 'AP', '-mode', 'multiline', '-t', 'd', 'show', 'wlan0']) + self.call_nmcli_l(['-f', 'AP', '-mode', 'tabular', 'd', 'show', 'wlan0']) + self.call_nmcli_l(['-f', 'AP', '-mode', 'tabular', '-p', 'd', 'show', 'wlan0']) + self.call_nmcli_l(['-f', 'AP', '-mode', 'tabular', '-t', 'd', 'show', 'wlan0']) + + self.call_nmcli_l(['-f', 'ALL', 'd', 'wifi']) + + self.call_nmcli_l(['c']) + + self.call_nmcli_l(['c', 's', 'con-1']) + + def test_003(self): + self.init_001() + + replace_stdout = [] + + replace_stdout.append((lambda: self.srv.findConnectionUuid('con-xx1'), 'UUID-con-xx1-REPLACED-REPLACED-REPLA')) + + self.call_nmcli(['c', 'add', 'type', 'ethernet', 'ifname', '*', 'con-name', 'con-xx1'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['c', 's'], + replace_stdout = replace_stdout) + + replace_stdout.append((lambda: self.srv.findConnectionUuid('ethernet'), 'UUID-ethernet-REPLACED-REPLACED-REPL')) + + self.call_nmcli(['c', 'add', 'type', 'ethernet', 'ifname', '*'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['c', 's'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'ALL', 'c', 's'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['--complete-args', '-f', 'ALL', 'c', 's', ''], + replace_stdout = replace_stdout, + sort_lines_stdout = True) + + # activate the same profile on multiple devices. Our stub-implmentation + # is fine with that... although NetworkManager service would reject + # such a configuration by deactivating the profile first. But note that + # that is only an internal behavior of NetworkManager service. The D-Bus + # API perfectly allows for one profile to be active multiple times. Also + # note, that there is always a short time where one profile goes down, + # while another is activating. Hence, while real NetworkManager commonly + # does not allow that multiple profiles *stay* connected at the same + # time, there is always the possibility that a profile is activating/active + # on a device, while also activating/deactivating in parallel. + for dev in ['eth0', 'eth1']: + self.call_nmcli(['con', 'up', 'ethernet', 'ifname', dev]) + + self.call_nmcli_l(['con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'ALL', 'con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'ALL', 'con', 's', 'ethernet'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['con', 's', 'ethernet'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'ALL', 'dev', 's', 'eth0'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'ALL', 'dev', 'show', 'eth0'], + replace_stdout = replace_stdout) + + self.async_wait() + + self.srv.setProperty('/org/freedesktop/NetworkManager/ActiveConnection/1', + 'State', + dbus.UInt32(NM.ActiveConnectionState.DEACTIVATING)) + + self.call_nmcli_l(['-f', 'ALL', 'con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'UUID,TYPE', 'con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'UUID,TYPE', '--mode', 'multiline', 'con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'UUID,TYPE', '--mode', 'multiline', '--terse', 'con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'UUID,TYPE', '--mode', 'multiline', '--pretty', 'con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'UUID,TYPE', '--mode', 'tabular', 'con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'UUID,TYPE', '--mode', 'tabular', '--terse', 'con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['-f', 'UUID,TYPE', '--mode', 'tabular', '--pretty', 'con'], + replace_stdout = replace_stdout) + + self.call_nmcli_l(['con', 's', 'ethernet'], + replace_stdout = replace_stdout) + + def test_004(self): + self.init_001() + + replace_stdout = [] + + replace_stdout.append((lambda: self.srv.findConnectionUuid('con-xx1'), 'UUID-con-xx1-REPLACED-REPLACED-REPLA')) + + self.call_nmcli(['c', 'add', 'type', 'wifi', 'ifname', '*', 'ssid', 'foobar', 'con-name', 'con-xx1'], + replace_stdout = replace_stdout) + + self.call_nmcli(['connection', 'mod', 'con-xx1', 'ip.gateway', '']) + self.call_nmcli(['connection', 'mod', 'con-xx1', 'ipv4.gateway', '172.16.0.1']) + self.call_nmcli(['connection', 'mod', 'con-xx1', 'ipv6.gateway', '::99']) + self.call_nmcli(['connection', 'mod', 'con-xx1', '802.abc', '']) + self.call_nmcli(['connection', 'mod', 'con-xx1', '802-11-wireless.band', 'a']) + +############################################################################### + +def main(): + global dbus_session_inited + + if len(sys.argv) >= 2 and sys.argv[1] == '--started-with-dbus-session': + dbus_session_inited = True + del sys.argv[1] + + if not dbus_session_inited: + # we don't have yet our own dbus-session. Reexec ourself with + # a new dbus-session. + try: + try: + os.execlp('dbus-run-session', 'dbus-run-session', '--', sys.executable, __file__, '--started-with-dbus-session', *sys.argv[1:]) + except OSError as e: + if e.errno != errno.ENOENT: + raise + # we have no dbus-run-session in path? Fall-through + # to skip tests gracefully + else: + raise Exception('unknown error during exec') + except Exception as e: + assert False, ("Failure to re-exec dbus-run-session: %s" % (str(e))) + + if not dbus_session_inited: + # we still don't have a D-Bus session. Probably dbus-run-session is not available. + # retry with dbus-launch + if os.system('type dbus-launch 1>/dev/null') == 0: + try: + os.execlp('bash', 'bash', '-e', '-c', + 'eval `dbus-launch --sh-syntax`;\n' + \ + 'trap "kill $DBUS_SESSION_BUS_PID" EXIT;\n' + \ + '\n' + \ + ' '.join([Util.quote(a) for a in [sys.executable, __file__, '--started-with-dbus-session'] + sys.argv[1:]]) + ' \n' + \ + '') + except Exception as e: + m = str(e) + else: + m = 'unknown error' + assert False, ('Failure to re-exec to start script with dbus-launch: %s' % (m)) + + r = unittest.main(exit = False) + + if conf.get(ENV_NM_TEST_REGENERATE): + make_filename = PathConfiguration.srcdir() + '/test-client.check-on-disk/Makefile.am' + s_new = '# generated with `NM_TEST_REGENERATE=1 make check`\n' + \ + '\n' + \ + 'clients_tests_expected_files = \\\n' + \ + ''.join([('\tclients/tests/test-client.check-on-disk/%s \\\n' % f) for f in sorted(file_list)]) + \ + '\t$(NULL)\n' + s_new = s_new.encode('utf-8') + if s_new != Util.file_read(make_filename): + with open(make_filename, 'wb') as f: + f.write(s_new) + + sys.exit(not r.result.wasSuccessful()) + +if __name__ == '__main__': + main() |