# This file is part of cloud-init. See LICENSE file for license information.
# pylint: disable=attribute-defined-outside-init

import base64
import copy
import json
from textwrap import dedent
from unittest import mock

import pytest

from cloudinit import util
from cloudinit.sources import DataSourceIBMCloud as ibm
from tests.unittests import helpers as test_helpers

D_PATH = "cloudinit.sources.DataSourceIBMCloud."


@mock.patch(D_PATH + "_is_xen", return_value=True)
@mock.patch(D_PATH + "_is_ibm_provisioning")
@mock.patch(D_PATH + "util.blkid")
class TestGetIBMPlatform:
    """Test the get_ibm_platform helper."""

    blkid_base = {
        "/dev/xvda1": {
            "DEVNAME": "/dev/xvda1",
            "LABEL": "cloudimg-bootfs",
            "TYPE": "ext3",
        },
        "/dev/xvda2": {
            "DEVNAME": "/dev/xvda2",
            "LABEL": "cloudimg-rootfs",
            "TYPE": "ext4",
        },
    }

    blkid_metadata_disk = {
        "/dev/xvdh1": {
            "DEVNAME": "/dev/xvdh1",
            "LABEL": "METADATA",
            "TYPE": "vfat",
            "SEC_TYPE": "msdos",
            "UUID": "681B-8C5D",
            "PARTUUID": "3d631e09-01",
        },
    }

    blkid_oscode_disk = {
        "/dev/xvdh": {
            "DEVNAME": "/dev/xvdh",
            "LABEL": "config-2",
            "TYPE": "vfat",
            "SEC_TYPE": "msdos",
            "UUID": ibm.IBM_CONFIG_UUID,
        }
    }

    @pytest.fixture(autouse=True)
    def fixtures(self):
        self.blkid_metadata = copy.deepcopy(self.blkid_base)
        self.blkid_metadata.update(copy.deepcopy(self.blkid_metadata_disk))

        self.blkid_oscode = copy.deepcopy(self.blkid_base)
        self.blkid_oscode.update(copy.deepcopy(self.blkid_oscode_disk))

    def test_id_template_live_metadata(self, m_blkid, m_is_prov, _m_xen):
        """identify TEMPLATE_LIVE_METADATA."""
        m_blkid.return_value = self.blkid_metadata
        m_is_prov.return_value = False
        assert (
            ibm.Platforms.TEMPLATE_LIVE_METADATA,
            "/dev/xvdh1",
        ) == ibm.get_ibm_platform()

    def test_id_template_prov_metadata(self, m_blkid, m_is_prov, _m_xen):
        """identify TEMPLATE_PROVISIONING_METADATA."""
        m_blkid.return_value = self.blkid_metadata
        m_is_prov.return_value = True
        assert (
            ibm.Platforms.TEMPLATE_PROVISIONING_METADATA,
            "/dev/xvdh1",
        ) == ibm.get_ibm_platform()

    def test_id_template_prov_nodata(self, m_blkid, m_is_prov, _m_xen):
        """identify TEMPLATE_PROVISIONING_NODATA."""
        m_blkid.return_value = self.blkid_base
        m_is_prov.return_value = True
        assert (
            ibm.Platforms.TEMPLATE_PROVISIONING_NODATA,
            None,
        ) == ibm.get_ibm_platform()

    def test_id_os_code(self, m_blkid, m_is_prov, _m_xen):
        """Identify OS_CODE."""
        m_blkid.return_value = self.blkid_oscode
        m_is_prov.return_value = False
        assert (ibm.Platforms.OS_CODE, "/dev/xvdh") == ibm.get_ibm_platform()

    def test_id_os_code_must_match_uuid(self, m_blkid, m_is_prov, _m_xen):
        """Test against false positive on openstack with non-ibm UUID."""
        blkid = self.blkid_oscode
        blkid["/dev/xvdh"]["UUID"] = "9999-9999"
        m_blkid.return_value = blkid
        m_is_prov.return_value = False
        assert (None, None) == ibm.get_ibm_platform()


@mock.patch(D_PATH + "_read_system_uuid", return_value=None)
@mock.patch(D_PATH + "get_ibm_platform")
class TestReadMD:
    """Test the read_datasource helper."""

    template_md = {
        "files": [],
        "network_config": {"content_path": "/content/interfaces"},
        "hostname": "ci-fond-ram",
        "name": "ci-fond-ram",
        "domain": "testing.ci.cloud-init.org",
        "meta": {"dsmode": "net"},
        "uuid": "8e636730-9f5d-c4a5-327c-d7123c46e82f",
        "public_keys": {"1091307": "ssh-rsa AAAAB3NzaC1...Hw== ci-pubkey"},
    }

    oscode_md = {
        "hostname": "ci-grand-gannet.testing.ci.cloud-init.org",
        "name": "ci-grand-gannet",
        "uuid": "2f266908-8e6c-4818-9b5c-42e9cc66a785",
        "random_seed": "bm90LXJhbmRvbQo=",
        "crypt_key": "ssh-rsa AAAAB3NzaC1yc2..n6z/",
        "configuration_token": "eyJhbGciOi..M3ZA",
        "public_keys": {"1091307": "ssh-rsa AAAAB3N..Hw== ci-pubkey"},
    }

    content_interfaces = dedent(
        """\
        auto lo
        iface lo inet loopback

        auto eth0
        allow-hotplug eth0
        iface eth0 inet static
        address 10.82.43.5
        netmask 255.255.255.192
        """
    )

    userdata = b"#!/bin/sh\necho hi mom\n"
    # meta.js file gets json encoded userdata as a list.
    meta_js = '["#!/bin/sh\necho hi mom\n"]'
    vendor_data = {
        "cloud-init": "#!/bin/bash\necho 'root:$6$5ab01p1m1' | chpasswd -e"
    }

    network_data = {
        "links": [
            {
                "id": "interface_29402281",
                "name": "eth0",
                "mtu": None,
                "type": "phy",
                "ethernet_mac_address": "06:00:f1:bd:da:25",
            },
            {
                "id": "interface_29402279",
                "name": "eth1",
                "mtu": None,
                "type": "phy",
                "ethernet_mac_address": "06:98:5e:d0:7f:86",
            },
        ],
        "networks": [
            {
                "id": "network_109887563",
                "link": "interface_29402281",
                "type": "ipv4",
                "ip_address": "10.82.43.2",
                "netmask": "255.255.255.192",
                "routes": [
                    {
                        "network": "10.0.0.0",
                        "netmask": "255.0.0.0",
                        "gateway": "10.82.43.1",
                    },
                    {
                        "network": "161.26.0.0",
                        "netmask": "255.255.0.0",
                        "gateway": "10.82.43.1",
                    },
                ],
            },
            {
                "id": "network_109887551",
                "link": "interface_29402279",
                "type": "ipv4",
                "ip_address": "108.168.194.252",
                "netmask": "255.255.255.248",
                "routes": [
                    {
                        "network": "0.0.0.0",
                        "netmask": "0.0.0.0",
                        "gateway": "108.168.194.249",
                    }
                ],
            },
        ],
        "services": [
            {"type": "dns", "address": "10.0.80.11"},
            {"type": "dns", "address": "10.0.80.12"},
        ],
    }

    sysuuid = "7f79ebf5-d791-43c3-a723-854e8389d59f"

    def _get_expected_metadata(self, os_md):
        """return expected 'metadata' for data loaded from meta_data.json."""
        os_md = copy.deepcopy(os_md)
        renames = (
            ("hostname", "local-hostname"),
            ("uuid", "instance-id"),
            ("public_keys", "public-keys"),
        )
        ret = {}
        for osname, mdname in renames:
            if osname in os_md:
                ret[mdname] = os_md[osname]
        if "random_seed" in os_md:
            ret["random_seed"] = base64.b64decode(os_md["random_seed"])

        return ret

    def test_provisioning_md(self, m_platform, m_sysuuid):
        """Provisioning env with a metadata disk should return None."""
        m_platform.return_value = (
            ibm.Platforms.TEMPLATE_PROVISIONING_METADATA,
            "/dev/xvdh",
        )
        assert ibm.read_md() is None

    def test_provisioning_no_metadata(self, m_platform, m_sysuuid):
        """Provisioning env with no metadata disk should return None."""
        m_platform.return_value = (
            ibm.Platforms.TEMPLATE_PROVISIONING_NODATA,
            None,
        )
        assert ibm.read_md() is None

    def test_provisioning_not_ibm(self, m_platform, m_sysuuid):
        """Provisioning env but not identified as IBM should return None."""
        m_platform.return_value = (None, None)
        assert ibm.read_md() is None

    def test_template_live(self, m_platform, m_sysuuid, tmp_path):
        """Template live environment should be identified."""
        tmpdir = str(tmp_path)
        m_platform.return_value = (
            ibm.Platforms.TEMPLATE_LIVE_METADATA,
            tmpdir,
        )
        m_sysuuid.return_value = self.sysuuid

        test_helpers.populate_dir(
            tmpdir,
            {
                "openstack/latest/meta_data.json": json.dumps(
                    self.template_md
                ),
                "openstack/latest/user_data": self.userdata,
                "openstack/content/interfaces": self.content_interfaces,
                "meta.js": self.meta_js,
            },
        )

        ret = ibm.read_md()
        if ret is None:  # this is needed for mypy - ensures ret is not None
            pytest.fail("read_md returned None unexpectedly")
        assert ibm.Platforms.TEMPLATE_LIVE_METADATA == ret["platform"]
        assert tmpdir == ret["source"]
        assert self.userdata == ret["userdata"]
        assert self._get_expected_metadata(self.template_md) == ret["metadata"]
        assert self.sysuuid == ret["system-uuid"]

    def test_os_code_live(self, m_platform, m_sysuuid, tmp_path):
        """Verify an os_code metadata path."""
        tmpdir = str(tmp_path)
        m_platform.return_value = (ibm.Platforms.OS_CODE, tmpdir)
        netdata = json.dumps(self.network_data)
        test_helpers.populate_dir(
            tmpdir,
            {
                "openstack/latest/meta_data.json": json.dumps(self.oscode_md),
                "openstack/latest/user_data": self.userdata,
                "openstack/latest/vendor_data.json": json.dumps(
                    self.vendor_data
                ),
                "openstack/latest/network_data.json": netdata,
            },
        )

        ret = ibm.read_md()
        if ret is None:  # this is needed for mypy - ensures ret is not None
            pytest.fail("read_md returned None unexpectedly")
        assert ibm.Platforms.OS_CODE == ret["platform"]
        assert tmpdir == ret["source"]
        assert self.userdata == ret["userdata"]
        assert self._get_expected_metadata(self.oscode_md) == ret["metadata"]

    def test_os_code_live_no_userdata(self, m_platform, m_sysuuid, tmp_path):
        """Verify os_code without user-data."""
        tmpdir = str(tmp_path)
        m_platform.return_value = (ibm.Platforms.OS_CODE, tmpdir)
        test_helpers.populate_dir(
            tmpdir,
            {
                "openstack/latest/meta_data.json": json.dumps(self.oscode_md),
                "openstack/latest/vendor_data.json": json.dumps(
                    self.vendor_data
                ),
            },
        )

        ret = ibm.read_md()
        if ret is None:  # this is needed for mypy - ensures ret is not None
            pytest.fail("read_md returned None unexpectedly")
        assert ibm.Platforms.OS_CODE == ret["platform"]
        assert tmpdir == ret["source"]
        assert ret["userdata"] is None
        assert self._get_expected_metadata(self.oscode_md) == ret["metadata"]


@pytest.mark.usefixtures("fake_filesystem")
class TestIsIBMProvisioning:
    """Test the _is_ibm_provisioning method."""

    inst_log = "/root/swinstall.log"
    prov_cfg = "/root/provisioningConfiguration.cfg"
    boot_ref = "/proc/1/environ"

    def test_no_config(self):
        """No provisioning config means not provisioning."""
        assert not ibm._is_ibm_provisioning()

    def test_config_only(self, tmp_path):
        """A provisioning config without a log means provisioning."""
        test_helpers.populate_dir(str(tmp_path), {self.prov_cfg: "key=value"})
        assert ibm._is_ibm_provisioning()

    def test_config_with_old_log(self, caplog, tmp_path):
        """A config with a log from previous boot is not provisioning."""
        data = {
            self.prov_cfg: ("key=value\nkey2=val2\n", -10),
            self.inst_log: ("log data\n", -30),
            self.boot_ref: ("PWD=/", 0),
        }
        test_helpers.populate_dir_with_ts(str(tmp_path), data)
        assert not ibm._is_ibm_provisioning()
        assert "from previous boot" in caplog.text

    def test_config_with_new_log(self, caplog, tmp_path):
        """A config with a log from this boot is provisioning."""
        data = {
            self.prov_cfg: ("key=value\nkey2=val2\n", -10),
            self.inst_log: ("log data\n", 30),
            self.boot_ref: ("PWD=/", 0),
        }
        test_helpers.populate_dir_with_ts(str(tmp_path), data)
        assert ibm._is_ibm_provisioning()
        assert "from current boot" in caplog.text

    def test_config_and_log_no_reference(self, caplog, tmp_path):
        """If the config and log existed, but no reference, assume not."""
        test_helpers.populate_dir(
            str(tmp_path),
            {self.prov_cfg: "key=value", self.inst_log: "log data\n"},
        )
        assert not ibm._is_ibm_provisioning()
        assert "no reference file" in caplog.text


class TestDataSourceIBMCloud:
    @pytest.fixture(autouse=True)
    def fixture(self, paths):
        util.ensure_dir(paths.cloud_dir)
        self.ds = ibm.DataSourceIBMCloud(sys_cfg={}, distro=None, paths=paths)

    def test_get_data_false(self):
        """When read_md returns None, get_data returns False."""
        with mock.patch(D_PATH + "read_md", return_value=None):
            assert not self.ds.get_data()

    def test_get_data_processes_read_md(self):
        """get_data processes and caches content returned by read_md."""
        md = {
            "metadata": {},
            "networkdata": "net",
            "platform": "plat",
            "source": "src",
            "system-uuid": "uuid",
            "userdata": "ud",
            "vendordata": "vd",
        }
        with mock.patch(D_PATH + "read_md", return_value=md):
            assert self.ds.get_data()
        assert "src" == self.ds.source
        assert "plat" == self.ds.platform
        assert {} == self.ds.metadata
        assert "ud" == self.ds.userdata_raw
        assert "net" == self.ds.network_json
        assert "uuid" == self.ds.system_uuid
        assert "ibmcloud" == self.ds.cloud_name
        assert "ibmcloud" == self.ds.platform_type
        assert "plat (src)" == self.ds.subplatform
