path: root/tests/unittests/test_cli.py

# This file is part of cloud-init. See LICENSE file for license information.

import contextlib
import io
import os
from collections import namedtuple

from cloudinit.cmd import main as cli
from cloudinit.util import load_file, load_json
from tests.unittests import helpers as test_helpers

mock = test_helpers.mock


class TestCLI(test_helpers.FilesystemMockingTestCase):

    with_logs = True

    def setUp(self):
        super(TestCLI, self).setUp()
        self.stderr = io.StringIO()
        self.patchStdoutAndStderr(stderr=self.stderr)

    def _call_main(self, sysv_args=None):
        if not sysv_args:
            sysv_args = ["cloud-init"]
        try:
            return cli.main(sysv_args=sysv_args)
        except SystemExit as e:
            return e.code

    def test_status_wrapper_errors_on_invalid_name(self):
        """status_wrapper will error when the name parameter is not valid.

        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"])

        def myaction():
            raise Exception("Should not call myaction")

        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())

    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"])

        def myaction():
            raise Exception("Should not call myaction")

        myargs = FakeArgs(("modules_name", myaction), False, "bogusmode")
        with self.assertRaises(ValueError) as cm:
            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())

    def test_status_wrapper_init_local_writes_fresh_status_info(self):
        """When running in init-local mode, status_wrapper writes status.json.

        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)
        # 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"}
            )

        FakeArgs = namedtuple("FakeArgs", ["action", "local", "mode"])

        def myaction(name, args):
            # Return an error to watch status capture them
            return "SomeDatasource", ["an error"]

        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"])
        self.assertFalse(
            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",
        )

    def test_no_arguments_shows_usage(self):
        exit_code = self._call_main()
        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
        ]
        error = self.stderr.getvalue()
        matches = [msg in error for msg in missing_subcommand_message]
        self.assertTrue(
            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",
        ]
        for subcommand in expected_subcommands:
            self.assertIn(subcommand, error)

    @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"])
        (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__)

    @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"])
        (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__)

    def test_conditional_subcommands_from_entry_point_sys_argv(self):
        """Subcommands from entry-point are properly parsed from sys.argv."""
        stdout = io.StringIO()
        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",
        ]
        conditional_subcommands = [
            "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"]):
                try:
                    cli.main()
                except SystemExit as e:
                    self.assertEqual(0, e.code)  # exit 2 on proper -h usage
        for error_message in expected_errors:
            self.assertIn(error_message, stdout.getvalue())

    def test_analyze_subcommand_parser(self):
        """The subcommand cloud-init analyze calls the correct subparser."""
        self._call_main(["cloud-init", "analyze"])
        # These subcommands only valid for cloud-init analyze script
        expected_subcommands = ["blame", "show", "dump"]
        error = self.stderr.getvalue()
        for subcommand in expected_subcommands:
            self.assertIn(subcommand, error)

    def test_collect_logs_subcommand_parser(self):
        """The subcommand cloud-init collect-logs calls the subparser."""
        # 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())

    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())

    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())

    def test_devel_subcommand_parser(self):
        """The subcommand cloud-init devel calls the correct subparser."""
        self._call_main(["cloud-init", "devel"])
        # These subcommands only valid for cloud-init schema script
        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"])
        self.assertEqual(1, exit_code)
        # Known whitebox output from schema subcommand
        self.assertEqual(
            "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
        """

        # 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", "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)

    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"])
        (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.assertFalse(parseargs.debug)
        self.assertFalse(parseargs.force)
        self.assertIsNone(parseargs.frequency)
        self.assertEqual("cc_ntp", parseargs.name)
        self.assertFalse(parseargs.report)

    @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"])
        (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.assertFalse(parseargs.debug)
        self.assertFalse(parseargs.force)
        self.assertEqual("up", parseargs.event)
        self.assertEqual("eth0", parseargs.interface)

    @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"])
        (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.assertFalse(parseargs.debug)
        self.assertFalse(parseargs.force)


# : ts=4 expandtab