diff options
Diffstat (limited to 'src/tests/client/test-client.py')
| -rwxr-xr-x | src/tests/client/test-client.py | 779 |
1 files changed, 564 insertions, 215 deletions
diff --git a/src/tests/client/test-client.py b/src/tests/client/test-client.py index b6f7cf00a4..9aa018b61c 100755 --- a/src/tests/client/test-client.py +++ b/src/tests/client/test-client.py @@ -147,6 +147,7 @@ except ImportError: try: from http.server import HTTPServer from http.server import BaseHTTPRequestHandler + from http.client import HTTPConnection, HTTPResponse except ImportError: HTTPServer = None @@ -411,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:]: @@ -540,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"}, + ) + ############################################################################### @@ -778,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 = "" @@ -874,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: @@ -896,134 +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) - - 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() @@ -1031,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: @@ -1051,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() @@ -1059,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 @@ -1068,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: @@ -1156,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(unittest.TestCase): + def setUp(self): + Util.skip_without_dbus_session() + Util.skip_without_NM() + self.ctx = NMTestContext(self._testMethodName) -class TestNmcli(TestNmClient): def call_nmcli_l( self, args, @@ -1253,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, @@ -1275,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) @@ -1316,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 ) @@ -1348,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): @@ -1424,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 @@ -1481,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"}} ) @@ -1545,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( @@ -1558,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" ) ) @@ -1590,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( @@ -1673,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), @@ -1685,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( @@ -1719,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( @@ -1762,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( @@ -1796,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) @@ -2073,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:") @@ -2097,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 @@ -2112,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") @@ -2123,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", @@ -2136,15 +2197,25 @@ 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" + + _ip1 = "172.31.26.249" + _ip2 = "172.31.176.249" + def cloud_setup_test(func): """ Runs the mock NetworkManager along with a mock cloud metadata service. """ 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. @@ -2160,35 +2231,48 @@ class TestNmCloudSetup(TestNmClient): # hallucinogenic substances. s.listen(5) + def pass_socket(): + os.dup2(s.fileno(), 3) + service_path = PathConfiguration.test_cloud_meta_mock_path() env = os.environ.copy() - env["LISTEN_FD"] = str(s.fileno()) + env["LISTEN_FDS"] = "1" p = subprocess.Popen( - [sys.executable, service_path], + [sys.executable, service_path, "--empty"], stdin=subprocess.PIPE, env=env, - pass_fds=(s.fileno(),), + pass_fds=(3,), + preexec_fn=pass_socket, ) - self.md_url = "http://%s:%d" % s.getsockname() + (hostaddr, port) = s.getsockname() + self.md_conn = HTTPConnection(hostaddr, port=port) + self.md_url = "http://%s:%d" % (hostaddr, port) s.close() - self.srv_start() - func(self) - self._nm_test_post() + error = None + + self.ctx.srv_start() + try: + func(self) + except Exception as e: + error = e + self.ctx.run_post() + self.md_conn.close() p.stdin.close() p.terminate() p.wait() - return f + if error: + raise error - @cloud_setup_test - def test_ec2(self): + return f + def _mock_devices(self): # Add a device with an active connection that has IPv4 configured - self.srv.op_AddObj("WiredDevice", iface="eth0") - 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"}, @@ -2198,16 +2282,220 @@ class TestNmCloudSetup(TestNmClient): ) # The second connection has no IPv4 - self.srv.op_AddObj("WiredDevice", iface="eth1") - 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", "", delay=0, ) + def _mock_path(self, path, body): + self.md_conn.request("PUT", path, body=body) + self.md_conn.getresponse().read() + + @cloud_setup_test + def test_aliyun(self): + self._mock_devices() + + _aliyun_meta = "/2016-01-01/meta-data/" + _aliyun_macs = _aliyun_meta + "network/interfaces/macs/" + self._mock_path(_aliyun_meta, "ami-id\n") + self._mock_path( + _aliyun_macs, TestNmCloudSetup._mac2 + "\n" + TestNmCloudSetup._mac1 + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac2 + "/vpc-cidr-block", "172.31.16.0/20" + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac2 + "/private-ipv4s", + TestNmCloudSetup._ip1, + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac2 + "/primary-ip-address", + TestNmCloudSetup._ip1, + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac2 + "/netmask", "255.255.255.0" + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac2 + "/gateway", "172.31.26.2" + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac1 + "/vpc-cidr-block", "172.31.166.0/20" + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac1 + "/private-ipv4s", + TestNmCloudSetup._ip2, + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac1 + "/primary-ip-address", + TestNmCloudSetup._ip2, + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac1 + "/netmask", "255.255.255.0" + ) + self._mock_path( + _aliyun_macs + TestNmCloudSetup._mac1 + "/gateway", "172.31.176.2" + ) + # Run nm-cloud-setup for the first time - nmc = self.call_pexpect( + nmc = Util.cmd_call_pexpect( + ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH, + [], + { + "NM_CLOUD_SETUP_ALIYUN_HOST": self.md_url, + "NM_CLOUD_SETUP_LOG": "trace", + "NM_CLOUD_SETUP_ALIYUN": "yes", + }, + ) + + nmc.pexp.expect("provider aliyun detected") + nmc.pexp.expect("found interfaces: 9E:C0:3E:92:24:2D, 53:E9:7E:52:8D:A8") + nmc.pexp.expect("get-config: start fetching meta data") + nmc.pexp.expect("get-config: success") + nmc.pexp.expect("meta data received") + # One of the devices has no IPv4 configuration to be modified + nmc.pexp.expect("device has no suitable applied connection. Skip") + # The other one was lacking an address set it up. + nmc.pexp.expect("some changes were applied for provider aliyun") + nmc.pexp.expect(pexpect.EOF) + + # Run nm-cloud-setup for the second time + nmc = Util.cmd_call_pexpect( + ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH, + [], + { + "NM_CLOUD_SETUP_ALIYUN_HOST": self.md_url, + "NM_CLOUD_SETUP_LOG": "trace", + "NM_CLOUD_SETUP_ALIYUN": "yes", + }, + ) + + nmc.pexp.expect("provider aliyun detected") + nmc.pexp.expect("found interfaces: 9E:C0:3E:92:24:2D, 53:E9:7E:52:8D:A8") + nmc.pexp.expect("get-config: starting") + nmc.pexp.expect("get-config: success") + nmc.pexp.expect("meta data received") + # No changes this time + nmc.pexp.expect('device needs no update to applied connection "con-eth0"') + nmc.pexp.expect("no changes were applied for provider aliyun") + nmc.pexp.expect(pexpect.EOF) + + Util.valgrind_check_log(nmc.valgrind_log, "test_aliyun") + + @cloud_setup_test + def test_azure(self): + self._mock_devices() + + _azure_meta = "/metadata/instance" + _azure_iface = _azure_meta + "/network/interface/" + _azure_query = "?format=text&api-version=2017-04-02" + self._mock_path(_azure_meta + _azure_query, "") + self._mock_path(_azure_iface + _azure_query, "0\n1\n") + self._mock_path( + _azure_iface + "0/macAddress" + _azure_query, TestNmCloudSetup._mac1 + ) + self._mock_path( + _azure_iface + "1/macAddress" + _azure_query, TestNmCloudSetup._mac2 + ) + self._mock_path(_azure_iface + "0/ipv4/ipAddress/" + _azure_query, "0\n") + self._mock_path(_azure_iface + "1/ipv4/ipAddress/" + _azure_query, "0\n") + self._mock_path( + _azure_iface + "0/ipv4/ipAddress/0/privateIpAddress" + _azure_query, + TestNmCloudSetup._ip1, + ) + self._mock_path( + _azure_iface + "1/ipv4/ipAddress/0/privateIpAddress" + _azure_query, + TestNmCloudSetup._ip2, + ) + self._mock_path( + _azure_iface + "0/ipv4/subnet/0/address/" + _azure_query, "172.31.16.0" + ) + self._mock_path( + _azure_iface + "1/ipv4/subnet/0/address/" + _azure_query, "172.31.166.0" + ) + self._mock_path(_azure_iface + "0/ipv4/subnet/0/prefix/" + _azure_query, "20") + self._mock_path(_azure_iface + "1/ipv4/subnet/0/prefix/" + _azure_query, "20") + + # Run nm-cloud-setup for the first time + nmc = Util.cmd_call_pexpect( + ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH, + [], + { + "NM_CLOUD_SETUP_AZURE_HOST": self.md_url, + "NM_CLOUD_SETUP_LOG": "trace", + "NM_CLOUD_SETUP_AZURE": "yes", + }, + ) + + nmc.pexp.expect("provider azure detected") + nmc.pexp.expect("found interfaces: 9E:C0:3E:92:24:2D, 53:E9:7E:52:8D:A8") + nmc.pexp.expect("found azure interfaces: 2") + nmc.pexp.expect("interface\[0]: found a matching device with hwaddr") + nmc.pexp.expect( + "interface\[0]: (received subnet address|received subnet prefix 20)" + ) + nmc.pexp.expect( + "interface\[0]: (received subnet address|received subnet prefix 20)" + ) + nmc.pexp.expect("get-config: success") + nmc.pexp.expect("meta data received") + # One of the devices has no IPv4 configuration to be modified + nmc.pexp.expect("device has no suitable applied connection. Skip") + # The other one was lacking an address set it up. + nmc.pexp.expect("some changes were applied for provider azure") + nmc.pexp.expect(pexpect.EOF) + + # Run nm-cloud-setup for the second time + nmc = Util.cmd_call_pexpect( + ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH, + [], + { + "NM_CLOUD_SETUP_AZURE_HOST": self.md_url, + "NM_CLOUD_SETUP_LOG": "trace", + "NM_CLOUD_SETUP_AZURE": "yes", + }, + ) + + nmc.pexp.expect("provider azure detected") + nmc.pexp.expect("found interfaces: 9E:C0:3E:92:24:2D, 53:E9:7E:52:8D:A8") + nmc.pexp.expect("get-config: starting") + nmc.pexp.expect("get-config: success") + nmc.pexp.expect("meta data received") + # No changes this time + nmc.pexp.expect('device needs no update to applied connection "con-eth0"') + nmc.pexp.expect("no changes were applied for provider azure") + nmc.pexp.expect(pexpect.EOF) + + Util.valgrind_check_log(nmc.valgrind_log, "test_azure") + + @cloud_setup_test + def test_ec2(self): + self._mock_devices() + + _ec2_macs = "/2018-09-24/meta-data/network/interfaces/macs/" + self._mock_path("/latest/meta-data/", "ami-id\n") + self._mock_path( + _ec2_macs, TestNmCloudSetup._mac2 + "\n" + TestNmCloudSetup._mac1 + ) + self._mock_path( + _ec2_macs + TestNmCloudSetup._mac2 + "/subnet-ipv4-cidr-block", + "172.31.16.0/20", + ) + self._mock_path( + _ec2_macs + TestNmCloudSetup._mac2 + "/local-ipv4s", TestNmCloudSetup._ip1 + ) + self._mock_path( + _ec2_macs + TestNmCloudSetup._mac1 + "/subnet-ipv4-cidr-block", + "172.31.166.0/20", + ) + self._mock_path( + _ec2_macs + TestNmCloudSetup._mac1 + "/local-ipv4s", TestNmCloudSetup._ip2 + ) + + # Run nm-cloud-setup for the first time + nmc = Util.cmd_call_pexpect( ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH, [], { @@ -2229,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, [], { @@ -2251,6 +2539,67 @@ class TestNmCloudSetup(TestNmClient): Util.valgrind_check_log(nmc.valgrind_log, "test_ec2") + @cloud_setup_test + def test_gcp(self): + self._mock_devices() + + gcp_meta = "/computeMetadata/v1/instance/" + gcp_iface = gcp_meta + "network-interfaces/" + self._mock_path(gcp_meta + "id", "") + self._mock_path(gcp_iface, "0\n1\n") + self._mock_path(gcp_iface + "0/mac", TestNmCloudSetup._mac1) + self._mock_path(gcp_iface + "1/mac", TestNmCloudSetup._mac2) + self._mock_path(gcp_iface + "0/forwarded-ips/", "0\n") + self._mock_path(gcp_iface + "0/forwarded-ips/0", TestNmCloudSetup._ip1) + self._mock_path(gcp_iface + "1/forwarded-ips/", "0\n") + self._mock_path(gcp_iface + "1/forwarded-ips/0", TestNmCloudSetup._ip2) + + # Run nm-cloud-setup for the first time + nmc = Util.cmd_call_pexpect( + ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH, + [], + { + "NM_CLOUD_SETUP_GCP_HOST": self.md_url, + "NM_CLOUD_SETUP_LOG": "trace", + "NM_CLOUD_SETUP_GCP": "yes", + }, + ) + + nmc.pexp.expect("provider GCP detected") + nmc.pexp.expect("found interfaces: 9E:C0:3E:92:24:2D, 53:E9:7E:52:8D:A8") + nmc.pexp.expect("found GCP interfaces: 2") + nmc.pexp.expect("GCP interface\[0]: found a requested device with hwaddr") + nmc.pexp.expect("get-config: success") + nmc.pexp.expect("meta data received") + # One of the devices has no IPv4 configuration to be modified + nmc.pexp.expect("device has no suitable applied connection. Skip") + # The other one was lacking an address set it up. + nmc.pexp.expect("some changes were applied for provider GCP") + nmc.pexp.expect(pexpect.EOF) + + # Run nm-cloud-setup for the second time + nmc = Util.cmd_call_pexpect( + ENV_NM_TEST_CLIENT_CLOUD_SETUP_PATH, + [], + { + "NM_CLOUD_SETUP_GCP_HOST": self.md_url, + "NM_CLOUD_SETUP_LOG": "trace", + "NM_CLOUD_SETUP_GCP": "yes", + }, + ) + + nmc.pexp.expect("provider GCP detected") + nmc.pexp.expect("found interfaces: 9E:C0:3E:92:24:2D, 53:E9:7E:52:8D:A8") + nmc.pexp.expect("get-config: starting") + nmc.pexp.expect("get-config: success") + nmc.pexp.expect("meta data received") + # No changes this time + nmc.pexp.expect('device needs no update to applied connection "con-eth0"') + nmc.pexp.expect("no changes were applied for provider GCP") + nmc.pexp.expect(pexpect.EOF) + + Util.valgrind_check_log(nmc.valgrind_log, "test_gcp") + ############################################################################### |