#!/usr/bin/env python from __future__ import print_function ############################################################################### # # This test starts NetworkManager stub service in a user D-Bus session, # and runs nmcli against it. The output is recorded and compared to a pre-generated # expected output (clients/tests/test-client.check-on-disk/*.expected) which # is also committed to git. # ############################################################################### # # HOWTO: Regenerate output # # When adjusting the tests, or when making changes to nmcli that intentionally # change the output, the expected output must be regenerated. # # $ make install # # (step not required every time) # # The test also compare the translated output, hence, the translation # # file must be installed at the configured --prefix. # # You don't need to type `make install` every time, but a suitable version # # of translations must be installed. In practice, the tests only care about # # Polish (pl) translations. # # The important part is that translations work. Test # # $ LANG=pl_PL.UTF-8 ./clients/cli/nmcli --version # # also ensure that `locale -a` reports the Polish locale. # $ rm -rf clients/tests/test-client.check-on-disk/*.expected # # (step seldom required) # # Sometimes, if you want to be sure that the test would generate # # exactly the same .expected files, purge the previous version first. # # This is only necessary, when you remove test from this file. # $ NM_TEST_REGENERATE=1 make check-local-clients-tests-test-client # # Set NM_TEST_REGENERATE=1 to regenerate all files. # $ git diff ... ; git add ... # # (optional step) # # Inspect what changed, and whether it makes sense. Then commit changes # # to git. # ############################################################################### # # Environment variables to configure test: # (optional) The build dir. Optional, mainly used to find the nmcli binary (in case # ENV_NM_TEST_CLIENT_NMCLI_PATH is not set. ENV_NM_TEST_CLIENT_BUILDDIR = 'NM_TEST_CLIENT_BUILDDIR' # (optional) Path to nmcli. By default, it looks for nmcli in build dir. # In particular, you can test also a nmcli binary installed somewhere else. ENV_NM_TEST_CLIENT_NMCLI_PATH = 'NM_TEST_CLIENT_NMCLI_PATH' # (optional) The test also compares tranlsated output (l10n). This requires, # that you first install the translation in the right place. So, by default, # if a test for a translation fails, it will mark the test as skipped, and not # fail the tests. Under the assumption, that the test cannot succeed currently. # By setting NM_TEST_CLIENT_CHECK_L10N=1, you can force a failure of the test. ENV_NM_TEST_CLIENT_CHECK_L10N = 'NM_TEST_CLIENT_CHECK_L10N' # Regenerate the .expected files. Instead of asserting, rewrite the files # on disk with the expected output. ENV_NM_TEST_REGENERATE = 'NM_TEST_REGENERATE' # ############################################################################### 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 ############################################################################### 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 @staticmethod def canonical_script_filename(): p = 'clients/tests/test-client.py' assert (PathConfiguration.top_srcdir() + '/' + p) == os.path.abspath(__file__) return p ############################################################################### dbus_session_inited = False _DEFAULT_ARG = object() _UNSTABLE_OUTPUT = 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) @staticmethod def memoize_nullary(nullary_func): result = [] def closure(): if not result: result.append(nullary_func()) return result[0] return closure _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 = NM.utils_get_timestamp_msec() while True: if p.poll() is not None: return p.returncode if start + (timeout * 1000) < NM.utils_get_timestamp_msec(): 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]) @staticmethod def debug_dbus_interface(): # this is for printf debugging, not used in actual code. os.system('busctl --user --verbose call org.freedesktop.NetworkManager /org/freedesktop org.freedesktop.DBus.ObjectManager GetManagedObjects | cat') @staticmethod def iter_nmcli_output_modes(): for mode in [[], ['--mode', 'tabular'], ['--mode', 'multiline']]: for fmt in [[], ['--pretty'], ['--terse']]: for color in [[], ['--color', 'yes']]: yield mode + fmt + color ############################################################################### 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, seed): service_path = PathConfiguration.test_networkmanager_service_path() self._conn = dbus.SessionBus() env = os.environ.copy() env['NM_TEST_NETWORKMANAGER_SERVICE_SEED'] = seed p = subprocess.Popen([sys.executable, service_path], stdin = subprocess.PIPE, env = env) start = NM.utils_get_timestamp_msec() 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 (NM.utils_get_timestamp_msec() - start) >= 4000: 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, do_verify_strict = True): return self.op_AddConnection(connection, do_verify_strict) def findConnectionUuid(self, con_id, required = True): try: u = Util.iter_single(self.op_FindConnections(con_id = con_id))[1] assert u, ("Invalid uuid %s" % (u)) except Exception as e: if not required: return None 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) ############################################################################### class NmTestBase(unittest.TestCase): pass class TestNmcli(NmTestBase): @staticmethod def _read_expected(filename): results_expect = [] content_expect = Util.file_read(filename) try: base_idx = 0 size_prefix = 'size: '.encode('utf8') while True: if not content_expect[base_idx:base_idx + 10].startswith(size_prefix): raise Exception("Unexpected token") j = base_idx + len(size_prefix) i = j if Util.python_has_version(3, 0): eol = ord('\n') else: eol = '\n' while content_expect[i] != eol: i += 1 i = i + 1 + int(content_expect[j:i]) results_expect.append(content_expect[base_idx:i]) if len(content_expect) == i: break base_idx = i except Exception as e: results_expect = None return content_expect, results_expect def call_nmcli_l(self, args, check_on_disk = _DEFAULT_ARG, fatal_warnings = _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, fatal_warnings, 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, fatal_warnings = _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, fatal_warnings, 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, fatal_warnings, 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) self.assertTrue(os.path.abspath(frame.f_code.co_filename).endswith('/'+PathConfiguration.canonical_script_filename())) calling_location = '%s:%d:%s()/%d' % (PathConfiguration.canonical_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' env['ASAN_OPTIONS'] = 'detect_leaks=0' env['XDG_CONFIG_HOME'] = PathConfiguration.srcdir() if fatal_warnings is _DEFAULT_ARG or fatal_warnings: env['G_DEBUG'] = 'fatal-warnings' 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 or expected_stdout is _UNSTABLE_OUTPUT) and (expected_stderr is _DEFAULT_ARG or expected_stderr is _UNSTABLE_OUTPUT)) 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): if expected_stdout is _UNSTABLE_OUTPUT: stdout = ''.encode('utf-8') else: stdout = Util.replace_text(stdout, replace_stdout) if expected_stderr is _UNSTABLE_OUTPUT: stderr = ''.encode('utf-8') else: 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 and expected_stderr is not _UNSTABLE_OUTPUT: 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 and expected_stdout is not _UNSTABLE_OUTPUT: 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) if fatal_warnings is _DEFAULT_ARG: if expected_returncode != -5: self.assertNotEqual(returncode, -5) elif fatal_warnings: if expected_returncode is None: self.assertEqual(returncode, -5) if check_on_disk: cmd = '$NMCLI %s' % (' '.join([Util.quote(a) for a in args[1:]])), content = ('location: %s\n' % (calling_location)).encode('utf8') + \ ('cmd: %s\n' % (cmd)).encode('utf8') + \ ('lang: %s\n' % (lang)).encode('utf8') + \ ('returncode: %d\n' % (returncode)).encode('utf8') if len(stdout) > 0: content += ('stdout: %d bytes\n>>>\n' % (len(stdout))).encode('utf8') + \ stdout + \ '\n<<<\n'.encode('utf8') if len(stderr) > 0: content += ('stderr: %d bytes\n>>>\n' % (len(stderr))).encode('utf8') + \ stderr + \ '\n<<<\n'.encode('utf8') content = ('size: %s\n' % (len(content))).encode('utf8') + \ content self._results.append({ 'test_name' : test_name, 'ignore_l10n_diff' : ignore_l10n_diff, 'content' : content, }) 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:15]: async_job.start() def async_wait(self): while self._async_jobs: self.async_start() self._async_jobs.pop(0).wait() def _nm_test_pre(self): self._calling_num = {} self._skip_test_for_l10n_diff = [] self._async_jobs = [] self._results = [] self.srv = NMStubServer(self._testMethodName) def _nm_test_post(self): self.async_wait() self.srv.shutdown() self.srv = None self._calling_num = None results = self._results self._results = None skip_test_for_l10n_diff = self._skip_test_for_l10n_diff self._skip_test_for_l10n_diff = None test_name = self._testMethodName filename = os.path.abspath(PathConfiguration.srcdir() + '/test-client.check-on-disk/' + test_name + '.expected') regenerate = conf.get(ENV_NM_TEST_REGENERATE) content_expect, results_expect = self._read_expected(filename) if results_expect is None: if not regenerate: self.fail("Failed to parse expected file '%s'. Let the test write the file by rerunning with NM_TEST_REGENERATE=1" % (filename)) else: for i in range(0, min(len(results_expect), len(results))): n = results[i] if results_expect[i] == n['content']: continue if regenerate: continue if n['ignore_l10n_diff']: skip_test_for_l10n_diff.append(n['test_name']) continue print("\n\n\nThe file '%s' does not have the expected content:" % (filename)) print("ACTUAL OUTPUT:\n[[%s]]\n" % (n['content'])) print("EXPECT OUTPUT:\n[[%s]]\n" % (results_expect[i])) print("Let the test write the file by rerunning with NM_TEST_REGENERATE=1") print("See howto in %s for details.\n" % (PathConfiguration.canonical_script_filename())) sys.stdout.flush() self.fail("Unexpected output of command, expected %s. Rerun test with NM_TEST_REGENERATE=1 to regenerate files" % (filename)) if len(results_expect) != len(results): if not regenerate: print("\n\n\nThe number of tests in %s does not match the expected content (%s vs %s):" % (filename, len(results_expect), len(results))) if len(results_expect) < len(results): print("ACTUAL OUTPUT:\n[[%s]]\n" % (results[len(results_expect)]['content'])) else: print("EXPECT OUTPUT:\n[[%s]]\n" % (results_expect[len(results)])) print("Let the test write the file by rerunning with NM_TEST_REGENERATE=1") print("See howto in %s for details.\n" % (PathConfiguration.canonical_script_filename())) sys.stdout.flush() self.fail("Unexpected output of command, expected %s. Rerun test with NM_TEST_REGENERATE=1 to regenerate files" % (filename)) if regenerate: content_new = ''.join([r['content'] for r in results]) if content_new != content_expect: try: 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)) if 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. self.skipTest("Skipped asserting for localized tests %s. Set NM_TEST_CLIENT_CHECK_L10N=1 to force fail." % (','.join(skip_test_for_l10n_diff))) def nm_test(func): def f(self): self._nm_test_pre() func(self) self._nm_test_post() return f 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?") 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', rsnf = 0x0) self.srv.op_AddObj('WifiAp', device = 'wlan0') NM_AP_FLAGS = getattr(NM, '80211ApSecurityFlags') rsnf = 0x0 rsnf = rsnf | NM_AP_FLAGS.PAIR_TKIP rsnf = rsnf | NM_AP_FLAGS.PAIR_CCMP rsnf = rsnf | NM_AP_FLAGS.GROUP_TKIP rsnf = rsnf | NM_AP_FLAGS.GROUP_CCMP rsnf = rsnf | NM_AP_FLAGS.KEY_MGMT_SAE self.srv.op_AddObj('WifiAp', device = 'wlan0', wpaf = 0x0, rsnf = rsnf) self.srv.op_AddObj('WifiAp', device = 'wlan1') self.srv.addConnection( { 'connection': { 'type': '802-3-ethernet', 'id': 'con-1', }, }) @nm_test 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 Util.iter_nmcli_output_modes(): self.call_nmcli_l(mode + ['general', 'permissions']) @nm_test 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']) @nm_test def test_003(self): self.init_001() replace_stdout = [] replace_stdout.append((Util.memoize_nullary(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((Util.memoize_nullary(lambda: self.srv.findConnectionUuid('con-gsm1')), 'UUID-con-gsm1-REPLACED-REPLACED-REPL')) self.call_nmcli(['connection', 'add', 'type', 'gsm', 'autoconnect', 'no', 'con-name', 'con-gsm1', 'ifname', '*', 'apn', 'xyz.con-gsm1', 'serial.baud', '5', 'serial.send-delay', '100', 'serial.pari', '1', 'ipv4.dns-options', ' '], replace_stdout = replace_stdout) replace_stdout.append((Util.memoize_nullary(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) self.call_nmcli_l(['con', 's', 'con-gsm1'], replace_stdout = replace_stdout) # 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', '-a'], replace_stdout = replace_stdout) self.call_nmcli_l(['-f', 'ACTIVE-PATH,DEVICE,UUID', 'con', 's', '-act'], replace_stdout = replace_stdout) self.call_nmcli_l(['-f', 'UUID,NAME', 'con', 's', '--active'], replace_stdout = replace_stdout) self.call_nmcli_l(['-f', 'ALL', 'con', 's', 'ethernet'], replace_stdout = replace_stdout) self.call_nmcli_l(['-f', 'GENERAL.STATE', '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', 'status'], replace_stdout = replace_stdout) self.call_nmcli_l(['-f', 'ALL', 'dev', 'show', 'eth0'], replace_stdout = replace_stdout) self.call_nmcli_l(['-f', 'ALL', '-t', 'dev', 'show', 'eth0'], replace_stdout = replace_stdout) self.async_wait() self.srv.setProperty('/org/freedesktop/NetworkManager/ActiveConnection/1', 'State', dbus.UInt32(NM.ActiveConnectionState.DEACTIVATING)) for i in [0, 1]: if i == 1: self.async_wait() self.srv.op_ConnectionSetVisible(False, con_id = 'ethernet') for mode in Util.iter_nmcli_output_modes(): self.call_nmcli_l(mode + ['-f', 'ALL', 'con'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'UUID,TYPE', 'con'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['con', 's', 'ethernet'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['c', 's', '/org/freedesktop/NetworkManager/ActiveConnection/1'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'show', 'eth0'], replace_stdout = replace_stdout) @nm_test def test_004(self): self.init_001() replace_stdout = [] replace_stdout.append((Util.memoize_nullary(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'], lang = 'pl') 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']) self.call_nmcli(['connection', 'mod', 'con-xx1', 'ipv4.addresses', '192.168.77.5/24', 'ipv4.routes', '2.3.4.5/32 192.168.77.1', 'ipv6.addresses', '1:2:3:4::6/64', 'ipv6.routes', '1:2:3:4:5:6::5/128']) self.call_nmcli_l(['con', 's', 'con-xx1'], replace_stdout = replace_stdout) self.async_wait() replace_stdout.append((Util.memoize_nullary(lambda: self.srv.findConnectionUuid('con-vpn-1')), 'UUID-con-vpn-1-REPLACED-REPLACED-REP')) self.call_nmcli(['connection', 'add', 'type', 'vpn', 'con-name', 'con-vpn-1', 'ifname', '*', 'vpn-type', 'openvpn', 'vpn.data', 'key1 = val1, key2 = val2, key3=val3'], replace_stdout = replace_stdout) self.call_nmcli_l(['con', 's'], replace_stdout = replace_stdout) self.call_nmcli_l(['con', 's', 'con-vpn-1'], replace_stdout = replace_stdout) self.call_nmcli(['con', 'up', 'con-xx1']) self.call_nmcli_l(['con', 's'], replace_stdout = replace_stdout) self.call_nmcli(['con', 'up', 'con-vpn-1']) self.call_nmcli_l(['con', 's'], replace_stdout = replace_stdout) self.call_nmcli_l(['con', 's', 'con-vpn-1'], replace_stdout = replace_stdout) self.async_wait() self.srv.setProperty('/org/freedesktop/NetworkManager/ActiveConnection/2', 'VpnState', dbus.UInt32(NM.VpnConnectionState.ACTIVATED)) for mode in Util.iter_nmcli_output_modes(): self.call_nmcli_l(mode + ['con', 's', 'con-vpn-1'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['con', 's', 'con-vpn-1'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'ALL', 'con', 's', 'con-vpn-1'], replace_stdout = replace_stdout) # This only filters 'vpn' settings from the connection profile. # Contrary to '-f GENERAL' below, it does not show the properties of # the activated VPN connection. This is a nmcli bug. self.call_nmcli_l(mode + ['-f', 'VPN', 'con', 's', 'con-vpn-1'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'GENERAL', 'con', 's', 'con-vpn-1'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['dev', 's'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'status'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['dev', 'show'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'show'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['dev', 'show', 'wlan0'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'show', 'wlan0'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'GENERAL,GENERAL.HWADDR,WIFI-PROPERTIES', 'dev', 'show', 'wlan0'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'GENERAL,GENERAL.HWADDR,WIFI-PROPERTIES', 'dev', 'show', 'wlan0'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'DEVICE,TYPE,DBUS-PATH', 'dev'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'ALL', 'device', 'wifi', 'list' ], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'COMMON', 'device', 'wifi', 'list' ], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'NAME,SSID,SSID-HEX,BSSID,MODE,CHAN,FREQ,RATE,SIGNAL,BARS,SECURITY,WPA-FLAGS,RSN-FLAGS,DEVICE,ACTIVE,IN-USE,DBUS-PATH', 'device', 'wifi', 'list'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'ALL', 'device', 'wifi', 'list', 'bssid', 'C0:E2:BE:E8:EF:B6'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'COMMON', 'device', 'wifi', 'list', 'bssid', 'C0:E2:BE:E8:EF:B6'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'NAME,SSID,SSID-HEX,BSSID,MODE,CHAN,FREQ,RATE,SIGNAL,BARS,SECURITY,WPA-FLAGS,RSN-FLAGS,DEVICE,ACTIVE,IN-USE,DBUS-PATH', 'device', 'wifi', 'list', 'bssid', 'C0:E2:BE:E8:EF:B6'], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'ALL', 'device', 'show', 'wlan0' ], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'COMMON', 'device', 'show', 'wlan0' ], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['-f', 'GENERAL,CAPABILITIES,WIFI-PROPERTIES,AP,WIRED-PROPERTIES,WIMAX-PROPERTIES,NSP,IP4,DHCP4,IP6,DHCP6,BOND,TEAM,BRIDGE,VLAN,BLUETOOTH,CONNECTIONS', 'device', 'show', 'wlan0' ], replace_stdout = replace_stdout) self.call_nmcli_l(mode + ['dev', 'lldp', 'list', 'ifname', 'eth0'], replace_stdout = replace_stdout) ############################################################################### 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) sys.exit(not r.result.wasSuccessful()) if __name__ == '__main__': main()