diff options
Diffstat (limited to 'tests/unittests/test_cli.py')
| -rw-r--r-- | tests/unittests/test_cli.py | 304 |
1 files changed, 204 insertions, 100 deletions
diff --git a/tests/unittests/test_cli.py b/tests/unittests/test_cli.py index 74f85959..bed73a93 100644 --- a/tests/unittests/test_cli.py +++ b/tests/unittests/test_cli.py @@ -1,13 +1,13 @@ # This file is part of cloud-init. See LICENSE file for license information. -import os +import contextlib import io +import os from collections import namedtuple from cloudinit.cmd import main as cli -from cloudinit.tests import helpers as test_helpers from cloudinit.util import load_file, load_json - +from tests.unittests import helpers as test_helpers mock = test_helpers.mock @@ -23,7 +23,7 @@ class TestCLI(test_helpers.FilesystemMockingTestCase): def _call_main(self, sysv_args=None): if not sysv_args: - sysv_args = ['cloud-init'] + sysv_args = ["cloud-init"] try: return cli.main(sysv_args=sysv_args) except SystemExit as e: @@ -35,36 +35,37 @@ class TestCLI(test_helpers.FilesystemMockingTestCase): Valid name values are only init and modules. """ tmpd = self.tmp_dir() - data_d = self.tmp_path('data', tmpd) - link_d = self.tmp_path('link', tmpd) - FakeArgs = namedtuple('FakeArgs', ['action', 'local', 'mode']) + data_d = self.tmp_path("data", tmpd) + link_d = self.tmp_path("link", tmpd) + FakeArgs = namedtuple("FakeArgs", ["action", "local", "mode"]) def myaction(): - raise Exception('Should not call myaction') + raise Exception("Should not call myaction") - myargs = FakeArgs(('doesnotmatter', myaction), False, 'bogusmode') + myargs = FakeArgs(("doesnotmatter", myaction), False, "bogusmode") with self.assertRaises(ValueError) as cm: - cli.status_wrapper('init1', myargs, data_d, link_d) - self.assertEqual('unknown name: init1', str(cm.exception)) - self.assertNotIn('Should not call myaction', self.logs.getvalue()) + cli.status_wrapper("init1", myargs, data_d, link_d) + self.assertEqual("unknown name: init1", str(cm.exception)) + self.assertNotIn("Should not call myaction", self.logs.getvalue()) def test_status_wrapper_errors_on_invalid_modes(self): """status_wrapper will error if a parameter combination is invalid.""" tmpd = self.tmp_dir() - data_d = self.tmp_path('data', tmpd) - link_d = self.tmp_path('link', tmpd) - FakeArgs = namedtuple('FakeArgs', ['action', 'local', 'mode']) + data_d = self.tmp_path("data", tmpd) + link_d = self.tmp_path("link", tmpd) + FakeArgs = namedtuple("FakeArgs", ["action", "local", "mode"]) def myaction(): - raise Exception('Should not call myaction') + raise Exception("Should not call myaction") - myargs = FakeArgs(('modules_name', myaction), False, 'bogusmode') + myargs = FakeArgs(("modules_name", myaction), False, "bogusmode") with self.assertRaises(ValueError) as cm: - cli.status_wrapper('modules', myargs, data_d, link_d) + cli.status_wrapper("modules", myargs, data_d, link_d) self.assertEqual( "Invalid cloud init mode specified 'modules-bogusmode'", - str(cm.exception)) - self.assertNotIn('Should not call myaction', self.logs.getvalue()) + str(cm.exception), + ) + self.assertNotIn("Should not call myaction", self.logs.getvalue()) def test_status_wrapper_init_local_writes_fresh_status_info(self): """When running in init-local mode, status_wrapper writes status.json. @@ -72,78 +73,90 @@ class TestCLI(test_helpers.FilesystemMockingTestCase): Old status and results artifacts are also removed. """ tmpd = self.tmp_dir() - data_d = self.tmp_path('data', tmpd) - link_d = self.tmp_path('link', tmpd) - status_link = self.tmp_path('status.json', link_d) + data_d = self.tmp_path("data", tmpd) + link_d = self.tmp_path("link", tmpd) + status_link = self.tmp_path("status.json", link_d) # Write old artifacts which will be removed or updated. for _dir in data_d, link_d: test_helpers.populate_dir( - _dir, {'status.json': 'old', 'result.json': 'old'}) + _dir, {"status.json": "old", "result.json": "old"} + ) - FakeArgs = namedtuple('FakeArgs', ['action', 'local', 'mode']) + FakeArgs = namedtuple("FakeArgs", ["action", "local", "mode"]) def myaction(name, args): # Return an error to watch status capture them - return 'SomeDatasource', ['an error'] + return "SomeDatasource", ["an error"] - myargs = FakeArgs(('ignored_name', myaction), True, 'bogusmode') - cli.status_wrapper('init', myargs, data_d, link_d) + myargs = FakeArgs(("ignored_name", myaction), True, "bogusmode") + cli.status_wrapper("init", myargs, data_d, link_d) # No errors reported in status - status_v1 = load_json(load_file(status_link))['v1'] - self.assertEqual(['an error'], status_v1['init-local']['errors']) - self.assertEqual('SomeDatasource', status_v1['datasource']) + status_v1 = load_json(load_file(status_link))["v1"] + self.assertEqual(["an error"], status_v1["init-local"]["errors"]) + self.assertEqual("SomeDatasource", status_v1["datasource"]) self.assertFalse( - os.path.exists(self.tmp_path('result.json', data_d)), - 'unexpected result.json found') + os.path.exists(self.tmp_path("result.json", data_d)), + "unexpected result.json found", + ) self.assertFalse( - os.path.exists(self.tmp_path('result.json', link_d)), - 'unexpected result.json link found') + os.path.exists(self.tmp_path("result.json", link_d)), + "unexpected result.json link found", + ) def test_no_arguments_shows_usage(self): exit_code = self._call_main() - self.assertIn('usage: cloud-init', self.stderr.getvalue()) + self.assertIn("usage: cloud-init", self.stderr.getvalue()) self.assertEqual(2, exit_code) def test_no_arguments_shows_error_message(self): exit_code = self._call_main() missing_subcommand_message = [ - 'too few arguments', # python2.7 msg - 'the following arguments are required: subcommand' # python3 msg + "too few arguments", # python2.7 msg + "the following arguments are required: subcommand", # python3 msg ] error = self.stderr.getvalue() - matches = ([msg in error for msg in missing_subcommand_message]) + matches = [msg in error for msg in missing_subcommand_message] self.assertTrue( - any(matches), 'Did not find error message for missing subcommand') + any(matches), "Did not find error message for missing subcommand" + ) self.assertEqual(2, exit_code) def test_all_subcommands_represented_in_help(self): """All known subparsers are represented in the cloud-int help doc.""" self._call_main() error = self.stderr.getvalue() - expected_subcommands = ['analyze', 'clean', 'devel', 'dhclient-hook', - 'features', 'init', 'modules', 'single'] + expected_subcommands = [ + "analyze", + "clean", + "devel", + "dhclient-hook", + "features", + "init", + "modules", + "single", + ] for subcommand in expected_subcommands: self.assertIn(subcommand, error) - @mock.patch('cloudinit.cmd.main.status_wrapper') + @mock.patch("cloudinit.cmd.main.status_wrapper") def test_init_subcommand_parser(self, m_status_wrapper): """The subcommand 'init' calls status_wrapper passing init.""" - self._call_main(['cloud-init', 'init']) + self._call_main(["cloud-init", "init"]) (name, parseargs) = m_status_wrapper.call_args_list[0][0] - self.assertEqual('init', name) - self.assertEqual('init', parseargs.subcommand) - self.assertEqual('init', parseargs.action[0]) - self.assertEqual('main_init', parseargs.action[1].__name__) + self.assertEqual("init", name) + self.assertEqual("init", parseargs.subcommand) + self.assertEqual("init", parseargs.action[0]) + self.assertEqual("main_init", parseargs.action[1].__name__) - @mock.patch('cloudinit.cmd.main.status_wrapper') + @mock.patch("cloudinit.cmd.main.status_wrapper") def test_modules_subcommand_parser(self, m_status_wrapper): """The subcommand 'modules' calls status_wrapper passing modules.""" - self._call_main(['cloud-init', 'modules']) + self._call_main(["cloud-init", "modules"]) (name, parseargs) = m_status_wrapper.call_args_list[0][0] - self.assertEqual('modules', name) - self.assertEqual('modules', parseargs.subcommand) - self.assertEqual('modules', parseargs.action[0]) - self.assertEqual('main_modules', parseargs.action[1].__name__) + self.assertEqual("modules", name) + self.assertEqual("modules", parseargs.subcommand) + self.assertEqual("modules", parseargs.action[0]) + self.assertEqual("main_modules", parseargs.action[1].__name__) def test_conditional_subcommands_from_entry_point_sys_argv(self): """Subcommands from entry-point are properly parsed from sys.argv.""" @@ -151,14 +164,22 @@ class TestCLI(test_helpers.FilesystemMockingTestCase): self.patchStdoutAndStderr(stdout=stdout) expected_errors = [ - 'usage: cloud-init analyze', 'usage: cloud-init clean', - 'usage: cloud-init collect-logs', 'usage: cloud-init devel', - 'usage: cloud-init status'] + "usage: cloud-init analyze", + "usage: cloud-init clean", + "usage: cloud-init collect-logs", + "usage: cloud-init devel", + "usage: cloud-init status", + ] conditional_subcommands = [ - 'analyze', 'clean', 'collect-logs', 'devel', 'status'] + "analyze", + "clean", + "collect-logs", + "devel", + "status", + ] # The cloud-init entrypoint calls main without passing sys_argv for subcommand in conditional_subcommands: - with mock.patch('sys.argv', ['cloud-init', subcommand, '-h']): + with mock.patch("sys.argv", ["cloud-init", subcommand, "-h"]): try: cli.main() except SystemExit as e: @@ -168,9 +189,9 @@ class TestCLI(test_helpers.FilesystemMockingTestCase): def test_analyze_subcommand_parser(self): """The subcommand cloud-init analyze calls the correct subparser.""" - self._call_main(['cloud-init', 'analyze']) + self._call_main(["cloud-init", "analyze"]) # These subcommands only valid for cloud-init analyze script - expected_subcommands = ['blame', 'show', 'dump'] + expected_subcommands = ["blame", "show", "dump"] error = self.stderr.getvalue() for subcommand in expected_subcommands: self.assertIn(subcommand, error) @@ -180,94 +201,177 @@ class TestCLI(test_helpers.FilesystemMockingTestCase): # Provide -h param to collect-logs to avoid having to mock behavior. stdout = io.StringIO() self.patchStdoutAndStderr(stdout=stdout) - self._call_main(['cloud-init', 'collect-logs', '-h']) - self.assertIn('usage: cloud-init collect-log', stdout.getvalue()) + self._call_main(["cloud-init", "collect-logs", "-h"]) + self.assertIn("usage: cloud-init collect-log", stdout.getvalue()) def test_clean_subcommand_parser(self): """The subcommand cloud-init clean calls the subparser.""" # Provide -h param to clean to avoid having to mock behavior. stdout = io.StringIO() self.patchStdoutAndStderr(stdout=stdout) - self._call_main(['cloud-init', 'clean', '-h']) - self.assertIn('usage: cloud-init clean', stdout.getvalue()) + self._call_main(["cloud-init", "clean", "-h"]) + self.assertIn("usage: cloud-init clean", stdout.getvalue()) def test_status_subcommand_parser(self): """The subcommand cloud-init status calls the subparser.""" # Provide -h param to clean to avoid having to mock behavior. stdout = io.StringIO() self.patchStdoutAndStderr(stdout=stdout) - self._call_main(['cloud-init', 'status', '-h']) - self.assertIn('usage: cloud-init status', stdout.getvalue()) + self._call_main(["cloud-init", "status", "-h"]) + self.assertIn("usage: cloud-init status", stdout.getvalue()) def test_devel_subcommand_parser(self): """The subcommand cloud-init devel calls the correct subparser.""" - self._call_main(['cloud-init', 'devel']) + self._call_main(["cloud-init", "devel"]) # These subcommands only valid for cloud-init schema script - expected_subcommands = ['schema'] + expected_subcommands = ["schema"] error = self.stderr.getvalue() for subcommand in expected_subcommands: self.assertIn(subcommand, error) def test_wb_devel_schema_subcommand_parser(self): """The subcommand cloud-init schema calls the correct subparser.""" - exit_code = self._call_main(['cloud-init', 'devel', 'schema']) + exit_code = self._call_main(["cloud-init", "devel", "schema"]) self.assertEqual(1, exit_code) # Known whitebox output from schema subcommand self.assertEqual( - 'Expected one of --config-file, --system or --docs arguments\n', - self.stderr.getvalue()) + "Error:\n" + "Expected one of --config-file, --system or --docs arguments\n", + self.stderr.getvalue(), + ) + + def test_wb_devel_schema_subcommand_doc_all_spot_check(self): + """Validate that doc content has correct values from known examples. + + Ensure that schema doc is returned + """ - def test_wb_devel_schema_subcommand_doc_content(self): - """Validate that doc content is sane from known examples.""" + # Note: patchStdoutAndStderr() is convenient for reducing boilerplate, + # but inspecting the code for debugging is not ideal + # contextlib.redirect_stdout() provides similar behavior as a context + # manager stdout = io.StringIO() - self.patchStdoutAndStderr(stdout=stdout) - self._call_main(['cloud-init', 'devel', 'schema', '--docs', 'all']) - expected_doc_sections = [ - '**Supported distros:** all', - '**Supported distros:** alpine, centos, debian, fedora', - '**Config schema**:\n **resize_rootfs:** (true/false/noblock)', - '**Examples**::\n\n runcmd:\n - [ ls, -l, / ]\n' - ] + with contextlib.redirect_stdout(stdout): + self._call_main(["cloud-init", "devel", "schema", "--docs", "all"]) + expected_doc_sections = [ + "**Supported distros:** all", + "**Supported distros:** almalinux, alpine, centos, " + "cloudlinux, debian, eurolinux, fedora, miraclelinux, " + "openEuler, opensuse, photon, rhel, rocky, sles, ubuntu, " + "virtuozzo", + "**Config schema**:\n **resize_rootfs:** " + "(true/false/noblock)", + "**Examples**::\n\n runcmd:\n - [ ls, -l, / ]\n", + ] stdout = stdout.getvalue() for expected in expected_doc_sections: self.assertIn(expected, stdout) - @mock.patch('cloudinit.cmd.main.main_single') + def test_wb_devel_schema_subcommand_single_spot_check(self): + """Validate that doc content has correct values from known example. + + Validate 'all' arg + """ + + # Note: patchStdoutAndStderr() is convenient for reducing boilerplate, + # but inspecting the code for debugging is not ideal + # contextlib.redirect_stdout() provides similar behavior as a context + # manager + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + self._call_main( + ["cloud-init", "devel", "schema", "--docs", "cc_runcmd"] + ) + expected_doc_sections = [ + "Runcmd\n------\n**Summary:** Run arbitrary commands" + ] + stdout = stdout.getvalue() + for expected in expected_doc_sections: + self.assertIn(expected, stdout) + + def test_wb_devel_schema_subcommand_multiple_spot_check(self): + """Validate that doc content has correct values from known example. + + Validate single arg + """ + + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + self._call_main( + [ + "cloud-init", + "devel", + "schema", + "--docs", + "cc_runcmd", + "cc_resizefs", + ] + ) + expected_doc_sections = [ + "Runcmd\n------\n**Summary:** Run arbitrary commands", + "Resizefs\n--------\n**Summary:** Resize filesystem", + ] + stdout = stdout.getvalue() + for expected in expected_doc_sections: + self.assertIn(expected, stdout) + + def test_wb_devel_schema_subcommand_bad_arg_fails(self): + """Validate that doc content has correct values from known example. + + Validate multiple args + """ + + # Note: patchStdoutAndStderr() is convenient for reducing boilerplate, + # but inspecting the code for debugging is not ideal + # contextlib.redirect_stdout() provides similar behavior as a context + # manager + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + self._call_main( + ["cloud-init", "devel", "schema", "--docs", "garbage_value"] + ) + expected_doc_sections = ["Invalid --docs value"] + stderr = stderr.getvalue() + for expected in expected_doc_sections: + self.assertIn(expected, stderr) + + @mock.patch("cloudinit.cmd.main.main_single") def test_single_subcommand(self, m_main_single): """The subcommand 'single' calls main_single with valid args.""" - self._call_main(['cloud-init', 'single', '--name', 'cc_ntp']) + self._call_main(["cloud-init", "single", "--name", "cc_ntp"]) (name, parseargs) = m_main_single.call_args_list[0][0] - self.assertEqual('single', name) - self.assertEqual('single', parseargs.subcommand) - self.assertEqual('single', parseargs.action[0]) + self.assertEqual("single", name) + self.assertEqual("single", parseargs.subcommand) + self.assertEqual("single", parseargs.action[0]) self.assertFalse(parseargs.debug) self.assertFalse(parseargs.force) self.assertIsNone(parseargs.frequency) - self.assertEqual('cc_ntp', parseargs.name) + self.assertEqual("cc_ntp", parseargs.name) self.assertFalse(parseargs.report) - @mock.patch('cloudinit.cmd.main.dhclient_hook.handle_args') + @mock.patch("cloudinit.cmd.main.dhclient_hook.handle_args") def test_dhclient_hook_subcommand(self, m_handle_args): """The subcommand 'dhclient-hook' calls dhclient_hook with args.""" - self._call_main(['cloud-init', 'dhclient-hook', 'up', 'eth0']) + self._call_main(["cloud-init", "dhclient-hook", "up", "eth0"]) (name, parseargs) = m_handle_args.call_args_list[0][0] - self.assertEqual('dhclient-hook', name) - self.assertEqual('dhclient-hook', parseargs.subcommand) - self.assertEqual('dhclient-hook', parseargs.action[0]) + self.assertEqual("dhclient-hook", name) + self.assertEqual("dhclient-hook", parseargs.subcommand) + self.assertEqual("dhclient-hook", parseargs.action[0]) self.assertFalse(parseargs.debug) self.assertFalse(parseargs.force) - self.assertEqual('up', parseargs.event) - self.assertEqual('eth0', parseargs.interface) + self.assertEqual("up", parseargs.event) + self.assertEqual("eth0", parseargs.interface) - @mock.patch('cloudinit.cmd.main.main_features') + @mock.patch("cloudinit.cmd.main.main_features") def test_features_hook_subcommand(self, m_features): """The subcommand 'features' calls main_features with args.""" - self._call_main(['cloud-init', 'features']) + self._call_main(["cloud-init", "features"]) (name, parseargs) = m_features.call_args_list[0][0] - self.assertEqual('features', name) - self.assertEqual('features', parseargs.subcommand) - self.assertEqual('features', parseargs.action[0]) + self.assertEqual("features", name) + self.assertEqual("features", parseargs.subcommand) + self.assertEqual("features", parseargs.action[0]) self.assertFalse(parseargs.debug) self.assertFalse(parseargs.force) + # : ts=4 expandtab |