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

import os
from datetime import datetime
from io import StringIO

from cloudinit.cmd.devel import logs
from cloudinit.sources import INSTANCE_JSON_SENSITIVE_FILE
from cloudinit.subp import subp
from cloudinit.util import ensure_dir, load_file, write_file
from tests.unittests.helpers import (
    FilesystemMockingTestCase,
    mock,
    wrap_and_call,
)


@mock.patch("cloudinit.cmd.devel.logs.os.getuid")
class TestCollectLogs(FilesystemMockingTestCase):
    def setUp(self):
        super(TestCollectLogs, self).setUp()
        self.new_root = self.tmp_dir()
        self.run_dir = self.tmp_path("run", self.new_root)

    def test_collect_logs_with_userdata_requires_root_user(self, m_getuid):
        """collect-logs errors when non-root user collects userdata ."""
        m_getuid.return_value = 100  # non-root
        output_tarfile = self.tmp_path("logs.tgz")
        with mock.patch("sys.stderr", new_callable=StringIO) as m_stderr:
            self.assertEqual(
                1, logs.collect_logs(output_tarfile, include_userdata=True)
            )
        self.assertEqual(
            "To include userdata, root user is required."
            " Try sudo cloud-init collect-logs\n",
            m_stderr.getvalue(),
        )

    def test_collect_logs_creates_tarfile(self, m_getuid):
        """collect-logs creates a tarfile with all related cloud-init info."""
        m_getuid.return_value = 100
        log1 = self.tmp_path("cloud-init.log", self.new_root)
        write_file(log1, "cloud-init-log")
        log2 = self.tmp_path("cloud-init-output.log", self.new_root)
        write_file(log2, "cloud-init-output-log")
        ensure_dir(self.run_dir)
        write_file(self.tmp_path("results.json", self.run_dir), "results")
        write_file(
            self.tmp_path(INSTANCE_JSON_SENSITIVE_FILE, self.run_dir),
            "sensitive",
        )
        output_tarfile = self.tmp_path("logs.tgz")

        date = datetime.utcnow().date().strftime("%Y-%m-%d")
        date_logdir = "cloud-init-logs-{0}".format(date)

        version_out = "/usr/bin/cloud-init 18.2fake\n"
        expected_subp = {
            (
                "dpkg-query",
                "--show",
                "-f=${Version}\n",
                "cloud-init",
            ): "0.7fake\n",
            ("cloud-init", "--version"): version_out,
            ("dmesg",): "dmesg-out\n",
            ("journalctl", "--boot=0", "-o", "short-precise"): "journal-out\n",
            ("tar", "czvf", output_tarfile, date_logdir): "",
        }

        def fake_subp(cmd):
            cmd_tuple = tuple(cmd)
            if cmd_tuple not in expected_subp:
                raise AssertionError(
                    "Unexpected command provided to subp: {0}".format(cmd)
                )
            if cmd == ["tar", "czvf", output_tarfile, date_logdir]:
                subp(cmd)  # Pass through tar cmd so we can check output
            return expected_subp[cmd_tuple], ""

        fake_stderr = mock.MagicMock()

        wrap_and_call(
            "cloudinit.cmd.devel.logs",
            {
                "subp": {"side_effect": fake_subp},
                "sys.stderr": {"new": fake_stderr},
                "CLOUDINIT_LOGS": {"new": [log1, log2]},
                "CLOUDINIT_RUN_DIR": {"new": self.run_dir},
            },
            logs.collect_logs,
            output_tarfile,
            include_userdata=False,
        )
        # unpack the tarfile and check file contents
        subp(["tar", "zxvf", output_tarfile, "-C", self.new_root])
        out_logdir = self.tmp_path(date_logdir, self.new_root)
        self.assertFalse(
            os.path.exists(
                os.path.join(
                    out_logdir,
                    "run",
                    "cloud-init",
                    INSTANCE_JSON_SENSITIVE_FILE,
                )
            ),
            "Unexpected file found: %s" % INSTANCE_JSON_SENSITIVE_FILE,
        )
        self.assertEqual(
            "0.7fake\n", load_file(os.path.join(out_logdir, "dpkg-version"))
        )
        self.assertEqual(
            version_out, load_file(os.path.join(out_logdir, "version"))
        )
        self.assertEqual(
            "cloud-init-log",
            load_file(os.path.join(out_logdir, "cloud-init.log")),
        )
        self.assertEqual(
            "cloud-init-output-log",
            load_file(os.path.join(out_logdir, "cloud-init-output.log")),
        )
        self.assertEqual(
            "dmesg-out\n", load_file(os.path.join(out_logdir, "dmesg.txt"))
        )
        self.assertEqual(
            "journal-out\n", load_file(os.path.join(out_logdir, "journal.txt"))
        )
        self.assertEqual(
            "results",
            load_file(
                os.path.join(out_logdir, "run", "cloud-init", "results.json")
            ),
        )
        fake_stderr.write.assert_any_call("Wrote %s\n" % output_tarfile)

    def test_collect_logs_includes_optional_userdata(self, m_getuid):
        """collect-logs include userdata when --include-userdata is set."""
        m_getuid.return_value = 0
        log1 = self.tmp_path("cloud-init.log", self.new_root)
        write_file(log1, "cloud-init-log")
        log2 = self.tmp_path("cloud-init-output.log", self.new_root)
        write_file(log2, "cloud-init-output-log")
        userdata = self.tmp_path("user-data.txt", self.new_root)
        write_file(userdata, "user-data")
        ensure_dir(self.run_dir)
        write_file(self.tmp_path("results.json", self.run_dir), "results")
        write_file(
            self.tmp_path(INSTANCE_JSON_SENSITIVE_FILE, self.run_dir),
            "sensitive",
        )
        output_tarfile = self.tmp_path("logs.tgz")

        date = datetime.utcnow().date().strftime("%Y-%m-%d")
        date_logdir = "cloud-init-logs-{0}".format(date)

        version_out = "/usr/bin/cloud-init 18.2fake\n"
        expected_subp = {
            (
                "dpkg-query",
                "--show",
                "-f=${Version}\n",
                "cloud-init",
            ): "0.7fake",
            ("cloud-init", "--version"): version_out,
            ("dmesg",): "dmesg-out\n",
            ("journalctl", "--boot=0", "-o", "short-precise"): "journal-out\n",
            ("tar", "czvf", output_tarfile, date_logdir): "",
        }

        def fake_subp(cmd):
            cmd_tuple = tuple(cmd)
            if cmd_tuple not in expected_subp:
                raise AssertionError(
                    "Unexpected command provided to subp: {0}".format(cmd)
                )
            if cmd == ["tar", "czvf", output_tarfile, date_logdir]:
                subp(cmd)  # Pass through tar cmd so we can check output
            return expected_subp[cmd_tuple], ""

        fake_stderr = mock.MagicMock()

        wrap_and_call(
            "cloudinit.cmd.devel.logs",
            {
                "subp": {"side_effect": fake_subp},
                "sys.stderr": {"new": fake_stderr},
                "CLOUDINIT_LOGS": {"new": [log1, log2]},
                "CLOUDINIT_RUN_DIR": {"new": self.run_dir},
                "USER_DATA_FILE": {"new": userdata},
            },
            logs.collect_logs,
            output_tarfile,
            include_userdata=True,
        )
        # unpack the tarfile and check file contents
        subp(["tar", "zxvf", output_tarfile, "-C", self.new_root])
        out_logdir = self.tmp_path(date_logdir, self.new_root)
        self.assertEqual(
            "user-data", load_file(os.path.join(out_logdir, "user-data.txt"))
        )
        self.assertEqual(
            "sensitive",
            load_file(
                os.path.join(
                    out_logdir,
                    "run",
                    "cloud-init",
                    INSTANCE_JSON_SENSITIVE_FILE,
                )
            ),
        )
        fake_stderr.write.assert_any_call("Wrote %s\n" % output_tarfile)
