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

import contextlib
import io
import json
import logging
import os
import sys
from collections import namedtuple

import pytest

from cloudinit import helpers
from cloudinit.cmd import main as cli
from tests.unittests import helpers as test_helpers

mock = test_helpers.mock

M_PATH = "cloudinit.cmd.main."
Tmpdir = namedtuple("Tmpdir", ["tmpdir", "link_d", "data_d"])
FakeArgs = namedtuple("FakeArgs", ["action", "local", "mode"])


@pytest.fixture(autouse=True, scope="module")
def disable_setup_logging():
    # setup_basic_logging can change the logging level to WARNING, so
    # ensure it is always mocked
    with mock.patch(f"{M_PATH}loggers.setup_basic_logging", autospec=True):
        yield


@pytest.fixture()
def mock_status_wrapper(mocker, tmpdir):
    link_d = os.path.join(tmpdir, "link")
    data_d = os.path.join(tmpdir, "data")
    mocker.patch(
        "cloudinit.cmd.main.read_cfg_paths",
        return_value=mock.Mock(get_cpath=lambda _: data_d),
    )
    mocker.patch("cloudinit.cmd.main.os.path.normpath", return_value=link_d)
    yield Tmpdir(tmpdir, link_d, data_d)


class TestCLI:
    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

    @pytest.mark.parametrize(
        "action,name,match",
        [
            pytest.param(
                "doesnotmatter",
                "init1",
                "^unknown name: init1$",
                id="invalid_name",
            ),
            pytest.param(
                "modules_name",
                "modules",
                "^Invalid cloud init mode specified 'modules-bogusmode'$",
                id="invalid_modes",
            ),
        ],
    )
    def test_status_wrapper_errors(
        self, action, name, match, caplog, mock_status_wrapper
    ):
        my_action = mock.Mock()

        myargs = FakeArgs((action, my_action), False, "bogusmode")
        with pytest.raises(ValueError, match=match):
            cli.status_wrapper(name, myargs)
        assert [] == my_action.call_args_list

    @mock.patch("cloudinit.cmd.main.atomic_helper.write_json")
    def test_status_wrapper_init_local_writes_fresh_status_info(
        self,
        m_json,
        mock_status_wrapper,
    ):
        """When running in init-local mode, status_wrapper writes status.json.

        Old status and results artifacts are also removed.
        """
        data_d = mock_status_wrapper.data_d
        link_d = mock_status_wrapper.link_d
        # Write old artifacts which will be removed or updated.
        for _dir in data_d, link_d:
            test_helpers.populate_dir(
                str(_dir), {"status.json": "old", "result.json": "old"}
            )

        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)
        # No errors reported in status
        status_v1 = m_json.call_args_list[1][0][1]["v1"]
        assert status_v1.keys() == {
            "datasource",
            "init-local",
            "init",
            "modules-config",
            "modules-final",
            "stage",
        }
        assert ["an error"] == status_v1["init-local"]["errors"]
        assert "SomeDatasource" == status_v1["datasource"]
        assert False is os.path.exists(
            data_d.join("result.json")
        ), "unexpected result.json found"
        assert False is os.path.exists(
            link_d.join("result.json")
        ), "unexpected result.json link found"

    @mock.patch("cloudinit.cmd.main.atomic_helper.write_json")
    def test_status_wrapper_init_local_honor_cloud_dir(
        self, m_json, mocker, mock_status_wrapper
    ):
        """When running in init-local mode, status_wrapper honors cloud_dir."""
        cloud_dir = mock_status_wrapper.tmpdir.join("cloud")
        paths = helpers.Paths({"cloud_dir": str(cloud_dir)})
        mocker.patch(
            "cloudinit.config.schema.read_cfg_paths", return_value=paths
        )
        data_d = mock_status_wrapper.data_d
        link_d = mock_status_wrapper.link_d

        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)  # No explicit data_d

        # Access cloud_dir directly
        status_v1 = m_json.call_args_list[1][0][1]["v1"]
        assert ["an_error"] == status_v1["init-local"]["errors"]
        assert "SomeDatasource" == status_v1["datasource"]
        assert False is os.path.exists(
            data_d.join("result.json")
        ), "unexpected result.json found"
        assert False is os.path.exists(
            link_d.join("result.json")
        ), "unexpected result.json link found"

    def test_no_arguments_shows_usage(self, capsys):
        exit_code = self._call_main()
        _out, err = capsys.readouterr()
        assert "usage: cloud-init" in err
        assert 2 == exit_code

    def test_no_arguments_shows_error_message(self, capsys):
        exit_code = self._call_main()
        missing_subcommand_message = "a subcommand is required"
        _out, err = capsys.readouterr()
        assert (
            missing_subcommand_message in err
        ), "Did not find error message for missing subcommand"
        assert 2 == exit_code

    def test_all_subcommands_represented_in_help(self, capsys):
        """All known subparsers are represented in the cloud-int help doc."""
        self._call_main()
        _out, err = capsys.readouterr()
        expected_subcommands = [
            "analyze",
            "clean",
            "devel",
            "features",
            "init",
            "modules",
            "single",
            "schema",
        ]
        for subcommand in expected_subcommands:
            assert subcommand in err

    @pytest.mark.parametrize(
        "subcommand,log_to_stderr,mocks",
        (
            ("init", False, [mock.patch("cloudinit.cmd.main.status_wrapper")]),
            (
                "modules",
                False,
                [mock.patch("cloudinit.cmd.main.status_wrapper")],
            ),
            (
                "schema",
                True,
                [
                    mock.patch(
                        "cloudinit.stages.Init._read_cfg", return_value={}
                    ),
                    mock.patch("cloudinit.config.schema.handle_schema_args"),
                ],
            ),
        ),
    )
    @mock.patch("cloudinit.cmd.main.loggers.setup_basic_logging")
    def test_subcommands_log_to_stderr_via_setup_basic_logging(
        self, setup_basic_logging, subcommand, log_to_stderr, mocks
    ):
        """setup_basic_logging is called for modules to use stderr

        Subcommands with exception of 'init'  and 'modules' use
        setup_basic_logging to direct logged errors to stderr.
        """
        with contextlib.ExitStack() as mockstack:
            for mymock in mocks:
                mockstack.enter_context(mymock)
            self._call_main(["cloud-init", subcommand])
        if log_to_stderr:
            setup_basic_logging.assert_called_once_with(logging.WARNING)
        else:
            setup_basic_logging.assert_not_called()

    @pytest.mark.parametrize("subcommand", ["init", "modules"])
    @mock.patch("cloudinit.cmd.main.status_wrapper")
    def test_modules_subcommand_parser(self, m_status_wrapper, subcommand):
        """The subcommand 'subcommand' calls status_wrapper passing modules."""
        self._call_main(["cloud-init", subcommand])
        (name, parseargs) = m_status_wrapper.call_args_list[0][0]
        assert subcommand == name
        assert subcommand == parseargs.subcommand
        assert subcommand == parseargs.action[0]
        assert f"main_{subcommand}" == parseargs.action[1].__name__

    @pytest.mark.parametrize(
        "subcommand",
        [
            "analyze",
            "clean",
            "collect-logs",
            "devel",
            "status",
            "schema",
        ],
    )
    def test_conditional_subcommands_from_entry_point_sys_argv(
        self,
        subcommand,
        capsys,
        mock_status_wrapper,
    ):
        """Subcommands from entry-point are properly parsed from sys.argv."""
        expected_error = f"usage: cloud-init {subcommand}"
        # The cloud-init entrypoint calls main without passing sys_argv
        with mock.patch("sys.argv", ["cloud-init", subcommand, "-h"]):
            try:
                cli.main()
            except SystemExit as e:
                assert 0 == e.code  # exit 2 on proper -h usage
        out, _err = capsys.readouterr()
        assert expected_error in out

    @pytest.mark.parametrize(
        "subcommand",
        [
            "clean",
            "collect-logs",
            "status",
        ],
    )
    def test_subcommand_parser(self, subcommand, mock_status_wrapper):
        """cloud-init `subcommand` calls its subparser."""
        # Provide -h param to `subcommand` to avoid having to mock behavior.
        out = io.StringIO()
        with contextlib.redirect_stdout(out):
            self._call_main(["cloud-init", subcommand, "-h"])
        assert f"usage: cloud-init {subcommand}" in out.getvalue()

    @pytest.mark.parametrize(
        "subcommand",
        [
            "clean",
            "collect-logs",
            "status",
        ],
    )
    def test_subcommand_parser_shows_usage(self, subcommand, capsys):
        """cloud-init `subcommand` shows usage on error."""
        # Provide --invalid-arg to `subcommand` to trigger error.
        exit_code = self._call_main(["cloud-init", subcommand, "--invalid"])
        _out, err = capsys.readouterr()
        assert f"usage: cloud-init {subcommand}" in err
        assert 2 == exit_code

    @pytest.mark.parametrize(
        "args,expected_subcommands",
        [
            ([], ["schema"]),
            (["analyze"], ["blame", "show", "dump"]),
        ],
    )
    def test_subcommand_parser_multi_arg(
        self, args, expected_subcommands, capsys
    ):
        """The subcommand cloud-init schema calls the correct subparser."""
        self._call_main(["cloud-init"] + args)
        _out, err = capsys.readouterr()
        for subcommand in expected_subcommands:
            assert subcommand in err

    @mock.patch("cloudinit.stages.Init._read_cfg", return_value={})
    def test_wb_schema_subcommand_parser(self, m_read_cfg, capsys):
        """The subcommand cloud-init schema calls the correct subparser."""
        exit_code = self._call_main(["cloud-init", "schema"])
        _out, err = capsys.readouterr()
        assert 1 == exit_code
        # Known whitebox output from schema subcommand
        assert (
            "Error:\n"
            "Expected one of --config-file or --system arguments\n" in err
        )

    @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]
        assert "single" == name
        assert "single" == parseargs.subcommand
        assert "single" == parseargs.action[0]
        assert False is parseargs.debug
        assert False is parseargs.force
        assert None is parseargs.frequency
        assert "cc_ntp" == parseargs.name
        assert False is parseargs.report

    @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]
        assert "features" == name
        assert "features" == parseargs.subcommand
        assert "features" == parseargs.action[0]
        assert False is parseargs.debug
        assert False is parseargs.force

    def test_all_stages_with_tty(self, mocker):
        """Ensure all stages get called when using a tty."""
        mocker.patch("cloudinit.cmd.main.os.isatty", return_value=True)
        mocker.patch("cloudinit.cmd.main.sys.stdin.fileno")
        mocker.patch("cloudinit.cmd.main.socket.sd_notify")
        mocker.patch("cloudinit.cmd.main.socket.os.makedirs")
        mocker.patch("cloudinit.cmd.main.socket.os.remove")
        mocker.patch("cloudinit.cmd.main.socket.socket")
        m_sub_main = mocker.patch(
            "cloudinit.cmd.main.sub_main", return_value=0
        )

        self._call_main(["cloud-init", "--all-stages"])
        assert m_sub_main.call_count == 4
        assert m_sub_main.call_args_list[0][0][0].subcommand == "init"
        assert m_sub_main.call_args_list[0][0][0].local is True

        assert m_sub_main.call_args_list[1][0][0].subcommand == "init"
        assert m_sub_main.call_args_list[1][0][0].local is False

        assert m_sub_main.call_args_list[2][0][0].subcommand == "modules"
        assert m_sub_main.call_args_list[2][0][0].mode == "config"

        assert m_sub_main.call_args_list[3][0][0].subcommand == "modules"
        assert m_sub_main.call_args_list[3][0][0].mode == "final"


class TestSignalHandling:
    @mock.patch("cloudinit.cmd.main.atomic_helper.write_json")
    def test_status_wrapper_signal_sys_exit(
        self,
        m_json,
        mocker,
        mock_status_wrapper,
    ):
        """make sure that when sys.exit(N) is called, the correct code is
        returned
        """
        for code in [1, 2, 3, 4]:
            rc = cli.status_wrapper(
                "init",
                FakeArgs(
                    (
                        None,
                        # silence pylint false positive
                        # https://github.com/pylint-dev/pylint/issues/9557
                        lambda *_: sys.exit(code),  # pylint: disable=W0640
                    ),
                    False,
                    "bogusmode",
                ),
            )
            assert 1 == rc

            # assert that the status shows errors
            assert (
                f"sys.exit({code}) called"
                in m_json.call_args[0][1]["v1"]["init"]["errors"]
            )

    @mock.patch("cloudinit.cmd.main.atomic_helper.write_json")
    def test_status_wrapper_no_signal_sys_exit(
        self,
        m_json,
        mock_status_wrapper,
    ):
        """if sys.exit(0) is called, make sure that cloud-init doesn't log a
        warning"""
        # call status_wrapper() with the required args
        rc = cli.status_wrapper(
            "init",
            FakeArgs(
                (
                    None,
                    lambda *_: sys.exit(0),
                ),
                False,
                "bogusmode",
            ),
        )
        assert 0 == rc
        assert not m_json.call_args[0][1]["v1"]["init"]["errors"]

    @mock.patch("cloudinit.cmd.main.atomic_helper.write_json")
    def test_status_wrapper_signal_warnings(
        self,
        m_json,
        mock_status_wrapper,
    ):
        """If a stage is started and status.json already has a start time but
        no end time for that stage, this is an unknown state - make sure that
        a warning is logged.
        """

        # Write a status.json to the mocked temporary directory
        for dir in mock_status_wrapper.data_d, mock_status_wrapper.link_d:
            test_helpers.populate_dir(
                str(dir),
                {
                    "status.json": json.dumps(
                        {
                            "v1": {
                                "stage": "init",
                                "datasource": (
                                    "DataSourceNoCloud "
                                    "[seed=/var/.../seed/nocloud-net]"
                                    "[dsmode=net]"
                                ),
                                "init": {
                                    "errors": [],
                                    "recoverable_errors": {},
                                    "start": 124.567,
                                    "finished": None,
                                },
                                "init-local": {
                                    "errors": [],
                                    "recoverable_errors": {},
                                    "start": 100.0,
                                    "finished": 100.00001,
                                },
                                "modules-config": {
                                    "errors": [],
                                    "recoverable_errors": {},
                                    "start": None,
                                    "finished": None,
                                },
                                "modules-final": {
                                    "errors": [],
                                    "recoverable_errors": {},
                                    "start": None,
                                    "finished": None,
                                },
                            }
                        }
                    )
                },
            )
        # call status_wrapper() with the required args
        cli.status_wrapper(
            "init",
            FakeArgs(
                (
                    None,
                    lambda *_: ("SomeDataSource", []),
                ),
                False,
                "bogusmode",
            ),
        )

        # assert that the status shows recoverable errors
        assert (
            "Unexpected start time found for Network Stage. "
            "Was this stage restarted?"
            in m_json.call_args[0][1]["v1"]["init"]["recoverable_errors"][
                "WARNING"
            ]
        )