#!/usr/bin/python
#

# Copyright (C) 2006, 2007 Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
# IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.


"""Script for unittesting the hooks module"""


import unittest
import os
import time
import tempfile
import os.path

from ganeti import errors
from ganeti import opcodes
from ganeti import hooksmaster
from ganeti import backend
from ganeti import constants
from ganeti import cmdlib
from ganeti.rpc import node as rpc
from ganeti import compat
from ganeti import pathutils
from ganeti.constants import HKR_SUCCESS, HKR_FAIL, HKR_SKIP

from mocks import FakeConfig, FakeProc, FakeContext

import testutils


class FakeLU(cmdlib.LogicalUnit):
  HPATH = "test"

  def BuildHooksEnv(self):
    return {}

  def BuildHooksNodes(self):
    return ["a"], ["a"]


class TestHooksRunner(unittest.TestCase):
  """Testing case for HooksRunner"""
  def setUp(self):
    self.torm = []
    self.tmpdir = tempfile.mkdtemp()
    self.torm.append((self.tmpdir, True))
    self.logdir = tempfile.mkdtemp()
    self.torm.append((self.logdir, True))
    self.hpath = "fake"
    self.ph_dirs = {}
    for i in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      dname = "%s/%s-%s.d" % (self.tmpdir, self.hpath, i)
      os.mkdir(dname)
      self.torm.append((dname, True))
      self.ph_dirs[i] = dname
    self.hr = backend.HooksRunner(hooks_base_dir=self.tmpdir)

  def tearDown(self):
    self.torm.reverse()
    for path, kind in self.torm:
      if kind:
        os.rmdir(path)
      else:
        os.unlink(path)

  def _rname(self, fname):
    return "/".join(fname.split("/")[-2:])

  def testEmpty(self):
    """Test no hooks"""
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), [])

  def testSkipNonExec(self):
    """Test skip non-exec file"""
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      fname = "%s/test" % self.ph_dirs[phase]
      f = open(fname, "w")
      f.close()
      self.torm.append((fname, False))
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
                           [(self._rname(fname), HKR_SKIP, "")])

  def testSkipInvalidName(self):
    """Test skip script with invalid name"""
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      fname = "%s/a.off" % self.ph_dirs[phase]
      f = open(fname, "w")
      f.write("#!/bin/sh\nexit 0\n")
      f.close()
      os.chmod(fname, 0700)
      self.torm.append((fname, False))
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
                           [(self._rname(fname), HKR_SKIP, "")])

  def testSkipDir(self):
    """Test skip directory"""
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      fname = "%s/testdir" % self.ph_dirs[phase]
      os.mkdir(fname)
      self.torm.append((fname, True))
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
                           [(self._rname(fname), HKR_SKIP, "")])

  def testSuccess(self):
    """Test success execution"""
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      fname = "%s/success" % self.ph_dirs[phase]
      f = open(fname, "w")
      f.write("#!/bin/sh\nexit 0\n")
      f.close()
      self.torm.append((fname, False))
      os.chmod(fname, 0700)
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
                           [(self._rname(fname), HKR_SUCCESS, "")])

  def testSymlink(self):
    """Test running a symlink"""
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      fname = "%s/success" % self.ph_dirs[phase]
      os.symlink("/bin/true", fname)
      self.torm.append((fname, False))
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
                           [(self._rname(fname), HKR_SUCCESS, "")])

  def testFail(self):
    """Test success execution"""
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      fname = "%s/success" % self.ph_dirs[phase]
      f = open(fname, "w")
      f.write("#!/bin/sh\nexit 1\n")
      f.close()
      self.torm.append((fname, False))
      os.chmod(fname, 0700)
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
                           [(self._rname(fname), HKR_FAIL, "")])

  def testCombined(self):
    """Test success, failure and skip all in one test"""
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      expect = []
      for fbase, ecode, rs in [("00succ", 0, HKR_SUCCESS),
                               ("10fail", 1, HKR_FAIL),
                               ("20inv.", 0, HKR_SKIP),
                               ]:
        fname = "%s/%s" % (self.ph_dirs[phase], fbase)
        f = open(fname, "w")
        f.write("#!/bin/sh\nexit %d\n" % ecode)
        f.close()
        self.torm.append((fname, False))
        os.chmod(fname, 0700)
        expect.append((self._rname(fname), rs, ""))
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), expect)

  def testOrdering(self):
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      expect = []
      for fbase in ["10s1",
                    "00s0",
                    "10sa",
                    "80sc",
                    "60sd",
                    ]:
        fname = "%s/%s" % (self.ph_dirs[phase], fbase)
        os.symlink("/bin/true", fname)
        self.torm.append((fname, False))
        expect.append((self._rname(fname), HKR_SUCCESS, ""))
      expect.sort()
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), expect)

  def testEnv(self):
    """Test environment execution"""
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      fbase = "success"
      fname = "%s/%s" % (self.ph_dirs[phase], fbase)
      os.symlink("/usr/bin/env", fname)
      self.torm.append((fname, False))
      env_snt = {"PHASE": phase}
      env_exp = "PHASE=%s" % phase
      self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, env_snt),
                           [(self._rname(fname), HKR_SUCCESS, env_exp)])


def FakeHooksRpcSuccess(node_list, hpath, phase, env):
  """Fake call_hooks_runner function.

  @rtype: dict of node -> L{rpc.RpcResult} with a successful script result
  @return: script execution from all nodes

  """
  rr = rpc.RpcResult
  return dict([(node, rr((True, [("utest", constants.HKR_SUCCESS, "ok")]),
                         node=node, call="FakeScriptOk"))
               for node in node_list])


class TestHooksMaster(unittest.TestCase):
  """Testing case for HooksMaster"""

  def _call_false(*args):
    """Fake call_hooks_runner function which returns False."""
    return False

  @staticmethod
  def _call_nodes_false(node_list, hpath, phase, env):
    """Fake call_hooks_runner function.

    @rtype: dict of node -> L{rpc.RpcResult} with an rpc error
    @return: rpc failure from all nodes

    """
    return dict([(node, rpc.RpcResult("error", failed=True,
                  node=node, call="FakeError")) for node in node_list])

  @staticmethod
  def _call_script_fail(node_list, hpath, phase, env):
    """Fake call_hooks_runner function.

    @rtype: dict of node -> L{rpc.RpcResult} with a failed script result
    @return: script execution failure from all nodes

    """
    rr = rpc.RpcResult
    return dict([(node, rr((True, [("utest", constants.HKR_FAIL, "err")]),
                           node=node, call="FakeScriptFail"))
                  for node in node_list])

  def setUp(self):
    self.op = opcodes.OpCode()
    # WARNING: here we pass None as RpcRunner instance since we know
    # our usage via HooksMaster will not use lu.rpc
    self.lu = FakeLU(FakeProc(), self.op, FakeConfig(),
                     None, (123, "/foo/bar"), None)

  def testTotalFalse(self):
    """Test complete rpc failure"""
    hm = hooksmaster.HooksMaster.BuildFromLu(self._call_false, self.lu)
    self.failUnlessRaises(errors.HooksFailure,
                          hm.RunPhase, constants.HOOKS_PHASE_PRE)
    hm.RunPhase(constants.HOOKS_PHASE_POST)

  def testIndividualFalse(self):
    """Test individual node failure"""
    hm = hooksmaster.HooksMaster.BuildFromLu(self._call_nodes_false, self.lu)
    hm.RunPhase(constants.HOOKS_PHASE_PRE)
    #self.failUnlessRaises(errors.HooksFailure,
    #                      hm.RunPhase, constants.HOOKS_PHASE_PRE)
    hm.RunPhase(constants.HOOKS_PHASE_POST)

  def testScriptFalse(self):
    """Test individual rpc failure"""
    hm = hooksmaster.HooksMaster.BuildFromLu(self._call_script_fail, self.lu)
    self.failUnlessRaises(errors.HooksAbort,
                          hm.RunPhase, constants.HOOKS_PHASE_PRE)
    hm.RunPhase(constants.HOOKS_PHASE_POST)

  def testScriptSucceed(self):
    """Test individual rpc failure"""
    hm = hooksmaster.HooksMaster.BuildFromLu(FakeHooksRpcSuccess, self.lu)
    for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
      hm.RunPhase(phase)


class FakeEnvLU(cmdlib.LogicalUnit):
  HPATH = "env_test_lu"
  HTYPE = constants.HTYPE_GROUP

  def __init__(self, *args):
    cmdlib.LogicalUnit.__init__(self, *args)
    self.hook_env = None

  def BuildHooksEnv(self):
    assert self.hook_env is not None
    return self.hook_env

  def BuildHooksNodes(self):
    return (["a"], ["a"])


class FakeNoHooksLU(cmdlib.NoHooksLU):
  pass


class TestHooksRunnerEnv(unittest.TestCase):
  def setUp(self):
    self._rpcs = []

    self.op = opcodes.OpTestDummy(result=False, messages=[], fail=False)
    self.lu = FakeEnvLU(FakeProc(), self.op, FakeContext(), None)

  def _HooksRpc(self, *args):
    self._rpcs.append(args)
    return FakeHooksRpcSuccess(*args)

  def _CheckEnv(self, env, phase, hpath):
    self.assertTrue(env["PATH"].startswith("/sbin"))
    self.assertEqual(env["GANETI_HOOKS_PHASE"], phase)
    self.assertEqual(env["GANETI_HOOKS_PATH"], hpath)
    self.assertEqual(env["GANETI_OP_CODE"], self.op.OP_ID)
    self.assertEqual(env["GANETI_HOOKS_VERSION"], str(constants.HOOKS_VERSION))
    self.assertEqual(env["GANETI_DATA_DIR"], pathutils.DATA_DIR)
    if "GANETI_OBJECT_TYPE" in env:
      self.assertEqual(env["GANETI_OBJECT_TYPE"], constants.HTYPE_GROUP)
    else:
      self.assertTrue(self.lu.HTYPE is None)

  def testEmptyEnv(self):
    # Check pre-phase hook
    self.lu.hook_env = {}
    hm = hooksmaster.HooksMaster.BuildFromLu(self._HooksRpc, self.lu)
    hm.RunPhase(constants.HOOKS_PHASE_PRE)

    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(node_list, set(["node_a.example.com"]))
    self.assertEqual(hpath, self.lu.HPATH)
    self.assertEqual(phase, constants.HOOKS_PHASE_PRE)
    self._CheckEnv(env, constants.HOOKS_PHASE_PRE, self.lu.HPATH)

    # Check post-phase hook
    self.lu.hook_env = {}
    hm.RunPhase(constants.HOOKS_PHASE_POST)

    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(node_list, set(["node_a.example.com"]))
    self.assertEqual(hpath, self.lu.HPATH)
    self.assertEqual(phase, constants.HOOKS_PHASE_POST)
    self._CheckEnv(env, constants.HOOKS_PHASE_POST, self.lu.HPATH)

    self.assertRaises(IndexError, self._rpcs.pop)

  def testEnv(self):
    # Check pre-phase hook
    self.lu.hook_env = {
      "FOO": "pre-foo-value",
      }
    hm = hooksmaster.HooksMaster.BuildFromLu(self._HooksRpc, self.lu)
    hm.RunPhase(constants.HOOKS_PHASE_PRE)

    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(node_list, set(["node_a.example.com"]))
    self.assertEqual(hpath, self.lu.HPATH)
    self.assertEqual(phase, constants.HOOKS_PHASE_PRE)
    self.assertEqual(env["GANETI_FOO"], "pre-foo-value")
    self.assertFalse(compat.any(key.startswith("GANETI_POST") for key in env))
    self._CheckEnv(env, constants.HOOKS_PHASE_PRE, self.lu.HPATH)

    # Check post-phase hook
    self.lu.hook_env = {
      "FOO": "post-value",
      "BAR": 123,
      }
    hm.RunPhase(constants.HOOKS_PHASE_POST)

    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(node_list, set(["node_a.example.com"]))
    self.assertEqual(hpath, self.lu.HPATH)
    self.assertEqual(phase, constants.HOOKS_PHASE_POST)
    self.assertEqual(env["GANETI_FOO"], "pre-foo-value")
    self.assertEqual(env["GANETI_POST_FOO"], "post-value")
    self.assertEqual(env["GANETI_POST_BAR"], "123")
    self.assertFalse("GANETI_BAR" in env)
    self._CheckEnv(env, constants.HOOKS_PHASE_POST, self.lu.HPATH)

    self.assertRaises(IndexError, self._rpcs.pop)

    # Check configuration update hook
    hm.RunConfigUpdate()
    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(set(node_list), set([self.lu.cfg.GetMasterNodeName()]))
    self.assertEqual(hpath, constants.HOOKS_NAME_CFGUPDATE)
    self.assertEqual(phase, constants.HOOKS_PHASE_POST)
    self._CheckEnv(env, constants.HOOKS_PHASE_POST,
                   constants.HOOKS_NAME_CFGUPDATE)
    self.assertFalse(compat.any(key.startswith("GANETI_POST") for key in env))
    self.assertEqual(env["GANETI_FOO"], "pre-foo-value")
    self.assertRaises(IndexError, self._rpcs.pop)

  def testConflict(self):
    for name in ["DATA_DIR", "OP_CODE"]:
      self.lu.hook_env = { name: "value" }

      # Test using a clean HooksMaster instance
      hm = hooksmaster.HooksMaster.BuildFromLu(self._HooksRpc, self.lu)

      for phase in [constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST]:
        self.assertRaises(AssertionError, hm.RunPhase, phase)
        self.assertRaises(IndexError, self._rpcs.pop)

  def testNoNodes(self):
    self.lu.hook_env = {}
    hm = hooksmaster.HooksMaster.BuildFromLu(self._HooksRpc, self.lu)
    hm.RunPhase(constants.HOOKS_PHASE_PRE, node_names=[])
    self.assertRaises(IndexError, self._rpcs.pop)

  def testSpecificNodes(self):
    self.lu.hook_env = {}

    nodes = [
      "node1.example.com",
      "node93782.example.net",
      ]

    hm = hooksmaster.HooksMaster.BuildFromLu(self._HooksRpc, self.lu)

    for phase in [constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST]:
      hm.RunPhase(phase, node_names=nodes)

      (node_list, hpath, rpc_phase, env) = self._rpcs.pop(0)
      self.assertEqual(set(node_list), set(nodes))
      self.assertEqual(hpath, self.lu.HPATH)
      self.assertEqual(rpc_phase, phase)
      self._CheckEnv(env, phase, self.lu.HPATH)

      self.assertRaises(IndexError, self._rpcs.pop)

  def testRunConfigUpdateNoPre(self):
    self.lu.hook_env = {
      "FOO": "value",
      }

    hm = hooksmaster.HooksMaster.BuildFromLu(self._HooksRpc, self.lu)
    hm.RunConfigUpdate()

    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(set(node_list), set([self.lu.cfg.GetMasterNodeName()]))
    self.assertEqual(hpath, constants.HOOKS_NAME_CFGUPDATE)
    self.assertEqual(phase, constants.HOOKS_PHASE_POST)
    self.assertEqual(env["GANETI_FOO"], "value")
    self.assertFalse(compat.any(key.startswith("GANETI_POST") for key in env))
    self._CheckEnv(env, constants.HOOKS_PHASE_POST,
                   constants.HOOKS_NAME_CFGUPDATE)

    self.assertRaises(IndexError, self._rpcs.pop)

  def testNoPreBeforePost(self):
    self.lu.hook_env = {
      "FOO": "value",
      }

    hm = hooksmaster.HooksMaster.BuildFromLu(self._HooksRpc, self.lu)
    hm.RunPhase(constants.HOOKS_PHASE_POST)

    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(node_list, set(["node_a.example.com"]))
    self.assertEqual(hpath, self.lu.HPATH)
    self.assertEqual(phase, constants.HOOKS_PHASE_POST)
    self.assertEqual(env["GANETI_FOO"], "value")
    self.assertEqual(env["GANETI_POST_FOO"], "value")
    self._CheckEnv(env, constants.HOOKS_PHASE_POST, self.lu.HPATH)

    self.assertRaises(IndexError, self._rpcs.pop)

  def testNoHooksLU(self):
    self.lu = FakeNoHooksLU(FakeProc(), self.op, FakeContext(), None)
    self.assertRaises(AssertionError, self.lu.BuildHooksEnv)
    self.assertRaises(AssertionError, self.lu.BuildHooksNodes)

    hm = hooksmaster.HooksMaster.BuildFromLu(self._HooksRpc, self.lu)
    self.assertEqual(hm.pre_env, {})
    self.assertRaises(IndexError, self._rpcs.pop)

    hm.RunPhase(constants.HOOKS_PHASE_PRE)
    self.assertRaises(IndexError, self._rpcs.pop)

    hm.RunPhase(constants.HOOKS_PHASE_POST)
    self.assertRaises(IndexError, self._rpcs.pop)

    hm.RunConfigUpdate()

    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(set(node_list), set([self.lu.cfg.GetMasterNodeName()]))
    self.assertEqual(hpath, constants.HOOKS_NAME_CFGUPDATE)
    self.assertEqual(phase, constants.HOOKS_PHASE_POST)
    self.assertFalse(compat.any(key.startswith("GANETI_POST") for key in env))
    self._CheckEnv(env, constants.HOOKS_PHASE_POST,
                   constants.HOOKS_NAME_CFGUPDATE)
    self.assertRaises(IndexError, self._rpcs.pop)

    assert isinstance(self.lu, FakeNoHooksLU), "LU was replaced"


class FakeEnvWithCustomPostHookNodesLU(cmdlib.LogicalUnit):
  HPATH = "env_test_lu"
  HTYPE = constants.HTYPE_GROUP

  def __init__(self, *args):
    cmdlib.LogicalUnit.__init__(self, *args)

  def BuildHooksEnv(self):
    return {}

  def BuildHooksNodes(self):
    return (["a"], ["a"])

  def PreparePostHookNodes(self, post_hook_node_uuids):
    return post_hook_node_uuids + ["b"]


class TestHooksRunnerEnv(unittest.TestCase):
  def setUp(self):
    self._rpcs = []

    self.op = opcodes.OpTestDummy(result=False, messages=[], fail=False)
    self.lu = FakeEnvWithCustomPostHookNodesLU(FakeProc(), self.op,
                                               FakeConfig(),
                                               None,
                                               (123, "/foo/bar"),
                                               None)

  def _HooksRpc(self, *args):
    self._rpcs.append(args)
    return FakeHooksRpcSuccess(*args)

  def testEmptyEnv(self):
    # Check pre-phase hook
    hm = hooksmaster.HooksMaster.BuildFromLu(self._HooksRpc, self.lu)
    hm.RunPhase(constants.HOOKS_PHASE_PRE)

    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(node_list, set(["a"]))

    # Check post-phase hook
    hm.RunPhase(constants.HOOKS_PHASE_POST)

    (node_list, hpath, phase, env) = self._rpcs.pop(0)
    self.assertEqual(node_list, set(["a", "b"]))

    self.assertRaises(IndexError, self._rpcs.pop)


if __name__ == "__main__":
  testutils.GanetiTestProgram()