colin/NetworkManager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test-client: merge branch 'th/client-tests-refactor-base-class'

Browse Source 
https://gitlab.freedesktop.org/NetworkManager/NetworkManager/-/merge_requests/1628
This commit is contained in:
Thomas Haller
2023-05-15 15:02:49 +02:00
parent 33f7ee0bb7 efc14fcbec
commit a235bb3c6f
1 changed files with 276 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files

491
src/tests/client/test-client.py
View File

@@ -411,6 +411,34 @@ class Util:
except:
return None
@staticmethod
def file_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
@staticmethod
def _replace_text_match_join(split_arr, replacement):
yield split_arr[0]
@@ -541,6 +569,147 @@ class Util:
idx = pexp.expect(pattern_list)
del pattern_list[idx]
@staticmethod
def skip_without_pexpect(_func=None):
if _func is None:
if pexpect is None:
raise unittest.SkipTest("pexpect not available")
return
def f(*a, **kw):
Util.skip_without_pexpect()
_func(*a, **kw)
return f
@staticmethod
def skip_without_dbus_session(_func=None):
if _func is None:
if not dbus_session_inited:
raise unittest.SkipTest(
"Own D-Bus session for testing is not initialized. Do you have dbus-run-session available?"
)
return
def f(*a, **kw):
Util.skip_without_dbus_session()
_func(*a, **kw)
return f
@staticmethod
def skip_without_NM(_func=None):
if _func is None:
if NM is None:
raise unittest.SkipTest(
"gi.NM is not available. Did you build with introspection?"
)
return
def f(*a, **kw):
Util.skip_without_NM()
_func(*a, **kw)
return f
@staticmethod
def cmd_create_env(
lang="C",
calling_num=None,
fatal_warnings=_DEFAULT_ARG,
extra_env=None,
):
if lang == "C":
language = ""
elif lang == "de_DE.utf8":
language = "de"
elif lang == "pl_PL.UTF-8":
language = "pl"
else:
raise AssertionError("invalid language %s" % (lang))
env = {}
for k in [
"LD_LIBRARY_PATH",
"DBUS_SESSION_BUS_ADDRESS",
"LIBNM_CLIENT_DEBUG",
"LIBNM_CLIENT_DEBUG_FILE",
]:
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()
if calling_num is not None:
env["NM_TEST_CALLING_NUM"] = str(calling_num)
if fatal_warnings is _DEFAULT_ARG or fatal_warnings:
env["G_DEBUG"] = "fatal-warnings"
if extra_env is not None:
for k, v in extra_env.items():
env[k] = v
return env
@staticmethod
def cmd_create_argv(cmd_path, args, with_valgrind=None):
if with_valgrind is None:
with_valgrind = conf.get(ENV_NM_TEST_VALGRIND)
valgrind_log = None
cmd = conf.get(cmd_path)
if with_valgrind:
valgrind_log = tempfile.mkstemp(prefix="nm-test-client-valgrind.")
argv = [
"valgrind",
"--quiet",
"--error-exitcode=37",
"--leak-check=full",
"--gen-suppressions=all",
(
"--suppressions="
+ PathConfiguration.top_srcdir()
+ "/valgrind.suppressions"
),
"--num-callers=100",
"--log-file=" + valgrind_log[1],
cmd,
]
libtool = conf.get(ENV_LIBTOOL)
if libtool:
argv = list(libtool) + ["--mode=execute"] + argv
else:
argv = [cmd]
argv.extend(args)
return argv, valgrind_log
@staticmethod
def cmd_call_pexpect(cmd_path, args, extra_env):
argv, valgrind_log = Util.cmd_create_argv(cmd_path, args)
env = Util.cmd_create_env(extra_env=extra_env)
pexp = pexpect.spawn(argv[0], argv[1:], timeout=10, env=env)
pexp.str_last_chars = 100000
typ = collections.namedtuple("CallPexpect", ["pexp", "valgrind_log"])
return typ(pexp, valgrind_log)
@staticmethod
def cmd_call_pexpect_nmcli(args):
return Util.cmd_call_pexpect(
ENV_NM_TEST_CLIENT_NMCLI_PATH,
args,
{"NO_COLOR": "1"},
)
###############################################################################
@@ -779,6 +948,12 @@ class NMStubServer:
)
return u
def ReplaceTextConUuid(self, con_name, replacement):
return Util.ReplaceTextSimple(
Util.memoize_nullary(lambda: self.findConnectionUuid(con_name)),
replacement,
)
def setProperty(self, path, propname, value, iface_name=None):
if iface_name is None:
iface_name = ""
@@ -875,21 +1050,25 @@ class AsyncProcess:
###############################################################################
MAX_JOBS = 15
class NMTestContext:
MAX_JOBS = 15
class TestNmClient(unittest.TestCase):
def __init__(self, *args, **kwargs):
def __init__(self, testMethodName):
self.testMethodName = testMethodName
self._calling_num = {}
self._skip_test_for_l10n_diff = []
self._async_jobs = []
self._results = []
self.ctx_results = []
self.srv = None
unittest.TestCase.__init__(self, *args, **kwargs)
def calling_num(self, calling_fcn):
calling_num = self._calling_num.get(calling_fcn, 0) + 1
self._calling_num[calling_fcn] = calling_num
return calling_num
def srv_start(self):
self.srv_shutdown()
self.srv = NMStubServer(self._testMethodName)
self.srv = NMStubServer(self.testMethodName)
def srv_shutdown(self):
if self.srv is not None:
@@ -897,136 +1076,18 @@ class TestNmClient(unittest.TestCase):
self.srv = None
srv.shutdown()
def ReplaceTextConUuid(self, con_name, replacement):
return Util.ReplaceTextSimple(
Util.memoize_nullary(lambda: self.srv.findConnectionUuid(con_name)),
replacement,
)
@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 _env(
self, lang="C", calling_num=None, fatal_warnings=_DEFAULT_ARG, extra_env=None
):
if lang == "C":
language = ""
elif lang == "de_DE.utf8":
language = "de"
elif lang == "pl_PL.UTF-8":
language = "pl"
else:
self.fail("invalid language %s" % (lang))
env = {}
for k in [
"LD_LIBRARY_PATH",
"DBUS_SESSION_BUS_ADDRESS",
"LIBNM_CLIENT_DEBUG",
"LIBNM_CLIENT_DEBUG_FILE",
]:
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()
if calling_num is not None:
env["NM_TEST_CALLING_NUM"] = str(calling_num)
if fatal_warnings is _DEFAULT_ARG or fatal_warnings:
env["G_DEBUG"] = "fatal-warnings"
if extra_env is not None:
for k, v in extra_env.items():
env[k] = v
return env
def cmd_construct_argv(self, cmd_path, args, with_valgrind=None):
if with_valgrind is None:
with_valgrind = conf.get(ENV_NM_TEST_VALGRIND)
valgrind_log = None
cmd = conf.get(cmd_path)
if with_valgrind:
valgrind_log = tempfile.mkstemp(prefix="nm-test-client-valgrind.")
argv = [
"valgrind",
"--quiet",
"--error-exitcode=37",
"--leak-check=full",
"--gen-suppressions=all",
(
"--suppressions="
+ PathConfiguration.top_srcdir()
+ "/valgrind.suppressions"
),
"--num-callers=100",
"--log-file=" + valgrind_log[1],
cmd,
]
libtool = conf.get(ENV_LIBTOOL)
if libtool:
argv = list(libtool) + ["--mode=execute"] + argv
else:
argv = [cmd]
argv.extend(args)
return argv, valgrind_log
def call_pexpect(self, cmd_path, args, extra_env):
argv, valgrind_log = self.cmd_construct_argv(cmd_path, args)
env = self._env(extra_env=extra_env)
pexp = pexpect.spawn(argv[0], argv[1:], timeout=10, env=env)
pexp.str_last_chars = 100000
typ = collections.namedtuple("CallPexpect", ["pexp", "valgrind_log"])
return typ(pexp, valgrind_log)
def async_start(self, wait_all=False):
while True:
while True:
for async_job in list(self._async_jobs[0:MAX_JOBS]):
for async_job in list(self._async_jobs[0 : self.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]):
for async_job in list(self._async_jobs[0 : self.MAX_JOBS]):
if async_job.poll() is not None:
self._async_jobs.remove(async_job)
async_job.wait_and_complete()
@@ -1034,7 +1095,7 @@ class TestNmClient(unittest.TestCase):
jobs_running.append(async_job)
if len(jobs_running) >= len(self._async_jobs):
break
if len(jobs_running) >= MAX_JOBS:
if len(jobs_running) >= self.MAX_JOBS:
break
if not jobs_running:
@@ -1054,7 +1115,10 @@ class TestNmClient(unittest.TestCase):
def async_wait(self):
return self.async_start(wait_all=True)
def _nm_test_post(self):
def async_append_job(self, async_job):
self._async_jobs.append(async_job)
def run_post(self):
self.async_wait()
@@ -1062,8 +1126,8 @@ class TestNmClient(unittest.TestCase):
self._calling_num = None
results = self._results
self._results = None
results = self.ctx_results
self.ctx_results = None
if len(results) == 0:
return
@@ -1071,18 +1135,16 @@ class TestNmClient(unittest.TestCase):
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
+ self.testMethodName
+ ".expected"
)
regenerate = conf.get(ENV_NM_TEST_REGENERATE)
content_expect, results_expect = self._read_expected(filename)
content_expect, results_expect = Util.file_read_expected(filename)
if results_expect is None:
if not regenerate:
@@ -1159,21 +1221,21 @@ class TestNmClient(unittest.TestCase):
# 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(
raise unittest.SkipTest(
"Skipped asserting for localized tests %s. Set NM_TEST_CLIENT_CHECK_L10N=1 to force fail."
% (",".join(skip_test_for_l10n_diff))
)
###############################################################################
class TestNmcli(unittest.TestCase):
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?")
Util.skip_without_dbus_session()
Util.skip_without_NM()
self.ctx = NMTestContext(self._testMethodName)
class TestNmcli(TestNmClient):
def call_nmcli_l(
self,
args,
@@ -1256,9 +1318,6 @@ class TestNmcli(TestNmClient):
frame,
)
def call_nmcli_pexpect(self, args):
return self.call_pexpect(ENV_NM_TEST_CLIENT_NMCLI_PATH, args, {"NO_COLOR": "1"})
def _call_nmcli(
self,
args,
@@ -1278,11 +1337,10 @@ class TestNmcli(TestNmClient):
):
if sync_barrier:
self.async_wait()
self.ctx.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
calling_num = self.ctx.calling_num(calling_fcn)
test_name = "%s-%03d" % (calling_fcn, calling_num)
@@ -1319,7 +1377,7 @@ class TestNmcli(TestNmClient):
self.fail("invalid language %s" % (lang))
# Running under valgrind is not yet supported for those tests.
args, valgrind_log = self.cmd_construct_argv(
args, valgrind_log = Util.cmd_create_argv(
ENV_NM_TEST_CLIENT_NMCLI_PATH, args, with_valgrind=False
)
@@ -1351,8 +1409,8 @@ class TestNmcli(TestNmClient):
if expected_stderr is _DEFAULT_ARG:
expected_stderr = None
results_idx = len(self._results)
self._results.append(None)
results_idx = len(self.ctx.ctx_results)
self.ctx.ctx_results.append(None)
def complete_cb(async_job, returncode, stdout, stderr):
@@ -1427,55 +1485,47 @@ class TestNmcli(TestNmClient):
)
content = ("size: %s\n" % (len(content))).encode("utf8") + content
self._results[results_idx] = {
self.ctx.ctx_results[results_idx] = {
"test_name": test_name,
"ignore_l10n_diff": ignore_l10n_diff,
"content": content,
}
env = self._env(lang, calling_num, fatal_warnings, extra_env)
env = Util.cmd_create_env(lang, calling_num, fatal_warnings, extra_env)
async_job = AsyncProcess(args=args, env=env, complete_cb=complete_cb)
self._async_jobs.append(async_job)
self.ctx.async_append_job(async_job)
self.async_start(wait_all=sync_barrier)
self.ctx.async_start(wait_all=sync_barrier)
def nm_test(func):
def f(self):
self.srv_start()
self.ctx.srv_start()
func(self)
self._nm_test_post()
self.ctx.run_post()
return f
def nm_test_no_dbus(func):
def f(self):
func(self)
self._nm_test_post()
return f
def skip_without_pexpect(func):
def f(self):
if pexpect is None:
raise unittest.SkipTest("pexpect not available")
func(self)
self.ctx.run_post()
return f
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.ctx.srv.op_AddObj("WiredDevice", iface="eth0")
self.ctx.srv.op_AddObj("WiredDevice", iface="eth1")
self.ctx.srv.op_AddObj("WifiDevice", iface="wlan0")
self.ctx.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.ctx.srv.op_AddObj("WifiDevice", ident="wlan1/x", iface="wlan1")
self.srv.op_AddObj("WifiAp", device="wlan0", rsnf=0x0)
self.ctx.srv.op_AddObj("WifiAp", device="wlan0", rsnf=0x0)
self.srv.op_AddObj("WifiAp", device="wlan0")
self.ctx.srv.op_AddObj("WifiAp", device="wlan0")
NM_AP_FLAGS = getattr(NM, "80211ApSecurityFlags")
rsnf = 0x0
@@ -1484,11 +1534,11 @@ class TestNmcli(TestNmClient):
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.ctx.srv.op_AddObj("WifiAp", device="wlan0", wpaf=0x0, rsnf=rsnf)
self.srv.op_AddObj("WifiAp", device="wlan1")
self.ctx.srv.op_AddObj("WifiAp", device="wlan1")
self.srv.addConnection(
self.ctx.srv.addConnection(
{"connection": {"type": "802-3-ethernet", "id": "con-1"}}
)
@@ -1548,7 +1598,9 @@ class TestNmcli(TestNmClient):
replace_uuids = []
replace_uuids.append(
self.ReplaceTextConUuid("con-xx1", "UUID-con-xx1-REPLACED-REPLACED-REPLA")
self.ctx.srv.ReplaceTextConUuid(
"con-xx1", "UUID-con-xx1-REPLACED-REPLACED-REPLA"
)
)
self.call_nmcli(
@@ -1561,7 +1613,7 @@ class TestNmcli(TestNmClient):
for con_name, apn in con_gsm_list:
replace_uuids.append(
self.ReplaceTextConUuid(
self.ctx.srv.ReplaceTextConUuid(
con_name, "UUID-" + con_name + "-REPLACED-REPLACED-REPL"
)
)
@@ -1593,7 +1645,9 @@ class TestNmcli(TestNmClient):
)
replace_uuids.append(
self.ReplaceTextConUuid("ethernet", "UUID-ethernet-REPLACED-REPLACED-REPL")
self.ctx.srv.ReplaceTextConUuid(
"ethernet", "UUID-ethernet-REPLACED-REPLACED-REPL"
)
)
self.call_nmcli(
@@ -1676,9 +1730,9 @@ class TestNmcli(TestNmClient):
["-f", "ALL", "-t", "dev", "show", "eth0"], replace_stdout=replace_uuids
)
self.async_wait()
self.ctx.async_wait()
self.srv.setProperty(
self.ctx.srv.setProperty(
"/org/freedesktop/NetworkManager/ActiveConnection/1",
"State",
dbus.UInt32(NM.ActiveConnectionState.DEACTIVATING),
@@ -1688,8 +1742,8 @@ class TestNmcli(TestNmClient):
for i in [0, 1]:
if i == 1:
self.async_wait()
self.srv.op_ConnectionSetVisible(False, con_id="ethernet")
self.ctx.async_wait()
self.ctx.srv.op_ConnectionSetVisible(False, con_id="ethernet")
for mode in Util.iter_nmcli_output_modes():
self.call_nmcli_l(
@@ -1722,7 +1776,9 @@ class TestNmcli(TestNmClient):
replace_uuids = []
replace_uuids.append(
self.ReplaceTextConUuid("con-xx1", "UUID-con-xx1-REPLACED-REPLACED-REPLA")
self.ctx.srv.ReplaceTextConUuid(
"con-xx1", "UUID-con-xx1-REPLACED-REPLACED-REPLA"
)
)
self.call_nmcli(
@@ -1765,10 +1821,12 @@ class TestNmcli(TestNmClient):
)
self.call_nmcli_l(["con", "s", "con-xx1"], replace_stdout=replace_uuids)
self.async_wait()
self.ctx.async_wait()
replace_uuids.append(
self.ReplaceTextConUuid("con-vpn-1", "UUID-con-vpn-1-REPLACED-REPLACED-REP")
self.ctx.srv.ReplaceTextConUuid(
"con-vpn-1", "UUID-con-vpn-1-REPLACED-REPLACED-REP"
)
)
self.call_nmcli(
@@ -1799,16 +1857,16 @@ class TestNmcli(TestNmClient):
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.ctx.async_wait()
self.srv.setProperty(
self.ctx.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
[c[1] for c in self.ctx.srv.findConnections()], replace_uuids
)
self.call_nmcli_l([], replace_stdout=replace_uuids)
@@ -2076,10 +2134,10 @@ class TestNmcli(TestNmClient):
extra_env=no_dbus_env,
)
@skip_without_pexpect
@Util.skip_without_pexpect
@nm_test
def test_ask_mode(self):
nmc = self.call_nmcli_pexpect(["--ask", "c", "add"])
nmc = Util.cmd_call_pexpect_nmcli(["--ask", "c", "add"])
nmc.pexp.expect("Connection type:")
nmc.pexp.sendline("ethernet")
nmc.pexp.expect("Interface name:")
@@ -2100,11 +2158,11 @@ class TestNmcli(TestNmClient):
nmc.pexp.expect(pexpect.EOF)
Util.valgrind_check_log(nmc.valgrind_log, "test_ask_mode")
@skip_without_pexpect
@Util.skip_without_pexpect
@nm_test
def test_monitor(self):
def start_mon(self):
nmc = self.call_nmcli_pexpect(["monitor"])
nmc = Util.cmd_call_pexpect_nmcli(["monitor"])
nmc.pexp.expect("NetworkManager is running")
return nmc
@@ -2115,10 +2173,10 @@ class TestNmcli(TestNmClient):
nmc = start_mon(self)
self.srv.op_AddObj("WiredDevice", iface="eth0")
self.ctx.srv.op_AddObj("WiredDevice", iface="eth0")
nmc.pexp.expect("eth0: device created\r\n")
self.srv.addConnection(
self.ctx.srv.addConnection(
{"connection": {"type": "802-3-ethernet", "id": "con-1"}}
)
nmc.pexp.expect("con-1: connection profile created\r\n")
@@ -2126,7 +2184,7 @@ class TestNmcli(TestNmClient):
end_mon(self, nmc)
nmc = start_mon(self)
self.srv_shutdown()
self.ctx.srv_shutdown()
Util.pexpect_expect_all(
nmc.pexp,
"con-1: connection profile removed",
@@ -2139,7 +2197,11 @@ class TestNmcli(TestNmClient):
###############################################################################
class TestNmCloudSetup(TestNmClient):
class TestNmCloudSetup(unittest.TestCase):
def setUp(self):
Util.skip_without_dbus_session()
Util.skip_without_NM()
self.ctx = NMTestContext(self._testMethodName)
_mac1 = "9e:c0:3e:92:24:2d"
_mac2 = "53:e9:7e:52:8d:a8"
@@ -2153,8 +2215,7 @@ class TestNmCloudSetup(TestNmClient):
"""
def f(self):
if pexpect is None:
raise unittest.SkipTest("pexpect not available")
Util.skip_without_pexpect()
if tuple(sys.version_info[0:2]) < (3, 2):
# subprocess.Popen()'s "pass_fd" argument requires at least Python 3.2.
@@ -2191,12 +2252,12 @@ class TestNmCloudSetup(TestNmClient):
error = None
self.srv_start()
self.ctx.srv_start()
try:
func(self)
except Exception as e:
error = e
self._nm_test_post()
self.ctx.run_post()
self.md_conn.close()
p.stdin.close()
@@ -2210,8 +2271,8 @@ class TestNmCloudSetup(TestNmClient):
def _mock_devices(self):
# Add a device with an active connection that has IPv4 configured
self.srv.op_AddObj("WiredDevice", iface="eth0", mac="9e:c0:3e:92:24:2d")
self.srv.addAndActivateConnection(
self.ctx.srv.op_AddObj("WiredDevice", iface="eth0", mac="9e:c0:3e:92:24:2d")
self.ctx.srv.addAndActivateConnection(
{
"connection": {"type": "802-3-ethernet", "id": "con-eth0"},
"ipv4": {"method": "auto"},
@@ -2221,8 +2282,8 @@ class TestNmCloudSetup(TestNmClient):
)
# The second connection has no IPv4
self.srv.op_AddObj("WiredDevice", iface="eth1", mac="53:e9:7e:52:8d:a8")
self.srv.addAndActivateConnection(
self.ctx.srv.op_AddObj("WiredDevice", iface="eth1", mac="53:e9:7e:52:8d:a8")
self.ctx.srv.addAndActivateConnection(
{"connection": {"type": "802-3-ethernet", "id": "con-eth1"}},
"/org/freedesktop/NetworkManager/Devices/2",
"",
@@ -2279,7 +2340,7 @@ class TestNmCloudSetup(TestNmClient):
)
# Run nm-cloud-setup for the first time
nmc = self.call_pexpect(
nmc = Util.cmd_call_pexpect(
ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH,
[],
{
@@ -2301,7 +2362,7 @@ class TestNmCloudSetup(TestNmClient):
nmc.pexp.expect(pexpect.EOF)
# Run nm-cloud-setup for the second time
nmc = self.call_pexpect(
nmc = Util.cmd_call_pexpect(
ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH,
[],
{
@@ -2358,7 +2419,7 @@ class TestNmCloudSetup(TestNmClient):
self._mock_path(_azure_iface + "1/ipv4/subnet/0/prefix/" + _azure_query, "20")
# Run nm-cloud-setup for the first time
nmc = self.call_pexpect(
nmc = Util.cmd_call_pexpect(
ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH,
[],
{
@@ -2387,7 +2448,7 @@ class TestNmCloudSetup(TestNmClient):
nmc.pexp.expect(pexpect.EOF)
# Run nm-cloud-setup for the second time
nmc = self.call_pexpect(
nmc = Util.cmd_call_pexpect(
ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH,
[],
{
@@ -2434,7 +2495,7 @@ class TestNmCloudSetup(TestNmClient):
)
# Run nm-cloud-setup for the first time
nmc = self.call_pexpect(
nmc = Util.cmd_call_pexpect(
ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH,
[],
{
@@ -2456,7 +2517,7 @@ class TestNmCloudSetup(TestNmClient):
nmc.pexp.expect(pexpect.EOF)
# Run nm-cloud-setup for the second time
nmc = self.call_pexpect(
nmc = Util.cmd_call_pexpect(
ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH,
[],
{
@@ -2494,7 +2555,7 @@ class TestNmCloudSetup(TestNmClient):
self._mock_path(gcp_iface + "1/forwarded-ips/0", TestNmCloudSetup._ip2)
# Run nm-cloud-setup for the first time
nmc = self.call_pexpect(
nmc = Util.cmd_call_pexpect(
ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH,
[],
{
@@ -2517,7 +2578,7 @@ class TestNmCloudSetup(TestNmClient):
nmc.pexp.expect(pexpect.EOF)
# Run nm-cloud-setup for the second time
nmc = self.call_pexpect(
nmc = Util.cmd_call_pexpect(
ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH,
[],
{