# Copyright (C) 2017 Red Hat, Inc., Bryn M. Reeves <bmr@redhat.com>
#
# osprofile_tests.py - Boom OS profile tests.
#
# This file is part of the boom project.
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions
# of the GNU General Public License v.2.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
import unittest
import logging
from sys import stdout
from os import listdir, makedirs
from os.path import abspath, exists, join
import shutil

log = logging.getLogger()
log.level = logging.DEBUG
log.addHandler(logging.FileHandler("test.log"))

from boom.osprofile import *
from boom.hostprofile import *
from boom.bootloader import *
from boom import *

# For private member validation checks
import boom.hostprofile

from tests import *

BOOT_ROOT_TEST = abspath("./tests")
set_boot_path(BOOT_ROOT_TEST)

BOOM_ENTRY_MACHINE_ID="BOOM_ENTRY_MACHINE_ID"

def _count_value_in_key(dir_path, ext, key_name, xvalue,
                        exact=False, default=False):
    """Helper function to count the number of times ``xvalue`` appears
        as the value of key ``key_name`` in the set of files found in
        ``dir_path`` having file extension ``ext`.

        If ``exact`` is true the value is compared for equality.
        Otherwise the value is counted if the string ``value`` appears
        anywhere within the named key value.

        If ``default`` is ``True``, and the key is not found in a
        profile, it is assumed that ``xvalue`` is the default value,
        and that the relevant profile has inherited this value from
        the embedded ``OsProfile`` defaults.

        This is used to count profiles by property independently of
        the ``boom`` library modules for comparison of results
        returned from the API.
    """
    count = 0
    filecount = 1
    for f_name in listdir(dir_path):
        if not f_name.endswith(ext):
            continue
        f_path = join(dir_path, f_name)
        found_in_file = False
        with open(f_path, "r") as f:
            for line in f.readlines():
                (key, value) = parse_name_value(line)
                if key not in HOST_PROFILE_KEYS:
                    continue
                if key == key_name:
                    found_in_file = True
                    if not exact:
                        count += 1 if xvalue in value else 0
                    else:
                        count += 1 if xvalue == value else 0
            if default and not found_in_file:
                count += 1
            filecount += 1
    return count


class HostProfileTests(unittest.TestCase):
    # Master boom configuration path for sandbox
    boom_path = join(BOOT_ROOT_TEST, "boom")

    # Master BLS loader directory for sandbox
    loader_path = join(BOOT_ROOT_TEST, "loader")

    def setUp(self):
        reset_sandbox()

        # Sandbox paths
        boot_sandbox = join(SANDBOX_PATH, "boot")
        boom_sandbox = join(SANDBOX_PATH, "boot/boom")
        loader_sandbox = join(SANDBOX_PATH, "boot/loader")

        makedirs(boot_sandbox)
        shutil.copytree(self.boom_path, boom_sandbox)
        shutil.copytree(self.loader_path, loader_sandbox)

        set_boot_path(boot_sandbox)

        drop_host_profiles()
        drop_profiles()

    def tearDown(self):
        drop_host_profiles()
        drop_profiles()
        rm_sandbox()
        reset_boom_paths()

    # Module tests
    def test_import(self):
        import boom.hostprofile

    # Profile store tests

    def test_load_profiles(self):
        # Test that loading the test profiles succeeds.
        load_host_profiles()

    # HostProfile tests

    def test_HostProfile__str__(self):
        load_profiles()
        load_host_profiles()
        hp = HostProfile(machine_id="ffffffffffffffff", host_name="localhost",
                         label='', os_id="3fc389b")

        xstr = (
                'Host ID: "83fb23b393d6460e18e3694a8766b06ade021c3f",\n'
                'Host name: "localhost",\nMachine ID: "ffffffffffffffff",\n'
                'OS ID: "3fc389bba581e5b20c6a46c7fc31b04be465e973",\n'
                'Host label: "",\nName: "Red Hat Enterprise Linux Server", '
                'Short name: "rhel", Version: "7.2 (Maipo)",\n'
                'Version ID: "7.2", UTS release pattern: "el7",\n'
                'Kernel pattern: "/vmlinuz-%{version}", '
                'Initramfs pattern: "/initramfs-%{version}.img",\n'
                'Root options (LVM2): "rd.lvm.lv=%{lvm_root_lv}",\n'
                'Root options (BTRFS): "rootflags=%{btrfs_subvolume}",\n'
                'Options: "root=%{root_device} ro %{root_opts} rhgb quiet"'
        )

        self.assertEqual(str(hp), xstr)
        hp.delete_profile()

    def test_HostProfile__repr__(self):
        load_profiles()
        load_host_profiles()
        hp = HostProfile(machine_id="ffffffffffffffff", host_name="localhost",
                         label='', os_id="3fc389b")

        xrepr = ('HostProfile(profile_data={'
                 'BOOM_HOST_ID:"83fb23b393d6460e18e3694a8766b06ade021c3f", '
                 'BOOM_HOST_NAME:"localhost", '
                 'BOOM_ENTRY_MACHINE_ID:"ffffffffffffffff", '
                 'BOOM_OS_ID:"3fc389bba581e5b20c6a46c7fc31b04be465e973", '
                 'BOOM_HOST_LABEL:"", '
                 'BOOM_OS_NAME:"Red Hat Enterprise Linux Server", '
                 'BOOM_OS_SHORT_NAME:"rhel", BOOM_OS_VERSION:"7.2 (Maipo)", '
                 'BOOM_OS_VERSION_ID:"7.2", BOOM_OS_UNAME_PATTERN:"el7", '
                 'BOOM_OS_KERNEL_PATTERN:"/vmlinuz-%{version}", '
                 'BOOM_OS_INITRAMFS_PATTERN:"/initramfs-%{version}.img", '
                 'BOOM_OS_ROOT_OPTS_LVM2:"rd.lvm.lv=%{lvm_root_lv}", '
                 'BOOM_OS_ROOT_OPTS_BTRFS:"rootflags=%{btrfs_subvolume}", '
                 'BOOM_OS_OPTIONS:"root=%{root_device} ro %{root_opts} rhgb '
                 'quiet"})'
        )

        self.assertEqual(repr(hp), xrepr)
        hp.delete_profile()

    def test_HostProfile(self):
        # Test HostProfile init from kwargs
        with self.assertRaises(ValueError) as cm:
            hp = HostProfile(host_name="localhost", os_id="3fc389b")
        with self.assertRaises(ValueError) as cm:
            hp = HostProfile(machine_id="ffffffffffffffff", host_name="localhost")
        with self.assertRaises(ValueError) as cm:
            hp = HostProfile(machine_id="ffffffffffffffff", os_id="3fc389b")

        hp = HostProfile(machine_id="ffffffffffffffff", host_name="localhost",
                         os_id="3fc389b", label='')

        self.assertTrue(hp)

        # os_id for RHEL-7.2
        self.assertEqual(hp.os_id, "3fc389bba581e5b20c6a46c7fc31b04be465e973")

        hp.delete_profile()

    def test_HostProfile_from_profile_data(self):
        profile_data = {
            BOOM_ENTRY_MACHINE_ID: "fffffffffffffff",
            BOOM_HOST_NAME: "localhost",
            BOOM_OS_ID: "3fc389bba581e5b20c6a46c7fc31b04be465e973",
            BOOM_OS_KERNEL_PATTERN: "/vmlinuz-%{version}",
            BOOM_OS_INITRAMFS_PATTERN: "/initramfs-%{version}.img",
            BOOM_OS_ROOT_OPTS_LVM2: "rd.lvm.lv=%{lvm_root_lv} rh",
            BOOM_OS_ROOT_OPTS_BTRFS: "rootflags=%{btrfs_subvolume} rh",
            BOOM_OS_OPTIONS: "root=%{root_device} %{root_opts} rhgb quiet"
        }

        hp = HostProfile(profile_data=profile_data)
        self.assertTrue(hp)

        # Assert that overrides are present
        self.assertEqual(hp.root_opts_lvm2, "rd.lvm.lv=%{lvm_root_lv} rh")
        self.assertEqual(hp.root_opts_btrfs, "rootflags=%{btrfs_subvolume} rh")

        hp.delete_profile()

        # Remove the root options keys.
        profile_data.pop(BOOM_OS_ROOT_OPTS_LVM2, None)
        profile_data.pop(BOOM_OS_ROOT_OPTS_BTRFS, None)
        hp = HostProfile(profile_data=profile_data)

        # Assert that defaults are restored
        self.assertEqual(hp.root_opts_lvm2, "rd.lvm.lv=%{lvm_root_lv}")
        self.assertEqual(hp.root_opts_btrfs, "rootflags=%{btrfs_subvolume}")

        hp.delete_profile()

        # Remove the name key.
        profile_data.pop(BOOM_HOST_NAME, None)
        with self.assertRaises(ValueError) as cm:
            hp = HostProfile(profile_data=profile_data)

    def test_HostProfile_properties(self):
        hp = HostProfile(machine_id="fffffffffffffff", host_name="localhost",
                         os_id="3fc389b", label="")
        hp.kernel_pattern = "/vmlinuz-%{version}"
        hp.initramfs_pattern = "/initramfs-%{version}.img"
        hp.root_opts_lvm2 = "rd.lvm.lv=%{lvm_root_lv}"
        hp.root_opts_btrfs = "rootflags=%{btrfs_subvolume}"
        hp.options = "root=%{root_device} %{root_opts} rhgb quiet"

        self.assertEqual(hp.host_name, "localhost")
        self.assertEqual(hp.machine_id, "fffffffffffffff")
        self.assertEqual(hp.kernel_pattern, "/vmlinuz-%{version}")
        self.assertEqual(hp.initramfs_pattern,
                         "/initramfs-%{version}.img")
        self.assertEqual(hp.root_opts_lvm2, "rd.lvm.lv=%{lvm_root_lv}")
        self.assertEqual(hp.root_opts_btrfs,
                         "rootflags=%{btrfs_subvolume}")
        self.assertEqual(hp.options,
                         "root=%{root_device} %{root_opts} rhgb quiet")
        hp.delete_profile()

    def test_HostProfile_write(self):
        hp = HostProfile(machine_id="fffffffffffffff", host_name="localhost",
                         os_id="3fc389b", label="")
        hp.write_profile()
        profile_path = join(boom_host_profiles_path(),
                            "%s-%s.host" % (hp.host_id, hp.host_name))
        self.assertTrue(exists(profile_path))
        hp.delete_profile()

    @unittest.skipIf(have_root(), "DAC controls do not apply to root")
    def test_load_host_profiles_no_read(self):
        # Set the /boot path to a non-writable path for the test user.
        set_boot_path("/boot")
        with self.assertRaises(OSError) as cm:
            load_host_profiles()
        # Re-set test /boot
        set_boot_path(BOOT_ROOT_TEST)

    @unittest.skipIf(have_root(), "DAC controls do not apply to root")
    def test_write_host_profiles_fail(self):
        load_host_profiles()
        # Set the /boot path to a non-writable path for the test user.
        set_boot_path("/boot")
        # Create a dirty profile to write
        hp = HostProfile(machine_id="ffffffffffffff1", host_name="localhost",
                         os_id="3fc389b", label="")
        write_host_profiles()

        # Clean up dummy profile
        hp.delete_profile()
        # Re-set test /boot
        set_boot_path(BOOT_ROOT_TEST)

    def test_osprofile_write_profiles(self):
        load_host_profiles()
        write_host_profiles()

    def test_hostprofile_find_profiles_by_id(self):
        # Reload host profiles from disk: a failure to clean up in an
        # earlier test may have left the profile list in an inconsistent
        # state.
        host_id = "6af0980a6607b20cda34a45d2869c9be020914b4"
        load_host_profiles()
        hp = HostProfile(machine_id="fffffffffffffff", host_name="localhost",
                         os_id="3fc389b", label="testhp")
        hp.write_profile()
        hp_list = find_host_profiles(selection=Selection(host_id=host_id))
        self.assertEqual(len(hp_list), 1)
        self.assertEqual(hp_list[0].host_id, host_id)
        hp.delete_profile()

    def test_hostprofile_find_profiles_by_host_name(self):
        host_name = "localhost"
        hp_list = find_host_profiles(selection=Selection(host_name=host_name))
        nr_profiles = 0
        for f in listdir(boom_profiles_path()):
            if f.endswith(host_name + ".host"):
                nr_profiles += 1
        self.assertTrue(len(hp_list), nr_profiles)

    def test_hostprofile_find_profiles_by_add_opts(self):
        add_opts = "debug"
        select = Selection(host_add_opts=add_opts)
        hp_list = find_host_profiles(selection=select)
        # Adjusted to current test data
        self.assertEqual(1, len(hp_list))
        self.assertEqual(add_opts, hp_list[0].add_opts)

    def test_hostprofile_find_profiles_by_del_opts(self):
        del_opts = "rhgb quiet"
        select = Selection(host_del_opts=del_opts)
        hp_list = find_host_profiles(selection=select)
        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.del_opts == del_opts:
                nr_hps += 1
        self.assertEqual(nr_hps, len(hp_list))
        self.assertEqual(del_opts, hp_list[0].del_opts)

    def test_hostprofile_find_profiles_by_os_id(self):
        os_id = "3fc389b"
        hp_list = find_host_profiles(selection=Selection(os_id=os_id))
        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.os_id.startswith(os_id):
                nr_hps += 1
        self.assertEqual(nr_hps, len(hp_list))
        self.assertTrue(hp_list[0].os_id.startswith(os_id))

    def test_hostprofile_find_profiles_by_os_name(self):
        os_name = "Fedora"
        hp_list = find_host_profiles(selection=Selection(os_name=os_name))
        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.os_name == os_name:
                nr_hps += 1
        self.assertEqual(nr_hps, len(hp_list))
        self.assertEqual(os_name, hp_list[0].os_name)

    def test_hostprofile_find_profiles_by_os_short_name(self):
        os_short_name = "fedora"
        select = Selection(os_short_name=os_short_name)
        hp_list = find_host_profiles(selection=select)
        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.os_short_name == os_short_name:
                nr_hps += 1
        self.assertEqual(nr_hps, len(hp_list))
        self.assertEqual(os_short_name, hp_list[0].os_short_name)

    def test_hostprofile_find_profiles_by_os_version(self):
        os_version = "7.2 (Maipo)"
        select = Selection(os_version=os_version)
        hp_list = find_host_profiles(selection=select)
        # Adjusted to current test data
        self.assertEqual(2, len(hp_list))
        self.assertEqual(os_version, hp_list[0].os_version)

    def test_hostprofile_find_profiles_by_os_version_id(self):
        os_version_id = "7.2"
        select = Selection(os_version_id=os_version_id)
        hp_list = find_host_profiles(selection=select)
        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.os_version_id == os_version_id:
                nr_hps += 1
        self.assertEqual(nr_hps, len(hp_list))
        self.assertEqual(os_version_id, hp_list[0].os_version_id)

    def test_hostprofile_find_profiles_by_uname_pattern(self):
        uname_pattern = "el7"
        select = Selection(os_uname_pattern=uname_pattern)
        hp_list = find_host_profiles(selection=select)
        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.uname_pattern == uname_pattern:
                nr_hps += 1
        self.assertEqual(nr_hps, len(hp_list))
        self.assertEqual(uname_pattern, hp_list[0].uname_pattern)

    def test_hostprofile_find_profiles_by_kernel_pattern(self):
        kernel_pattern = "/vmlinuz-%{version}"
        select = Selection(os_kernel_pattern=kernel_pattern)
        hp_list = find_host_profiles(selection=select)

        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.kernel_pattern == kernel_pattern:
                nr_hps += 1
        self.assertEqual(nr_hps, len(hp_list))
        self.assertEqual(kernel_pattern, hp_list[0].kernel_pattern)

        # Non-matching
        kernel_pattern = "NOTAREALKERNELPATTERN"
        select = Selection(os_kernel_pattern=kernel_pattern)
        hp_list = find_host_profiles(selection=select)
        self.assertFalse(hp_list)

    def test_hostprofile_find_profiles_by_initramfs_pattern(self):
        initramfs_pattern = "/initramfs-%{version}.img"
        select = Selection(os_initramfs_pattern=initramfs_pattern)
        hp_list = find_host_profiles(selection=select)

        host_path = boom_host_profiles_path()
        profile_count = _count_value_in_key(host_path, ".host",
                                            BOOM_OS_INITRAMFS_PATTERN,
                                            initramfs_pattern,
                                            exact=True, default=True)

        self.assertEqual(profile_count, len(hp_list))
        self.assertEqual(initramfs_pattern, hp_list[0].initramfs_pattern)

        # Non-matching
        initramfs_pattern = "NOTAREALINITRAMFSPATTERN"
        select = Selection(os_initramfs_pattern=initramfs_pattern)
        hp_list = find_host_profiles(selection=select)
        # Adjusted to current test data
        self.assertFalse(hp_list)

    def test_hostprofile_find_profiles_by_options(self):
        options = "root=%{root_device} ro %{root_opts} rhgb quiet"
        select = Selection(os_options=options)
        hp_list = find_host_profiles(selection=select)
        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.options == options:
                nr_hps += 1
        self.assertEqual(nr_hps, len(hp_list))
        self.assertEqual(options, hp_list[0].options)

        # Non-matching
        options = "root=/dev/nodev ro rhgb quiet"
        select = Selection(os_options=options)
        hp_list = find_host_profiles(selection=select)
        # Adjusted to current test data
        self.assertFalse(hp_list)

    def test_hostprofile_find_host_profiles_not_loaded(self):
        # Find with automatic load
        hps = find_host_profiles()
        self.assertTrue(hps)

    def test_host_min_id_width(self):
        import boom.hostprofile
        xwidth = 7 # Adjusted to current test data
        load_host_profiles()
        width = boom.hostprofile.min_host_id_width()
        self.assertEqual(xwidth, width)

    def test_machine_min_id_width(self):
        import boom.hostprofile
        xwidth = 13 # Adjusted to current test data
        load_host_profiles()
        width = boom.hostprofile.min_machine_id_width()
        self.assertEqual(xwidth, width)

    def test_find_host_host_id(self):
        # Non-existent host_id
        hps = find_host_profiles(Selection(host_id="fffffff"))
        self.assertFalse(hps)

        host_id1 = "373ccd1"
        # Valid single host_id
        hps = find_host_profiles(Selection(host_id=host_id1))
        self.assertTrue(hps)

        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.host_id.startswith(host_id1):
                nr_hps += 1
        self.assertEqual(nr_hps, len(hps))

        # Two host profiles exist for this host_id (and with no label).
        # This is because this host is used for "hand edited host"
        # testing. Although this is not a valid configuratio that can
        # be reached using the Boom CLI it is still expected to work
        # and to produce consistent API behaviour.
        host_id2 = "5ebcb1f"
        hps = find_host_profiles(Selection(host_id=host_id2))
        self.assertTrue(hps)

        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.host_id.startswith(host_id2):
                nr_hps += 1
        self.assertEqual(nr_hps, len(hps))

        # Valid host_id scoped by label
        host_label = "ALABEL"
        hps = find_host_profiles(Selection(host_id=host_id1,
                                           host_label=host_label))
        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.host_id.startswith(host_id1):
                if hp.label == host_label:
                    nr_hps += 1
        self.assertTrue(hps)
        self.assertEqual(nr_hps, len(hps))

    def test_find_host_host_name(self):
        host_name = "localhost.localdomain"
        hps = find_host_profiles(Selection(host_name=host_name))
        self.assertTrue(hps)
        nr_hps = 0
        for hp in boom.hostprofile._host_profiles:
            if hp.host_name == host_name:
                nr_hps += 1
        self.assertEqual(nr_hps, len(hps))

    def test_find_host_host_short_name(self):
        load_host_profiles()
        host_name = "localhost"
        hps = find_host_profiles(Selection(host_short_name=host_name))
        self.assertTrue(hps)

        host_path = boom_host_profiles_path()
        profile_count = _count_value_in_key(host_path, ".host",
                                            BOOM_HOST_NAME, host_name)
        # Adjusted to current test data
        self.assertEqual(profile_count, len(hps))

    def test_find_host_host_label(self):
        hps = find_host_profiles(Selection(host_label="ALABEL"))
        self.assertTrue(hps)
        # Adjusted to current test data
        self.assertEqual(1, len(hps))

    def test_get_host_profile_by_id(self):
        load_host_profiles()
        m_id1 = "fffffffffffffff"
        hp = get_host_profile_by_id(m_id1)
        self.assertEqual(m_id1, hp.machine_id)

    def test_get_host_profile_by_id_not_loaded(self):
        m_id1 = "fffffffffffffff"


        hp = get_host_profile_by_id(m_id1)
        self.assertEqual(m_id1, hp.machine_id)

    def test_get_host_profile_by_id_and_label(self):
        m_id1 = "611f38fd887d41dea7ffffffffffff"
        label = "ALABEL"
        hp = get_host_profile_by_id(m_id1, label=label)
        self.assertEqual(m_id1, hp.machine_id)
        self.assertEqual(label, hp.label)

    def test_get_host_profile_by_id_no_match(self):
        m_id1 = "bazquxfoo"
        hp = get_host_profile_by_id(m_id1)
        self.assertFalse(hp)

    def test_match_host_profile(self):
        bes = find_entries(Selection(boot_id="242d946"))
        self.assertTrue(bes)
        be = bes[0]
        self.assertTrue(be)
        machine_id = be.machine_id
        hp = match_host_profile(be)
        self.assertTrue(hp)
        self.assertEqual(be.machine_id, hp.machine_id)

# vim: set et ts=4 sw=4 :
