diff options
author | Thomas Haller <thaller@redhat.com> | 2023-05-15 15:02:49 +0200 |
---|---|---|
committer | Thomas Haller <thaller@redhat.com> | 2023-05-15 15:03:09 +0200 |
commit | a235bb3c6f03e93b92dda231f32a5fbc178d8328 (patch) | |
tree | f8360c5f54430cb14ed7165cbcb85ca2014fc677 | |
parent | 33f7ee0bb7f92900b5f7f929e4feff6f6b872472 (diff) | |
parent | efc14fcbec90cc057f37d8a4fca48c5b5b15b786 (diff) | |
download | NetworkManager-a235bb3c6f03e93b92dda231f32a5fbc178d8328.tar.gz |
test-client: merge branch 'th/client-tests-refactor-base-class'
https://gitlab.freedesktop.org/NetworkManager/NetworkManager/-/merge_requests/1628
-rwxr-xr-x | src/tests/client/test-client.py | 491 |
1 files changed, 276 insertions, 215 deletions
diff --git a/src/tests/client/test-client.py b/src/tests/client/test-client.py index dc9efd8553..9aa018b61c 100755 --- a/src/tests/client/test-client.py +++ b/src/tests/client/test-client.py @@ -412,6 +412,34 @@ class Util: 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] for t in split_arr[1:]: @@ -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)) ) - 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?") + +############################################################################### -class TestNmcli(TestNmClient): +class TestNmcli(unittest.TestCase): + def setUp(self): + Util.skip_without_dbus_session() + Util.skip_without_NM() + self.ctx = NMTestContext(self._testMethodName) + 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, [], { |