diff options
Diffstat (limited to 'clients/tests/test-client.py')
| -rwxr-xr-x | clients/tests/test-client.py | 1372 |
1 files changed, 853 insertions, 519 deletions
diff --git a/clients/tests/test-client.py b/clients/tests/test-client.py index 8bbbdbfc..062fc9f3 100755 --- a/clients/tests/test-client.py +++ b/clients/tests/test-client.py @@ -39,33 +39,33 @@ from __future__ import print_function # (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' +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' +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' +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' +ENV_NM_TEST_REGENERATE = "NM_TEST_REGENERATE" # whether the file location should include the line number. That is useful # only for debugging, to correlate the expected output with the test. # Obviously, since the expected output is commited to git without line numbers, # you'd have to first NM_TEST_REGENERATE the test expected data, with line # numbers enabled. -ENV_NM_TEST_WITH_LINENO = 'NM_TEST_WITH_LINENO' +ENV_NM_TEST_WITH_LINENO = "NM_TEST_WITH_LINENO" -ENV_NM_TEST_ASAN_OPTIONS = 'NM_TEST_ASAN_OPTIONS' -ENV_NM_TEST_LSAN_OPTIONS = 'NM_TEST_LSAN_OPTIONS' -ENV_NM_TEST_UBSAN_OPTIONS = 'NM_TEST_UBSAN_OPTIONS' +ENV_NM_TEST_ASAN_OPTIONS = "NM_TEST_ASAN_OPTIONS" +ENV_NM_TEST_LSAN_OPTIONS = "NM_TEST_LSAN_OPTIONS" +ENV_NM_TEST_UBSAN_OPTIONS = "NM_TEST_UBSAN_OPTIONS" # ############################################################################### @@ -76,7 +76,7 @@ try: import gi from gi.repository import GLib - gi.require_version('NM', '1.0') + gi.require_version("NM", "1.0") from gi.repository import NM except Exception as e: GLib = None @@ -100,8 +100,8 @@ import io ############################################################################### -class PathConfiguration: +class PathConfiguration: @staticmethod def srcdir(): # this is the directory where the test script itself lies. @@ -115,16 +115,19 @@ class PathConfiguration: @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)) + 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__) + p = "clients/tests/test-client.py" + assert (PathConfiguration.top_srcdir() + "/" + p) == os.path.abspath(__file__) return p + ############################################################################### dbus_session_inited = False @@ -134,17 +137,18 @@ _UNSTABLE_OUTPUT = object() ############################################################################### + class Util: _signal_no_lookup = { - 1: "SIGHUP", - 2: "SIGINT", - 3: "SIGQUIT", - 4: "SIGILL", - 5: "SIGTRAP", - 6: "SIGABRT", - 8: "SIGFPE", - 9: "SIGKILL", + 1: "SIGHUP", + 2: "SIGINT", + 3: "SIGQUIT", + 4: "SIGILL", + 5: "SIGTRAP", + 6: "SIGABRT", + 8: "SIGFPE", + 9: "SIGKILL", 11: "SIGSEGV", 12: "SIGSYS", 13: "SIGPIPE", @@ -174,10 +178,10 @@ class Util: return s @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) + 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): @@ -190,14 +194,17 @@ class Util: @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 + _find_unsafe = re.compile( + r"[^\w@%+=:,./-]", re.ASCII if sys.version_info[0] >= 3 else 0 + ).search @staticmethod def quote(s): @@ -210,16 +217,18 @@ class Util: return "'" + s.replace("'", "'\"'\"'") + "'" @staticmethod - def popen_wait(p, timeout = 0): - (res, b_stdout, b_stderr) = Util.popen_wait_read(p, timeout = timeout, read_std_pipes = False) + def popen_wait(p, timeout=0): + (res, b_stdout, b_stderr) = Util.popen_wait_read( + p, timeout=timeout, read_std_pipes=False + ) return res @staticmethod - def popen_wait_read(p, timeout = 0, read_std_pipes = True): + def popen_wait_read(p, timeout=0, read_std_pipes=True): start = NM.utils_get_timestamp_msec() delay = 0.0005 - b_stdout = b'' - b_stderr = b'' + b_stdout = b"" + b_stderr = b"" res = None while True: if read_std_pipes: @@ -230,7 +239,7 @@ class Util: break if timeout == 0: break - assert(timeout > 0) + assert timeout > 0 remaining = timeout - ((NM.utils_get_timestamp_msec() - start) / 1000.0) if remaining <= 0: break @@ -240,14 +249,14 @@ class Util: @staticmethod def buffer_read(buf): - b = b'' + b = b"" while True: try: b1 = buf.read() except io.BlockingIOError: - b1 = b'' + b1 = b"" except IOError: - b1 = b'' + b1 = b"" if not b1: return b b += b1 @@ -280,14 +289,14 @@ class Util: idx = 0 rx = 0 while True: - rx += (l - idx) + rx += l - idx if rx >= r or idx == l - 1: yield jobs[idx] break idx += 1 @staticmethod - def iter_single(itr, min_num = 1, max_num = 1): + def iter_single(itr, min_num=1, max_num=1): itr = list(itr) n = 0 v = None @@ -297,15 +306,19 @@ class Util: break v = c if n < min_num: - raise AssertionError("Expected at least %s elements, but %s found" % (min_num, n)) + 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)) + 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: + with open(filename, "rb") as f: return f.read() except: return None @@ -316,7 +329,7 @@ class Util: return text needs_encode = Util.python_has_version(3) and Util.is_string(text) if needs_encode: - text = text.encode('utf-8') + text = text.encode("utf-8") text = [text] for replace in replace_arr: try: @@ -327,8 +340,8 @@ class Util: if not v_search: continue v_replace = replace[1] - v_search = v_search.encode('utf-8') - v_replace = v_replace.encode('utf-8') + v_search = v_search.encode("utf-8") + v_replace = v_replace.encode("utf-8") text2 = [] for t in text: if isinstance(t, tuple): @@ -337,42 +350,40 @@ class Util: t2 = t.split(v_search) text2.append(t2[0]) for t3 in t2[1:]: - text2.append( (v_replace,) ) + text2.append((v_replace,)) text2.append(t3) text = text2 - bb = b''.join([(t[0] if isinstance(t, tuple) else t) for t in text]) + bb = b"".join([(t[0] if isinstance(t, tuple) else t) for t in text]) if needs_encode: - bb = bb.decode('utf-8') + bb = bb.decode("utf-8") return bb @staticmethod def replace_text_sort_list(lst, replace_arr): - lst = [ (Util.replace_text(elem, replace_arr), elem) for elem in lst ] + lst = [(Util.replace_text(elem, replace_arr), elem) for elem in lst] lst = sorted(lst) - lst = [ tup[1] for tup in lst ] + lst = [tup[1] for tup in lst] return list(lst) @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') + 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']]: + for mode in [[], ["--mode", "tabular"], ["--mode", "multiline"]]: + for fmt in [[], ["--pretty"], ["--terse"]]: + for color in [[], ["--color", "yes"]]: yield mode + fmt + color + ############################################################################### -class Configuration: +class Configuration: def __init__(self): self._values = {} @@ -381,14 +392,18 @@ class Configuration: 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()) + 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") + v = os.path.abspath( + self.get(ENV_NM_TEST_CLIENT_BUILDDIR) + "/clients/cli/nmcli" + ) except: pass if not os.path.exists(v): @@ -402,41 +417,48 @@ class Configuration: # # 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') + 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') + v = os.environ.get(ENV_NM_TEST_REGENERATE, "0") == "1" elif name == ENV_NM_TEST_WITH_LINENO: - v = (os.environ.get(ENV_NM_TEST_WITH_LINENO, '0') == '1') - elif name in [ ENV_NM_TEST_ASAN_OPTIONS, ENV_NM_TEST_LSAN_OPTIONS, ENV_NM_TEST_UBSAN_OPTIONS ]: + v = os.environ.get(ENV_NM_TEST_WITH_LINENO, "0") == "1" + elif name in [ + ENV_NM_TEST_ASAN_OPTIONS, + ENV_NM_TEST_LSAN_OPTIONS, + ENV_NM_TEST_UBSAN_OPTIONS, + ]: v = os.environ.get(name, None) if v is None: if name == ENV_NM_TEST_ASAN_OPTIONS: - v = 'detect_leaks=1' - #v += ' fast_unwind_on_malloc=false' + v = "detect_leaks=1" + # v += ' fast_unwind_on_malloc=false' elif name == ENV_NM_TEST_LSAN_OPTIONS: - v = '' + v = "" elif name == ENV_NM_TEST_UBSAN_OPTIONS: - v = 'print_stacktrace=1:halt_on_error=1' + v = "print_stacktrace=1:halt_on_error=1" else: - assert(False) + assert False else: raise Exception() self._values[name] = v return v + conf = Configuration() ############################################################################### -class NMStubServer: +class NMStubServer: @staticmethod def _conn_get_main_object(conn): try: - return conn.get_object('org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager') + return conn.get_object( + "org.freedesktop.NetworkManager", "/org/freedesktop/NetworkManager" + ) except: return None @@ -444,18 +466,22 @@ class NMStubServer: 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) + 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)) + 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 @@ -463,10 +489,14 @@ class NMStubServer: p.stdin.close() p.kill() Util.popen_wait(p, 1) - raise Exception("after starting stub service the D-Bus name was not claimed in time") + 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._nmiface = dbus.Interface( + nmobj, "org.freedesktop.NetworkManager.LibnmGlibTest" + ) self._p = p def shutdown(self): @@ -481,14 +511,17 @@ class NMStubServer: if Util.popen_wait(p, 1) is None: raise Exception("Stub service did not exit in time") if self._conn_get_main_object(conn) is not None: - raise Exception("Stub service is not still here although it should shut down") + 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) + 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) @@ -504,60 +537,57 @@ class NMStubServer: raise AttributeError(member) return self._MethodProxy(self, member[3:]) - def addConnection(self, connection, do_verify_strict = True): + def addConnection(self, connection, do_verify_strict=True): return self.op_AddConnection(connection, do_verify_strict) def findConnections(self, **kwargs): if kwargs: lst = self.op_FindConnections(**kwargs) else: - lst = self.op_FindConnections( { } ) - return list([ (str(elem[0]), str(elem[1]), str(elem[2])) for elem in lst ]) + lst = self.op_FindConnections({}) + return list([(str(elem[0]), str(elem[1]), str(elem[2])) for elem in lst]) - def findConnectionUuid(self, con_id, required = True): + def findConnectionUuid(self, con_id, required=True): try: - u = Util.iter_single(self.findConnections(con_id = con_id))[1] - assert u, ("Invalid uuid %s" % (u)) + u = Util.iter_single(self.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))) + raise AssertionError( + "Unexpectedly not found connection %s: %s" % (con_id, str(e)) + ) return u - def setProperty(self, path, propname, value, iface_name = None): + def setProperty(self, path, propname, value, iface_name=None): if iface_name is None: - iface_name = '' - self.op_SetProperties([ - (path, [ - (iface_name, [ - (propname, value), - ]), - ]), - ]) + iface_name = "" + self.op_SetProperties( + [(path, [(iface_name, [(propname, value),]),]),] + ) + ############################################################################### -class AsyncProcess(): - def __init__(self, - args, - env, - complete_cb, - max_waittime_msec = 20000): +class AsyncProcess: + def __init__(self, args, env, complete_cb, max_waittime_msec=20000): self._args = list(args) self._env = env self._complete_cb = complete_cb self._max_waittime_msec = max_waittime_msec def start(self): - if not hasattr(self, '_p'): + if not hasattr(self, "_p"): self._p_start_timestamp = NM.utils_get_timestamp_msec() - self._p_stdout_buf = b'' - self._p_stderr_buf = b'' - self._p = subprocess.Popen(self._args, - stdout = subprocess.PIPE, - stderr = subprocess.PIPE, - env = self._env) + self._p_stdout_buf = b"" + self._p_stderr_buf = b"" + self._p = subprocess.Popen( + self._args, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=self._env, + ) Util.buffer_set_nonblock(self._p.stdout) Util.buffer_set_nonblock(self._p.stderr) @@ -568,9 +598,11 @@ class AsyncProcess(): # frequently. # Worst case, we will think that the process did not time out, # when in fact it was running longer than max-waittime. - return self._max_waittime_msec - (NM.utils_get_timestamp_msec() - self._p_start_timestamp) + return self._max_waittime_msec - ( + NM.utils_get_timestamp_msec() - self._p_start_timestamp + ) - def poll(self, timeout = 0): + def poll(self, timeout=0): self.start() (return_code, b_stdout, b_stderr) = Util.popen_wait_read(self._p, timeout) @@ -578,9 +610,10 @@ class AsyncProcess(): self._p_stdout_buf += b_stdout self._p_stderr_buf += b_stderr - if return_code is None \ - and self._timeout_remaining_time() <= 0: - raise Exception("process is still running after timeout: %s" % (' '.join(self._args))) + if return_code is None and self._timeout_remaining_time() <= 0: + raise Exception( + "process is still running after timeout: %s" % (" ".join(self._args)) + ) return return_code def wait_and_complete(self): @@ -589,7 +622,9 @@ class AsyncProcess(): p = self._p self._p = None - (return_code, b_stdout, b_stderr) = Util.popen_wait_read(p, max(0, self._timeout_remaining_time()) / 1000) + (return_code, b_stdout, b_stderr) = Util.popen_wait_read( + p, max(0, self._timeout_remaining_time()) / 1000 + ) (stdout, stderr) = (p.stdout.read(), p.stderr.read()) p.stdout.close() p.stderr.close() @@ -602,35 +637,40 @@ class AsyncProcess(): if return_code is None: print(stdout) print(stderr) - raise Exception("process did not complete in time: %s" % (' '.join(self._args))) + raise Exception( + "process did not complete in time: %s" % (" ".join(self._args)) + ) self._complete_cb(self, return_code, stdout, stderr) + ############################################################################### + class NmTestBase(unittest.TestCase): pass + MAX_JOBS = 15 -class TestNmcli(NmTestBase): +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') + size_prefix = "size: ".encode("utf8") while True: - if not content_expect[base_idx:base_idx + 10].startswith(size_prefix): + 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') + eol = ord("\n") else: - eol = '\n' + eol = "\n" while content_expect[i] != eol: i += 1 i = i + 1 + int(content_expect[j:i]) @@ -643,52 +683,57 @@ class TestNmcli(NmTestBase): 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, - replace_cmd = None, - sort_lines_stdout = False, - extra_env = None, - sync_barrier = False): + 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, + replace_cmd=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, - replace_cmd, - 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, - replace_cmd = None, - sort_lines_stdout = False, - extra_env = None, - sync_barrier = None): + 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, + replace_cmd, + 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, + replace_cmd=None, + sort_lines_stdout=False, + extra_env=None, + sync_barrier=None, + ): frame = sys._getframe(1) @@ -696,43 +741,47 @@ class TestNmcli(NmTestBase): assert lang is None else: if lang is None: - lang = 'C' + lang = "C" langs = [lang] if sync_barrier is None: - sync_barrier = (len(langs) == 1) + 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, - replace_cmd, - 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, - replace_cmd, - sort_lines_stdout, - extra_env, - sync_barrier, - frame): + self._call_nmcli( + args, + lang, + check_on_disk, + fatal_warnings, + expected_returncode, + expected_stdout, + expected_stderr, + replace_stdout, + replace_stderr, + replace_cmd, + 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, + replace_cmd, + sort_lines_stdout, + extra_env, + sync_barrier, + frame, + ): if sync_barrier: self.async_wait() @@ -741,51 +790,63 @@ class TestNmcli(NmTestBase): 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) + 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())) + self.assertTrue( + os.path.abspath(frame.f_code.co_filename).endswith( + "/" + PathConfiguration.canonical_script_filename() + ) + ) if conf.get(ENV_NM_TEST_WITH_LINENO): - calling_location = '%s:%d:%s()/%d' % (PathConfiguration.canonical_script_filename(), frame.f_lineno, frame.f_code.co_name, calling_num) + calling_location = "%s:%d:%s()/%d" % ( + PathConfiguration.canonical_script_filename(), + frame.f_lineno, + frame.f_code.co_name, + calling_num, + ) else: - calling_location = '%s:%s()/%d' % (PathConfiguration.canonical_script_filename(), frame.f_code.co_name, calling_num) - - if lang is None or lang == 'C': - lang = 'C' - language = '' - elif lang == 'de': - lang = 'de_DE.utf8' - language = 'de' - elif lang == 'pl': - lang = 'pl_PL.UTF-8' - language = 'pl' + calling_location = "%s:%s()/%d" % ( + PathConfiguration.canonical_script_filename(), + frame.f_code.co_name, + calling_num, + ) + + if lang is None or lang == "C": + lang = "C" + language = "" + elif lang == "de": + lang = "de_DE.utf8" + language = "de" + elif lang == "pl": + lang = "pl_PL.UTF-8" + language = "pl" else: - self.fail('invalid language %s' % (lang)) + 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']: + 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'] = conf.get(ENV_NM_TEST_ASAN_OPTIONS) - env['LSAN_OPTIONS'] = conf.get(ENV_NM_TEST_LSAN_OPTIONS) - env['LBSAN_OPTIONS'] = conf.get(ENV_NM_TEST_UBSAN_OPTIONS) - env['XDG_CONFIG_HOME'] = PathConfiguration.srcdir() - env['NM_TEST_CALLING_NUM'] = str(calling_num) + env["LANG"] = lang + env["LANGUAGE"] = language + env["LIBNM_USE_SESSION_BUS"] = "1" + env["LIBNM_USE_NO_UDEV"] = "1" + env["TERM"] = "linux" + env["ASAN_OPTIONS"] = conf.get(ENV_NM_TEST_ASAN_OPTIONS) + env["LSAN_OPTIONS"] = conf.get(ENV_NM_TEST_LSAN_OPTIONS) + env["LBSAN_OPTIONS"] = conf.get(ENV_NM_TEST_UBSAN_OPTIONS) + env["XDG_CONFIG_HOME"] = PathConfiguration.srcdir() + env["NM_TEST_CALLING_NUM"] = str(calling_num) if fatal_warnings is _DEFAULT_ARG or fatal_warnings: - env['G_DEBUG'] = 'fatal-warnings' + env["G_DEBUG"] = "fatal-warnings" args = [conf.get(ENV_NM_TEST_CLIENT_NMCLI_PATH)] + list(args) @@ -797,9 +858,17 @@ class TestNmcli(NmTestBase): replace_cmd = list(replace_cmd) 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)) + 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: @@ -810,26 +879,24 @@ class TestNmcli(NmTestBase): results_idx = len(self._results) self._results.append(None) - def complete_cb(async_job, - returncode, - stdout, - stderr): + def complete_cb(async_job, returncode, stdout, stderr): if expected_stdout is _UNSTABLE_OUTPUT: - stdout = '<UNSTABLE OUTPUT>'.encode('utf-8') + stdout = "<UNSTABLE OUTPUT>".encode("utf-8") else: stdout = Util.replace_text(stdout, replace_stdout) if expected_stderr is _UNSTABLE_OUTPUT: - stderr = '<UNSTABLE OUTPUT>'.encode('utf-8') + stderr = "<UNSTABLE OUTPUT>".encode("utf-8") else: stderr = Util.replace_text(stderr, replace_stderr) if sort_lines_stdout: - stdout = b'\n'.join(sorted(stdout.split(b'\n'))) + stdout = b"\n".join(sorted(stdout.split(b"\n"))) - ignore_l10n_diff = ( lang != 'C' - and not conf.get(ENV_NM_TEST_CLIENT_CHECK_L10N)) + 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: @@ -851,47 +918,53 @@ class TestNmcli(NmTestBase): self.assertNotEqual(returncode, -5) elif fatal_warnings: if expected_returncode is None: - self.assertEqual(returncode, -5) + self.assertEqual(returncode, -5) if check_on_disk: - cmd = '$NMCLI %s' % (' '.join([Util.quote(a) for a in args[1:]])) + cmd = "$NMCLI %s" % (" ".join([Util.quote(a) for a in args[1:]])) cmd = Util.replace_text(cmd, replace_cmd) if returncode < 0: - returncode_str = '%d (SIGNAL %s)' % (returncode, Util.signal_no_to_str(-returncode)) + returncode_str = "%d (SIGNAL %s)" % ( + returncode, + Util.signal_no_to_str(-returncode), + ) else: - returncode_str = '%d' % (returncode) - - content = ('location: %s\n' % (calling_location)).encode('utf8') + \ - ('cmd: %s\n' % (cmd)).encode('utf8') + \ - ('lang: %s\n' % (lang)).encode('utf8') + \ - ('returncode: %s\n' % (returncode_str)).encode('utf8') + returncode_str = "%d" % (returncode) + + content = ( + ("location: %s\n" % (calling_location)).encode("utf8") + + ("cmd: %s\n" % (cmd)).encode("utf8") + + ("lang: %s\n" % (lang)).encode("utf8") + + ("returncode: %s\n" % (returncode_str)).encode("utf8") + ) if len(stdout) > 0: - content += ('stdout: %d bytes\n>>>\n' % (len(stdout))).encode('utf8') + \ - stdout + \ - '\n<<<\n'.encode('utf8') + 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 + 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[results_idx] = { - 'test_name' : test_name, - 'ignore_l10n_diff' : ignore_l10n_diff, - 'content' : content, + "test_name": test_name, + "ignore_l10n_diff": ignore_l10n_diff, + "content": content, } - async_job = AsyncProcess(args = args, - env = env, - complete_cb = complete_cb) + async_job = AsyncProcess(args=args, env=env, complete_cb=complete_cb) self._async_jobs.append(async_job) - self.async_start(wait_all = sync_barrier) + self.async_start(wait_all=sync_barrier) - def async_start(self, wait_all = False): + def async_start(self, wait_all=False): while True: @@ -922,13 +995,13 @@ class TestNmcli(NmTestBase): # completes. Note that poll() itself will raise an exception if a # jobs times out. for async_job in Util.random_job(jobs_running): - if async_job.poll(timeout = 0.03) is not None: + if async_job.poll(timeout=0.03) is not None: self._async_jobs.remove(async_job) async_job.wait_and_complete() break def async_wait(self): - return self.async_start(wait_all = True) + return self.async_start(wait_all=True) def _nm_test_pre(self): self._calling_num = {} @@ -955,7 +1028,12 @@ class TestNmcli(NmTestBase): test_name = self._testMethodName - filename = os.path.abspath(PathConfiguration.srcdir() + '/test-client.check-on-disk/' + test_name + '.expected') + filename = os.path.abspath( + PathConfiguration.srcdir() + + "/test-client.check-on-disk/" + + test_name + + ".expected" + ) regenerate = conf.get(ENV_NM_TEST_REGENERATE) @@ -963,41 +1041,71 @@ class TestNmcli(NmTestBase): 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)) + 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']: + if results_expect[i] == n["content"]: continue if regenerate: continue - if n['ignore_l10n_diff']: - skip_test_for_l10n_diff.append(n['test_name']) + 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( + "\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())) + 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)) + 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))) + 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'])) + 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())) + 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)) + self.fail( + "Unexpected output of command, expected %s. Rerun test with NM_TEST_REGENERATE=1 to regenerate files" + % (filename) + ) if regenerate: - content_new = b''.join([r['content'] for r in results]) + content_new = b"".join([r["content"] for r in results]) if content_new != content_expect: try: - with open(filename, 'wb') as content_file: + 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)) @@ -1006,102 +1114,98 @@ class TestNmcli(NmTestBase): # 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))) + 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?") + 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') + 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("WifiDevice", ident="wlan1/x", iface="wlan1") - self.srv.op_AddObj('WifiAp', - device = 'wlan0', - rsnf = 0x0) + self.srv.op_AddObj("WifiAp", device="wlan0", rsnf=0x0) - self.srv.op_AddObj('WifiAp', - device = 'wlan0') + self.srv.op_AddObj("WifiAp", device="wlan0") - NM_AP_FLAGS = getattr(NM, '80211ApSecurityFlags') + 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="wlan0", wpaf=0x0, rsnf=rsnf) - self.srv.op_AddObj('WifiAp', - device = 'wlan1') + self.srv.op_AddObj("WifiAp", device="wlan1") - self.srv.addConnection( { - 'connection': { - 'type': '802-3-ethernet', - 'id': 'con-1', - }, - }) + 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( + ["-f", "AP", "-mode", "multiline", "-p", "d", "show", "wlan0"] + ) - self.call_nmcli_l(['c', 's']) + self.call_nmcli_l(["c", "s"]) - self.call_nmcli_l(['bogus', 's']) + self.call_nmcli_l(["bogus", "s"]) for mode in Util.iter_nmcli_output_modes(): - self.call_nmcli_l(mode + ['general', 'permissions']) + 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(["d"]) - self.call_nmcli_l(['-f', 'all', '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", "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(["-f", "ALL", "d", "wifi"]) - self.call_nmcli_l(['c']) + self.call_nmcli_l(["c"]) - self.call_nmcli_l(['c', 's', 'con-1']) + self.call_nmcli_l(["c", "s", "con-1"]) @nm_test def test_003(self): @@ -1109,36 +1213,76 @@ class TestNmcli(NmTestBase): replace_uuids = [] - replace_uuids.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_uuids) - - self.call_nmcli_l(['c', 's'], - replace_stdout = replace_uuids) - - replace_uuids.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_uuids) - - replace_uuids.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_uuids) - - self.call_nmcli_l(['c', 's'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(['-f', 'ALL', 'c', 's'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(['--complete-args', '-f', 'ALL', 'c', 's', ''], - replace_stdout = replace_uuids, - sort_lines_stdout = True) - - self.call_nmcli_l(['con', 's', 'con-gsm1'], - replace_stdout = replace_uuids) + replace_uuids.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_uuids, + ) + + self.call_nmcli_l(["c", "s"], replace_stdout=replace_uuids) + + replace_uuids.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_uuids, + ) + + replace_uuids.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_uuids, + ) + + self.call_nmcli_l(["c", "s"], replace_stdout=replace_uuids) + + self.call_nmcli_l(["-f", "ALL", "c", "s"], replace_stdout=replace_uuids) + + self.call_nmcli_l( + ["--complete-args", "-f", "ALL", "c", "s", ""], + replace_stdout=replace_uuids, + sort_lines_stdout=True, + ) + + self.call_nmcli_l(["con", "s", "con-gsm1"], replace_stdout=replace_uuids) # activate the same profile on multiple devices. Our stub-implmentation # is fine with that... although NetworkManager service would reject @@ -1150,72 +1294,91 @@ class TestNmcli(NmTestBase): # 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]) + for dev in ["eth0", "eth1"]: + self.call_nmcli(["con", "up", "ethernet", "ifname", dev]) - self.call_nmcli_l(['con'], - replace_stdout = replace_uuids) + self.call_nmcli_l(["con"], replace_stdout=replace_uuids) - self.call_nmcli_l(['-f', 'ALL', 'con'], - replace_stdout = replace_uuids) + self.call_nmcli_l(["-f", "ALL", "con"], replace_stdout=replace_uuids) - self.call_nmcli_l(['-f', 'ALL', 'con', 's', '-a'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + ["-f", "ALL", "con", "s", "-a"], replace_stdout=replace_uuids + ) - self.call_nmcli_l(['-f', 'ACTIVE-PATH,DEVICE,UUID', 'con', 's', '-act'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + ["-f", "ACTIVE-PATH,DEVICE,UUID", "con", "s", "-act"], + replace_stdout=replace_uuids, + ) - self.call_nmcli_l(['-f', 'UUID,NAME', 'con', 's', '--active'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + ["-f", "UUID,NAME", "con", "s", "--active"], + replace_stdout=replace_uuids, + ) - self.call_nmcli_l(['-f', 'ALL', 'con', 's', 'ethernet'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + ["-f", "ALL", "con", "s", "ethernet"], replace_stdout=replace_uuids + ) - self.call_nmcli_l(['-f', 'GENERAL.STATE', 'con', 's', 'ethernet'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + ["-f", "GENERAL.STATE", "con", "s", "ethernet"], + replace_stdout=replace_uuids, + ) - self.call_nmcli_l(['con', 's', 'ethernet'], - replace_stdout = replace_uuids) + self.call_nmcli_l(["con", "s", "ethernet"], replace_stdout=replace_uuids) - self.call_nmcli_l(['-f', 'ALL', 'dev', 'status'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + ["-f", "ALL", "dev", "status"], replace_stdout=replace_uuids + ) # test invalid call ('s' abbrevates 'status' and not 'show' - self.call_nmcli_l(['-f', 'ALL', 'dev', 's', 'eth0'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + ["-f", "ALL", "dev", "s", "eth0"], replace_stdout=replace_uuids + ) - self.call_nmcli_l(['-f', 'ALL', 'dev', 'show', 'eth0'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + ["-f", "ALL", "dev", "show", "eth0"], replace_stdout=replace_uuids + ) - self.call_nmcli_l(['-f', 'ALL', '-t', 'dev', 'show', 'eth0'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + ["-f", "ALL", "-t", "dev", "show", "eth0"], replace_stdout=replace_uuids + ) self.async_wait() - self.srv.setProperty('/org/freedesktop/NetworkManager/ActiveConnection/1', - 'State', - dbus.UInt32(NM.ActiveConnectionState.DEACTIVATING)) + 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') + 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_uuids) + self.call_nmcli_l( + mode + ["-f", "ALL", "con"], replace_stdout=replace_uuids + ) - self.call_nmcli_l(mode + ['-f', 'UUID,TYPE', 'con'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + mode + ["-f", "UUID,TYPE", "con"], replace_stdout=replace_uuids + ) - self.call_nmcli_l(mode + ['con', 's', 'ethernet'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + mode + ["con", "s", "ethernet"], replace_stdout=replace_uuids + ) - self.call_nmcli_l(mode + ['c', 's', '/org/freedesktop/NetworkManager/ActiveConnection/1'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + mode + + ["c", "s", "/org/freedesktop/NetworkManager/ActiveConnection/1"], + replace_stdout=replace_uuids, + ) - self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'show', 'eth0'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + mode + ["-f", "all", "dev", "show", "eth0"], + replace_stdout=replace_uuids, + ) @nm_test def test_004(self): @@ -1223,131 +1386,275 @@ class TestNmcli(NmTestBase): replace_uuids = [] - replace_uuids.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_uuids) - - 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_uuids) + replace_uuids.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_uuids, + ) + + 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_uuids) self.async_wait() - replace_uuids.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_uuids) - - self.call_nmcli_l(['con', 's'], - replace_stdout = replace_uuids) - self.call_nmcli_l(['con', 's', 'con-vpn-1'], - replace_stdout = replace_uuids) - - self.call_nmcli(['con', 'up', 'con-xx1']) - self.call_nmcli_l(['con', 's'], - replace_stdout = replace_uuids) - - self.call_nmcli(['con', 'up', 'con-vpn-1']) - self.call_nmcli_l(['con', 's'], - replace_stdout = replace_uuids) - self.call_nmcli_l(['con', 's', 'con-vpn-1'], - replace_stdout = replace_uuids) + replace_uuids.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_uuids, + ) + + self.call_nmcli_l(["con", "s"], replace_stdout=replace_uuids) + self.call_nmcli_l(["con", "s", "con-vpn-1"], replace_stdout=replace_uuids) + + self.call_nmcli(["con", "up", "con-xx1"]) + self.call_nmcli_l(["con", "s"], replace_stdout=replace_uuids) + + self.call_nmcli(["con", "up", "con-vpn-1"]) + self.call_nmcli_l(["con", "s"], replace_stdout=replace_uuids) + self.call_nmcli_l(["con", "s", "con-vpn-1"], replace_stdout=replace_uuids) self.async_wait() - self.srv.setProperty('/org/freedesktop/NetworkManager/ActiveConnection/2', - 'VpnState', - dbus.UInt32(NM.VpnConnectionState.ACTIVATED)) + self.srv.setProperty( + "/org/freedesktop/NetworkManager/ActiveConnection/2", + "VpnState", + dbus.UInt32(NM.VpnConnectionState.ACTIVATED), + ) - uuids = Util.replace_text_sort_list([ c[1] for c in self.srv.findConnections() ], replace_uuids) + uuids = Util.replace_text_sort_list( + [c[1] for c in self.srv.findConnections()], replace_uuids + ) for mode in Util.iter_nmcli_output_modes(): - self.call_nmcli_l(mode + ['con', 's', 'con-vpn-1'], - replace_stdout = replace_uuids) - self.call_nmcli_l(mode + ['con', 's', 'con-vpn-1'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + mode + ["con", "s", "con-vpn-1"], replace_stdout=replace_uuids + ) + self.call_nmcli_l( + mode + ["con", "s", "con-vpn-1"], replace_stdout=replace_uuids + ) - self.call_nmcli_l(mode + ['-f', 'ALL', 'con', 's', 'con-vpn-1'], - replace_stdout = replace_uuids) + self.call_nmcli_l( + mode + ["-f", "ALL", "con", "s", "con-vpn-1"], + replace_stdout=replace_uuids, + ) # 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_uuids) - - self.call_nmcli_l(mode + ['-f', 'GENERAL', 'con', 's', 'con-vpn-1'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['dev', 's'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'status'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['dev', 'show'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'show'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['dev', 'show', 'wlan0'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['-f', 'all', 'dev', 'show', 'wlan0'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['-f', 'GENERAL,GENERAL.HWADDR,WIFI-PROPERTIES', 'dev', 'show', 'wlan0'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['-f', 'GENERAL,GENERAL.HWADDR,WIFI-PROPERTIES', 'dev', 'show', 'wlan0'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['-f', 'DEVICE,TYPE,DBUS-PATH', 'dev'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['-f', 'ALL', 'device', 'wifi', 'list' ], - replace_stdout = replace_uuids) - self.call_nmcli_l(mode + ['-f', 'COMMON', 'device', 'wifi', 'list' ], - replace_stdout = replace_uuids) - 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_uuids) - self.call_nmcli_l(mode + ['-f', 'ALL', 'device', 'wifi', 'list', 'bssid', 'C0:E2:BE:E8:EF:B6'], - replace_stdout = replace_uuids) - self.call_nmcli_l(mode + ['-f', 'COMMON', 'device', 'wifi', 'list', 'bssid', 'C0:E2:BE:E8:EF:B6'], - replace_stdout = replace_uuids) - 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_uuids) - self.call_nmcli_l(mode + ['-f', 'ALL', 'device', 'show', 'wlan0' ], - replace_stdout = replace_uuids) - self.call_nmcli_l(mode + ['-f', 'COMMON', 'device', 'show', 'wlan0' ], - replace_stdout = replace_uuids) - 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_uuids) - - self.call_nmcli_l(mode + ['dev', 'lldp', 'list', 'ifname', 'eth0'], - replace_stdout = replace_uuids) - - self.call_nmcli_l(mode + ['-f', 'connection.id,connection.uuid,connection.type,connection.interface-name,802-3-ethernet.mac-address,vpn.user-name', 'connection', 'show' ] + uuids, - replace_stdout = replace_uuids, - replace_cmd = replace_uuids) + self.call_nmcli_l( + mode + ["-f", "VPN", "con", "s", "con-vpn-1"], + replace_stdout=replace_uuids, + ) + + self.call_nmcli_l( + mode + ["-f", "GENERAL", "con", "s", "con-vpn-1"], + replace_stdout=replace_uuids, + ) + + self.call_nmcli_l(mode + ["dev", "s"], replace_stdout=replace_uuids) + + self.call_nmcli_l( + mode + ["-f", "all", "dev", "status"], replace_stdout=replace_uuids + ) + + self.call_nmcli_l(mode + ["dev", "show"], replace_stdout=replace_uuids) + + self.call_nmcli_l( + mode + ["-f", "all", "dev", "show"], replace_stdout=replace_uuids + ) + + self.call_nmcli_l( + mode + ["dev", "show", "wlan0"], replace_stdout=replace_uuids + ) + + self.call_nmcli_l( + mode + ["-f", "all", "dev", "show", "wlan0"], + replace_stdout=replace_uuids, + ) + + self.call_nmcli_l( + mode + + [ + "-f", + "GENERAL,GENERAL.HWADDR,WIFI-PROPERTIES", + "dev", + "show", + "wlan0", + ], + replace_stdout=replace_uuids, + ) + + self.call_nmcli_l( + mode + + [ + "-f", + "GENERAL,GENERAL.HWADDR,WIFI-PROPERTIES", + "dev", + "show", + "wlan0", + ], + replace_stdout=replace_uuids, + ) + + self.call_nmcli_l( + mode + ["-f", "DEVICE,TYPE,DBUS-PATH", "dev"], + replace_stdout=replace_uuids, + ) + + self.call_nmcli_l( + mode + ["-f", "ALL", "device", "wifi", "list"], + replace_stdout=replace_uuids, + ) + self.call_nmcli_l( + mode + ["-f", "COMMON", "device", "wifi", "list"], + replace_stdout=replace_uuids, + ) + 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_uuids, + ) + self.call_nmcli_l( + mode + + ["-f", "ALL", "device", "wifi", "list", "bssid", "C0:E2:BE:E8:EF:B6"], + replace_stdout=replace_uuids, + ) + self.call_nmcli_l( + mode + + [ + "-f", + "COMMON", + "device", + "wifi", + "list", + "bssid", + "C0:E2:BE:E8:EF:B6", + ], + replace_stdout=replace_uuids, + ) + 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_uuids, + ) + self.call_nmcli_l( + mode + ["-f", "ALL", "device", "show", "wlan0"], + replace_stdout=replace_uuids, + ) + self.call_nmcli_l( + mode + ["-f", "COMMON", "device", "show", "wlan0"], + replace_stdout=replace_uuids, + ) + 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_uuids, + ) + + self.call_nmcli_l( + mode + ["dev", "lldp", "list", "ifname", "eth0"], + replace_stdout=replace_uuids, + ) + + self.call_nmcli_l( + mode + + [ + "-f", + "connection.id,connection.uuid,connection.type,connection.interface-name,802-3-ethernet.mac-address,vpn.user-name", + "connection", + "show", + ] + + uuids, + replace_stdout=replace_uuids, + replace_cmd=replace_uuids, + ) ############################################################################### + def main(): global dbus_session_inited - if len(sys.argv) >= 2 and sys.argv[1] == '--started-with-dbus-session': + if len(sys.argv) >= 2 and sys.argv[1] == "--started-with-dbus-session": dbus_session_inited = True del sys.argv[1] @@ -1356,37 +1663,64 @@ def main(): # a new dbus-session. try: try: - os.execlp('dbus-run-session', 'dbus-run-session', '--', sys.executable, __file__, '--started-with-dbus-session', *sys.argv[1:]) + 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') + raise Exception("unknown error during exec") except Exception as e: - assert False, ("Failure to re-exec dbus-run-session: %s" % (str(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: + 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' + \ - '') + 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)) + m = "unknown error" + assert False, "Failure to re-exec to start script with dbus-launch: %s" % ( + m + ) - r = unittest.main(exit = False) + r = unittest.main(exit=False) sys.exit(not r.result.wasSuccessful()) -if __name__ == '__main__': + +if __name__ == "__main__": main() |