#!/usr/bin/python2.6

#

#

# CDDL HEADER START

#

# The contents of this file are subject to the terms of the

# Common Development and Distribution License (the "License").

# You may not use this file except in compliance with the License.

#

# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

# or http://www.opensolaris.org/os/licensing.

# See the License for the specific language governing permissions

# and limitations under the License.

#

# When distributing Covered Code, include this CDDL HEADER in each

# file and include the License file at usr/src/OPENSOLARIS.LICENSE.

# If applicable, add the following below this CDDL HEADER, with the

# fields enclosed by brackets "[]" replaced with your own identifying

# information: Portions Copyright [yyyy] [name of copyright owner]

#

# CDDL HEADER END

#

#

#

'''Some unit tests to cover engine functionality'''

import logging

import os

import sys

import shutil

import threading

import unittest

import warnings

from itertools import izip

from logging import ERROR, DEBUG

import solaris_install.engine as engine

import solaris_install.engine.checkpoint_data as checkpoint_data

import osol_install.errsvc as errsvc

import osol_install.liberrsvc as liberrsvc

from empty_checkpoint import EmptyCheckpoint

from solaris_install.engine.test.engine_test_utils import reset_engine, \

    get_new_engine_instance

from solaris_install.data_object import DataObject

_THIS_DIR = os.path.dirname(os.path.abspath(__file__))

sys.path.append(_THIS_DIR)

class MockDataset(object):

    ''' Fake Dataset object so the real ZFS dataset object does

        not need to be used for testing

    '''

    def __init__(self, path):

        self.exists = True

        self.mountpoint = path

        self.snapped = (None, None)

        self.snapshot_list = []

    def snapshot(self, name, overwrite=False):

        self.snapped = (name, overwrite)

    def snapname(self, name):

        return self.mountpoint + '@' + name

    def rollback(self, name, recursive=True):

        pass

    def get(self, property):

        return getattr(self, property)

class MockDOC(object):

    ''' Fake DOC object so the real DataObjectCache object we do not rely

        on the actual DataObjectCache class for testing.

    '''

    def take_snapshot(self, dummy):

        self.snapshotted = dummy

    def insert_children(self, dummy):

        pass

    @property

    def persistent(self):

        return self

    def get_first_child(self, name=None):

        return self

    def clear(self):

        pass

    def load_from_snapshot(self, filename):

        self.loaded_from = filename

class MockCheckpointRegData(DataObject):

    ''' Fake CheckpointRegData object so we do not rely on the actual

        CheckpointRegData object for testing

    '''

    def __init__(self):

        DataObject.__init__(self, "MokeCheckpointRegData")

    def to_xml(self):

        return None

    def from_xml(cls, xml_node):

        return None

    def can_handle(cls, xml_node):

        return False

class MockCheckpointData(object):

    ''' Fake CheckpointData object so we do not rely on the actual

        CheckpointData object for testing

    '''

    def __init__(self):

        self.cp_info = MockCheckpointRegData()

        self.name = "MockCheckpointData"

        self.prog_reported = 0

        self.prog_est_ratio = 0

class EngineTest(unittest.TestCase):

    ''' Tests that validates the interfaces in the install engine code.

        All tests here do not require a user to be

        root to execute.

    '''

    def setUp(self):

        self.callback_results = (None, None)

        self.callback_executed = threading.Event()

        self.engine = get_new_engine_instance()

    def tearDown(self):

        # Force spawning of fresh singleton for each test.

        reset_engine(self.engine)

        self.engine = None

        self.callback_results = None

        self.callback_executed = None

    def _exec_cp_callback(self, status, errsvc):

        ''' Callback function for none-block execute_checkpoints() tests '''

        self.callback_results = (status, errsvc)

        self.callback_executed.set()

    def create_fake_cache_snapshots(self):

        ''' Create a few fake DOC snapshots in /tmp '''

        cp_names = ["cp_one", "cp_two", "cp_three", "cp_four"]

        self.full_cp_names = []

        # create a tmp directory for it

        self.cache_dir_name = ("/tmp/list_doc_snapshot_test_%s") % os.getpid()

        os.mkdir(self.cache_dir_name)

        for cp in cp_names:

                           engine.InstallEngine.CACHE_FILE_NAME_PREFIX + cp)

            shutil.copyfile("/etc/hosts", file_name)

            self.full_cp_names.append(file_name)

    def destroy_fake_cache_snapshots(self):

        ''' Destroyes the fake DOC snapshots in /tmp used for testing '''

        shutil.rmtree(self.cache_dir_name)

class SimpleEngineTests(EngineTest):

    '''Tests the less complicated engine functionality'''

    def test_engine_is_singleton(self):

        engine.InstallEngine._instance = None

        self.assertRaises(engine.SingletonError,

                          engine.InstallEngine.get_instance)

        self.assertTrue(isinstance(install_engine, engine.InstallEngine))

        engine_instance = engine.InstallEngine.get_instance()

        self.assertTrue(install_engine is engine_instance)

    def test_check_callback_None(self):

        '''Assert InstallEngine._check_callback accepts None '''

        try:

            self.engine._check_callback(None)

        except TypeError:

            self.fail("Engine did not accept 'None' for callback")

    def test_check_callback_varargs(self):

        '''Assert InstallEngine._check_callback accepts a vararg function'''

        def vararg_func(*args):

            pass

        try:

            self.engine._check_callback(vararg_func)

        except TypeError:

            self.fail("Engine did not accept function with *args param")

    def test_check_callback_two_args(self):

        '''Assert InstallEngine._check_callback accepts a two argument function'''

        def arg_func(arg1, arg2):

            pass

        try:

            self.engine._check_callback(arg_func)

        except TypeError:

            self.fail("Engine did not accept function with exactly two args")

    def test_check_callback_under_two_args(self):

        '''Asserts InstallEngine._check_callback fails on a single argument function'''

        def arg_func(arg1):

            pass

        self.assertRaises(TypeError, self.engine._check_callback, arg_func)

    def test_check_callback_over_two_required_args(self):

        '''Asserts InstallEngine._check_callback fails if 3 or more args are required'''

        def arg_func(arg1, arg2, arg3):

            pass

        self.assertRaises(TypeError, self.engine._check_callback, arg_func)

    def test_check_callback_with_kwargs(self):

        '''Asserts InstallEngine._check_callback handles functions with keyword args'''

        def kwarg_func_all(arg1=None, arg2=None, arg3=None):

            pass

        try:

            self.engine._check_callback(kwarg_func_all)

        except TypeError:

            self.fail("Engine did not accept function with kwargs for all"

                      "arguments")

    def test_check_callback_2_arg_optional_kwarg(self):

        '''Asserts InstallEngine._check_callback handles a function with an optional keyword argument'''

        def optional_kwarg(arg1, arg2, arg3=None):

            pass

        try:

            self.engine._check_callback(optional_kwarg)

        except TypeError:

            self.fail("Engine did not accept function with optional kwarg")

    def test_check_callback_bound_class_method(self):

        '''Asserts InstallEngine._check_callback handles bound class methods'''

        class DummyClass(object):

            def callback(self, status, errsvc):

                pass

        instance = DummyClass()

        try:

            self.engine._check_callback(instance.callback)

        except TypeError:

            self.fail("Engine did not accept bound class method")

    def test_check_callback_unbound_class_method(self):

        '''Asserts InstallEngine._check_callback rejects unbound class methods.

        (Unbound methods require a class instance as the first argument)'''

        class DummyClass(object):

            def callback(self, status, errsvc):

                pass

        self.assertRaises(TypeError, self.engine._check_callback,

                          DummyClass.callback)

    def test_snapshot_tmp_no_dataset(self):

        self.engine._dataset = None

        self.engine.data_object_cache = MockDOC()

        cp_data = MockCheckpointData()

        self.engine.snapshot(cp_data=cp_data)

        self.assertEqual(cp_data.zfs_snap, None)

        self.assertEqual(self.engine.doc.snapshotted,

                         cp_data.data_cache_path,

                         "DOC path for Checkpoint not set")

    def test_snapshot_tmp_dataset_no_exist(self):

        ds = MockDataset("mock")

        self.engine._dataset = ds

        self.engine.dataset.exists = False

        self.engine.data_object_cache = MockDOC()

        cp_data = MockCheckpointData()

        self.engine.snapshot(cp_data=cp_data)

        self.assertEqual(cp_data.zfs_snap, None)

        self.assertEqual(self.engine.doc.snapshotted,

                         cp_data.data_cache_path,

                         "DOC path for Checkpoint not set")

    def test_snapshot_zfs_dataset_exists(self):

        ds = MockDataset("mock")

        self.engine._dataset = ds

        self.engine.dataset.exists = True

        self.engine.data_object_cache = MockDOC()

        cp_data = MockCheckpointData()

        self.engine.snapshot(cp_data=cp_data)

        self.assertEqual(cp_data.zfs_snap, ds.snapped[0])

        self.assertEqual(self.engine.doc.snapshotted,

                         cp_data.data_cache_path,

                         "DOC path for Checkpoint not set")

class EngineCheckpointsBase(EngineTest):

    def setUp(self):

        EngineTest.setUp(self)

        if _THIS_DIR == ".":

            self.cp_data_args = ("empty_checkpoint", "EmptyCheckpoint")

            self.failed_cp_data_args = ("empty_checkpoint",

                                        "FailureEmptyCheckpoint")

        else:

            self.cp_data_args = (_THIS_DIR + "/empty_checkpoint",

                                 "EmptyCheckpoint")

            self.failed_cp_data_args = (_THIS_DIR + "/empty_checkpoint",

                                        "FailureEmptyCheckpoint")

        if _THIS_DIR.startswith("/"):

            self.expected_mod_name = "empty_checkpoint"

            self.expected_mod_path = _THIS_DIR

        else:

            self.expected_mod_name = self.cp_data_args[0]

            self.expected_mod_path = None

        self.name_list = ["one", "two", "three", "four", "five"]

        # Squelch the logging during checkpoint registration

        # (it clutters the test output)

        level = engine.LOGGER.level

        engine.LOGGER.level = 100

        for name in self.name_list:

            self.engine.register_checkpoint(name, *self.cp_data_args)

        engine.LOGGER.level = level

    def tearDown(self):

        EngineTest.tearDown(self)

class EngineCheckpointsTest(EngineCheckpointsBase):

    '''More complicated InstallEngine tests requiring DOC, checkpoints, etc.'''

    def test_register_checkpoint(self):

        self.assertEquals(len(self.name_list), len(self.engine._checkpoints))

        for cp_data in self.engine._checkpoints:

            self.assertEquals(cp_data.cp_info.mod_name, self.expected_mod_name,

                              "Module name: Expected '%s', got '%s'" %

                              (self.expected_mod_name, cp_data.cp_info.mod_name))

            self.assertEquals(cp_data.cp_info.module_path, self.expected_mod_path,

                              "Module path: Expected '%s', got '%s'" %

                              (self.expected_mod_path, cp_data.cp_info.module_path))

    def test_register_duplicate_checkpoint(self):

        '''Assert cannot register two checkpoints with the same name'''

        self.assertRaises(engine.ChkptRegistrationError,

                          self.engine.register_checkpoint,

                          self.name_list[1], *self.cp_data_args)

    def test_get_full_checkpoint_list(self):

        '''InstallEngine.get_exec_list - Default args (full list)'''

        exec_list = self.engine.get_exec_list()

        self.assertEquals(len(self.name_list), len(exec_list))

        for exec_cp, cp_name in zip(exec_list, self.name_list):

            self.assertEquals(exec_cp.name, cp_name, "Exec list returned "

                            "checkpoints out of order")

    def test_get_cp_list_pause_before(self):

        '''InstallEngine.get_exec_list - pause_before arg works'''

        stop_at = 2

        exec_list = self.engine.get_exec_list(pause_before=self.name_list[stop_at])

        self.assertEquals(len(self.name_list[:stop_at]), len(exec_list))

        for exec_cp, cp_name in zip(exec_list, self.name_list):

            self.assertEquals(exec_cp.name, cp_name, "Exec list returned "

                            "checkpoints out of order")

    def test_get_cp_list_resume_from(self):

        '''InstallEngine.get_exec_list - resume_from arg works'''

        for cp_data in self.engine._checkpoints:

            cp_data.completed = True

        resume_from = 1

        exec_list = self.engine.get_exec_list(start_from=self.name_list[resume_from])

        self.assertEquals(len(self.name_list[resume_from:]), len(exec_list))

        for exec_cp, cp_name in zip(exec_list, self.name_list[resume_from:]):

            self.assertEquals(exec_cp.name, cp_name, "Exec list returned "

                            "checkpoints out of order")

    def test_get_cp_list_pause_before_resume(self):

        '''InstallEngine.get_exec_list - Error when pause_before cp is before the start_from cp in sequence'''

        for cp_data in self.engine._checkpoints:

            cp_data.completed = True

        resume_from = 2

        pause_before = 1

        self.assertRaises(engine.UsageError,

                          self.engine.get_exec_list, start_from=self.name_list[resume_from],

                          pause_before=self.name_list[pause_before])

    def test_get_cp_list_resume_prev_cp_not_complete(self):

        '''InstallEngine.get_exec_list - Error when prior cp's not complete'''

        resume_from = 1

        self.assertRaises(engine.UsageError,

                          self.engine.get_exec_list, start_from=self.name_list[resume_from])

    def test_resume_from_last_complete(self):

        '''InstallEngine.get_exec_list - Verify correct exec list is returned when some checkpoints are completed '''

        self.engine._checkpoints[0].completed = True

        first_inc = self.engine.get_first_incomplete()

        self.assertTrue(first_inc is self.engine._checkpoints[1])

        exec_list = self.engine.get_exec_list(start_from=None)

        self.assertFalse(self.engine._checkpoints[0] in exec_list)

    def test_resume_from_last_none_complete(self):

        '''InstallEngine.get_exec_list - Verify when none of the checkpoints are complete'''

        first_inc = self.engine.get_first_incomplete()

        self.assertTrue(first_inc is self.engine._checkpoints[0])

        exec_list = self.engine.get_exec_list()

        self.assertEquals(len(self.name_list), len(exec_list))

        for exec_cp, cp_name in zip(exec_list, self.name_list):

            self.assertEquals(exec_cp.name, cp_name, "Exec list returned "

                              "checkpoints out of order")

    def test_resume_from_last_all_complete(self):

        '''InstallEngine.get_exec_list - Verify nothing is returned when all cp's complete'''

        for cp_data in self.engine._checkpoints:

            cp_data.completed = True

        first_inc = self.engine.get_first_incomplete()

        self.assertTrue(first_inc is None)

        exec_list = self.engine.get_exec_list()

        self.assertEquals(0, len(exec_list))

    def test_get_cp_list_resume_no_exist(self):

        '''InstallEngine.get_exec_list - Verify UsageError on nonexistent start_from cp'''

        self.assertRaises(engine.UsageError,

                          self.engine.get_exec_list, start_from="NOEXIST")

    def test_get_cp_list_pause_before_no_exist(self):

        '''InstallEngine.get_exec_list - Verify UsageError on nonexistent pause_before cp'''

        self.assertRaises(engine.UsageError,

                          self.engine.get_exec_list, pause_before="NOEXIST")

    def test_get_cache_filename_no_dataset(self):

        self.engine.dataset = None

        path = self.engine.get_cache_filename("cp")

        self.assertEquals(path, os.path.join(self.engine._tmp_cache_path,

                                             ".data_cache.cp"))

    def test_get_cache_filename_dataset(self):

        self.engine._tmp_cache_path = "/a/b/c"

        dataset = MockDataset(self.engine._tmp_cache_path)

        self.engine._dataset = dataset

        path = self.engine.get_cache_filename("cp")

        self.assertEquals(path, os.path.join(dataset.mountpoint,

                                             ".data_cache.cp"))

        self.engine._tmp_cache_path = None