New upstream version 1.32.12 upstream/1.32.12

1 files changed, 1742 insertions, 0 deletions

diff --git a/src/tests/client/test-client.py b/src/tests/client/test-client.py
new file mode 100755
index 00000000..97f1abd9
--- /dev/null
+++ b/src/tests/client/test-client.py
@@ -0,0 +1,1742 @@
+#!/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 (src/tests/client/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 ./src/nmcli/nmcli --version
+#    # also ensure that `locale -a` reports the Polish locale.
+#    #
+#    # How to get that right depends on your distro. First check that
+#    # `LANG=pl_PL.UTF-8 /usr/bin/nmcli --version` gives you Polish output
+#    # and that it works in principle.
+#    # Then install NetworkManager in the configured prefix (make install)
+#    # and verify that ./src/nmcli/nmcli also gives you Polish.
+#  $ NM_TEST_REGENERATE=1 make check-local-tests-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"
+
+# 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_ASAN_OPTIONS = "NM_TEST_ASAN_OPTIONS"
+ENV_NM_TEST_LSAN_OPTIONS = "NM_TEST_LSAN_OPTIONS"
+ENV_NM_TEST_UBSAN_OPTIONS = "NM_TEST_UBSAN_OPTIONS"
+
+#
+###############################################################################
+
+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 fcntl
+import dbus
+import time
+import random
+import dbus.service
+import dbus.mainloop.glib
+import io
+
+###############################################################################
+
+
+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 = "src/tests/client/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:
+
+    _signal_no_lookup = {
+        1: "SIGHUP",
+        2: "SIGINT",
+        3: "SIGQUIT",
+        4: "SIGILL",
+        5: "SIGTRAP",
+        6: "SIGABRT",
+        8: "SIGFPE",
+        9: "SIGKILL",
+        11: "SIGSEGV",
+        12: "SIGSYS",
+        13: "SIGPIPE",
+        14: "SIGALRM",
+        15: "SIGTERM",
+        16: "SIGURG",
+        17: "SIGSTOP",
+        18: "SIGTSTP",
+        19: "SIGCONT",
+        20: "SIGCHLD",
+        21: "SIGTTIN",
+        22: "SIGTTOU",
+        23: "SIGPOLL",
+        24: "SIGXCPU",
+        25: "SIGXFSZ",
+        26: "SIGVTALRM",
+        27: "SIGPROF",
+        30: "SIGUSR1",
+        31: "SIGUSR2",
+    }
+
+    @classmethod
+    def signal_no_to_str(cls, signal):
+        s = cls._signal_no_lookup.get(signal, None)
+        if s is None:
+            return "<unknown %d>" % (signal)
+        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
+        )
+
+    @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=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):
+        start = NM.utils_get_timestamp_msec()
+        delay = 0.0005
+        b_stdout = b""
+        b_stderr = b""
+        res = None
+        while True:
+            if read_std_pipes:
+                b_stdout += Util.buffer_read(p.stdout)
+                b_stderr += Util.buffer_read(p.stderr)
+            if p.poll() is not None:
+                res = p.returncode
+                break
+            if timeout == 0:
+                break
+            assert timeout > 0
+            remaining = timeout - ((NM.utils_get_timestamp_msec() - start) / 1000.0)
+            if remaining <= 0:
+                break
+            delay = min(delay * 2, remaining, 0.05)
+            time.sleep(delay)
+        return (res, b_stdout, b_stderr)
+
+    @staticmethod
+    def buffer_read(buf):
+        b = b""
+        while True:
+            try:
+                b1 = buf.read()
+            except io.BlockingIOError:
+                b1 = b""
+            except IOError:
+                b1 = b""
+            if not b1:
+                return b
+            b += b1
+
+    @staticmethod
+    def buffer_set_nonblock(buf):
+        fd = buf.fileno()
+        fl = fcntl.fcntl(fd, fcntl.F_GETFL)
+        fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
+
+    @staticmethod
+    def random_job(jobs):
+        jobs = list(jobs)
+        l = len(jobs)
+        t = l * (l + 1) / 2
+        while True:
+            # we return a random jobs from the list, but the indexes at the front of
+            # the list are more likely. The idea is, that those jobs were started first,
+            # and are expected to complete first. As we poll, we want to check more frequently
+            # on the elements at the beginning of the list...
+            #
+            # Let's assign probabilities with an arithmetic series.
+            # That is, if there are 16 jobs, then the first gets weighted
+            # with 16, the second with 15, then 14, and so on, until the
+            # last has weight 1. That means, the first element is 16 times
+            # more probable than the last.
+            # Element at idx (starting with 0) is picked with probability
+            #    1 / (l*(l+1)/2) * (l - idx)
+            r = random.random() * t
+            idx = 0
+            rx = 0
+            while True:
+                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):
+        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
+        needs_encode = Util.python_has_version(3) and Util.is_string(text)
+        if needs_encode:
+            text = text.encode("utf-8")
+        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
+        bb = b"".join([(t[0] if isinstance(t, tuple) else t) for t in text])
+        if needs_encode:
+            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 = sorted(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"
+        )
+
+    @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) + "/src/nmcli/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"
+        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(name, None)
+            if v is None:
+                if name == ENV_NM_TEST_ASAN_OPTIONS:
+                    v = "detect_leaks=1"
+                    # v += ' fast_unwind_on_malloc=false'
+                elif name == ENV_NM_TEST_LSAN_OPTIONS:
+                    v = ""
+                elif name == ENV_NM_TEST_UBSAN_OPTIONS:
+                    v = "print_stacktrace=1:halt_on_error=1"
+                else:
+                    assert False
+        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, 1)
+                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):
+        conn = self._conn
+        p = self._p
+        self._nmobj = None
+        self._nmiface = None
+        self._conn = None
+        self._p = None
+        p.stdin.close()
+        p.kill()
+        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"
+            )
+
+    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.
+                args = list(args)
+                args.append(kwargs)
+            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 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])
+
+    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)
+        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, 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"):
+            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,
+            )
+            Util.buffer_set_nonblock(self._p.stdout)
+            Util.buffer_set_nonblock(self._p.stderr)
+
+    def _timeout_remaining_time(self):
+        # note that we call this during poll() and wait_and_complete().
+        # we don't know the exact time when the process terminated,
+        # so this is only approximately correct, if we call poll/wait
+        # 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
+        )
+
+    def poll(self, timeout=0):
+        self.start()
+
+        (return_code, b_stdout, b_stderr) = Util.popen_wait_read(self._p, timeout)
+
+        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))
+            )
+        return return_code
+
+    def wait_and_complete(self):
+        self.start()
+
+        p = self._p
+        self._p = None
+
+        (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()
+
+        stdout = self._p_stdout_buf + b_stdout + stdout
+        stderr = self._p_stderr_buf + b_stderr + stderr
+        del self._p_stdout_buf
+        del self._p_stderr_buf
+
+        if return_code is None:
+            print(stdout)
+            print(stderr)
+            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):
+    @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,
+        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,
+    ):
+
+        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,
+                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()
+
+        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()
+            )
+        )
+
+        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,
+            )
+        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"
+        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"] = 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"
+
+        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 replace_cmd is not None:
+            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
+                )
+            )
+        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
+
+        results_idx = len(self._results)
+        self._results.append(None)
+
+        def complete_cb(async_job, returncode, stdout, stderr):
+
+            if expected_stdout is _UNSTABLE_OUTPUT:
+                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")
+            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:]]))
+                cmd = Util.replace_text(cmd, replace_cmd)
+
+                if returncode < 0:
+                    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")
+                )
+                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[results_idx] = {
+                    "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)
+
+        self.async_start(wait_all=sync_barrier)
+
+    def async_start(self, wait_all=False):
+
+        while True:
+
+            while True:
+                for async_job in list(self._async_jobs[0:MAX_JOBS]):
+                    async_job.start()
+                # start up to MAX_JOBS jobs, but poll() and complete those
+                # that are already exited. Retry, until there are no more
+                # jobs to start, or until MAX_JOBS are running.
+                jobs_running = []
+                for async_job in list(self._async_jobs[0:MAX_JOBS]):
+                    if async_job.poll() is not None:
+                        self._async_jobs.remove(async_job)
+                        async_job.wait_and_complete()
+                        continue
+                    jobs_running.append(async_job)
+                if len(jobs_running) >= len(self._async_jobs):
+                    break
+                if len(jobs_running) >= MAX_JOBS:
+                    break
+
+            if not jobs_running:
+                return
+            if not wait_all:
+                return
+
+            # in a loop, indefinitely poll the running jobs until we find one that
+            # 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:
+                    self._async_jobs.remove(async_job)
+                    async_job.wait_and_complete()
+                    break
+
+    def async_wait(self):
+        return self.async_start(wait_all=True)
+
+    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 = b"".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):
+        con_gsm_list = [
+            ("con-gsm1", "xyz.con-gsm1"),
+            ("con-gsm2", ""),
+            ("con-gsm3", " "),
+        ]
+
+        self.init_001()
+
+        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)
+
+        for con_name, apn in con_gsm_list:
+
+            replace_uuids.append(
+                (
+                    Util.memoize_nullary(lambda: self.srv.findConnectionUuid(con_name)),
+                    "UUID-" + con_name + "-REPLACED-REPLACED-REPL",
+                )
+            )
+
+            self.call_nmcli(
+                [
+                    "connection",
+                    "add",
+                    "type",
+                    "gsm",
+                    "autoconnect",
+                    "no",
+                    "con-name",
+                    con_name,
+                    "ifname",
+                    "*",
+                    "apn",
+                    apn,
+                    "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,
+        )
+
+        for con_name, apn in con_gsm_list:
+            self.call_nmcli_l(["con", "s", con_name], replace_stdout=replace_uuids)
+            self.call_nmcli_l(
+                ["-g", "all", "con", "s", con_name], replace_stdout=replace_uuids
+            )
+
+        # 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_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", "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", "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(["con", "s", "ethernet"], 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", "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),
+        )
+
+        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_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
+                    + ["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,
+                )
+
+    @nm_test
+    def test_004(self):
+        self.init_001()
+
+        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)
+
+        self.async_wait()
+
+        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
+        )
+
+        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 + ["-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,
+            )
+
+
+###############################################################################
+
+
+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()