From 1c32b59ebdd48b9a6315d4c29f72dcaabc645fe2 Mon Sep 17 00:00:00 2001 From: Thomas Haller Date: Thu, 11 May 2023 10:55:50 +0200 Subject: test-client: move TestNmClient.{cmd_construct_argv,_env}() to Util.cmd_create_{argv,env}() These two are stateless method, and even not only work for nmcli, but also for nm-cloud-setup. Move to the Util class, as static methods. --- src/tests/client/test-client.py | 209 +++++++++++++++++++++------------------- 1 file changed, 110 insertions(+), 99 deletions(-) diff --git a/src/tests/client/test-client.py b/src/tests/client/test-client.py index a3f094532b..3cd2a71108 100755 --- a/src/tests/client/test-client.py +++ b/src/tests/client/test-client.py @@ -612,6 +612,104 @@ class Util: 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"}, + ) + ############################################################################### @@ -974,90 +1072,6 @@ class TestNmClient(unittest.TestCase): replacement, ) - 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: @@ -1295,9 +1309,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, @@ -1358,7 +1369,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 ) @@ -1472,7 +1483,7 @@ class TestNmcli(TestNmClient): "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) @@ -2110,7 +2121,7 @@ class TestNmcli(TestNmClient): @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:") @@ -2135,7 +2146,7 @@ class TestNmcli(TestNmClient): @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 @@ -2309,7 +2320,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, [], { @@ -2331,7 +2342,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, [], { @@ -2388,7 +2399,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, [], { @@ -2417,7 +2428,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, [], { @@ -2464,7 +2475,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, [], { @@ -2486,7 +2497,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, [], { @@ -2524,7 +2535,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, [], { @@ -2547,7 +2558,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, [], { -- cgit v1.2.1