T2117: Cloud-init updated to 22.1

Merged with 22.1 tag from the upstream Cloud-init repository.

Our modules were slightly modified for compatibility with the new
version.

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