#!/usr/bin/python
#

# Copyright (C) 2006, 2007, 2008, 2010, 2012, 2013 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 objects module"""


import copy
import pprint
import unittest

from ganeti import constants
from ganeti import objects
from ganeti import errors
from ganeti import serializer

import testutils


class SimpleObject(objects.ConfigObject):
  __slots__ = ["a", "b"]


class TestDictState(unittest.TestCase):
  """Simple dict tansformation tests"""

  def testSimpleObjectToDict(self):
    o1 = SimpleObject(a="1")
    self.assertEquals(o1.ToDict(), {"a": "1"})
    self.assertEquals(o1.__getstate__(), {"a": "1"})
    self.assertEquals(o1.__getstate__(), o1.ToDict())
    o1.a = 2
    o1.b = 5
    self.assertEquals(o1.ToDict(), {"a": 2, "b": 5})
    o2 = SimpleObject.FromDict(o1.ToDict())
    self.assertEquals(o1.ToDict(), {"a": 2, "b": 5})


class TestClusterObject(unittest.TestCase):
  """Tests done on a L{objects.Cluster}"""

  def setUp(self):
    hvparams = {
      constants.HT_FAKE: {
        "foo": "bar",
        "bar": "foo",
        "foobar": "barfoo",
        },
      }
    os_hvp = {
      "lenny-image": {
        constants.HT_FAKE: {
          "foo": "baz",
          "foobar": "foobar",
          "blah": "blibb",
          "blubb": "blah",
          },
        constants.HT_XEN_PVM: {
          "root_path": "/dev/sda5",
          "foo": "foobar",
          },
        },
      "ubuntu-hardy": {
        },
      }
    ndparams = {
        constants.ND_OOB_PROGRAM: "/bin/cluster-oob",
        constants.ND_SPINDLE_COUNT: 1,
        constants.ND_EXCLUSIVE_STORAGE: False,
        }

    self.fake_cl = objects.Cluster(hvparams=hvparams, os_hvp=os_hvp,
                                   ndparams=ndparams)
    self.fake_cl.UpgradeConfig()

  def testGetHVDefaults(self):
    cl = self.fake_cl
    self.failUnlessEqual(cl.GetHVDefaults(constants.HT_FAKE),
                         cl.hvparams[constants.HT_FAKE])
    self.failUnlessEqual(cl.GetHVDefaults(None), {})
    defaults = cl.GetHVDefaults(constants.HT_XEN_PVM,
                                          os_name="lenny-image")
    for param, value in cl.os_hvp["lenny-image"][constants.HT_XEN_PVM].items():
      self.assertEqual(value, defaults[param])

  def testFillHvFullMerge(self):
    inst_hvparams = {
      "blah": "blubb",
      }

    fake_dict = constants.HVC_DEFAULTS[constants.HT_FAKE].copy()
    fake_dict.update({
      "foo": "baz",
      "bar": "foo",
      "foobar": "foobar",
      "blah": "blubb",
      "blubb": "blah",
      })
    fake_inst = objects.Instance(name="foobar",
                                 os="lenny-image",
                                 hypervisor=constants.HT_FAKE,
                                 hvparams=inst_hvparams)
    self.assertEqual(fake_dict, self.fake_cl.FillHV(fake_inst))

  def testFillHvGlobalParams(self):
    fake_inst = objects.Instance(name="foobar",
                                 os="ubuntu-hardy",
                                 hypervisor=constants.HT_FAKE,
                                 hvparams={})
    self.assertEqual(self.fake_cl.hvparams[constants.HT_FAKE],
                     self.fake_cl.FillHV(fake_inst))

  def testFillHvInstParams(self):
    inst_hvparams = {
      "blah": "blubb",
      }
    fake_inst = objects.Instance(name="foobar",
                                 os="ubuntu-hardy",
                                 hypervisor=constants.HT_XEN_PVM,
                                 hvparams=inst_hvparams)
    filled_conf = self.fake_cl.FillHV(fake_inst)
    for param, value in constants.HVC_DEFAULTS[constants.HT_XEN_PVM].items():
      if param == "blah":
        value = "blubb"
      self.assertEqual(value, filled_conf[param])

  def testFillHvDefaultParams(self):
    fake_inst = objects.Instance(name="foobar",
                                 os="ubuntu-hardy",
                                 hypervisor=constants.HT_XEN_PVM,
                                 hvparams={})
    self.assertEqual(constants.HVC_DEFAULTS[constants.HT_XEN_PVM],
                     self.fake_cl.FillHV(fake_inst))

  def testFillHvPartialParams(self):
    os = "lenny-image"
    fake_inst = objects.Instance(name="foobar",
                                 os=os,
                                 hypervisor=constants.HT_XEN_PVM,
                                 hvparams={})
    filled_conf = self.fake_cl.FillHV(fake_inst)
    for param, value in self.fake_cl.os_hvp[os][constants.HT_XEN_PVM].items():
      self.assertEqual(value, filled_conf[param])

  def testFillNdParamsCluster(self):
    fake_node = objects.Node(name="test",
                             ndparams={},
                             group="testgroup")
    fake_group = objects.NodeGroup(name="testgroup",
                                   ndparams={})
    self.assertEqual(self.fake_cl.ndparams,
                     self.fake_cl.FillND(fake_node, fake_group))

  def testFillNdParamsNodeGroup(self):
    fake_node = objects.Node(name="test",
                             ndparams={},
                             group="testgroup")
    group_ndparams = {
        constants.ND_OOB_PROGRAM: "/bin/group-oob",
        constants.ND_SPINDLE_COUNT: 10,
        constants.ND_EXCLUSIVE_STORAGE: True,
        constants.ND_OVS: True,
        constants.ND_OVS_LINK: "eth2",
        constants.ND_OVS_NAME: "openvswitch",
        constants.ND_SSH_PORT: 122,
        constants.ND_CPU_SPEED: 1.1,
        }
    fake_group = objects.NodeGroup(name="testgroup",
                                   ndparams=group_ndparams)
    self.assertEqual(group_ndparams,
                     self.fake_cl.FillND(fake_node, fake_group))

  def testFillNdParamsNode(self):
    node_ndparams = {
        constants.ND_OOB_PROGRAM: "/bin/node-oob",
        constants.ND_SPINDLE_COUNT: 2,
        constants.ND_EXCLUSIVE_STORAGE: True,
        constants.ND_OVS: True,
        constants.ND_OVS_LINK: "eth2",
        constants.ND_OVS_NAME: "openvswitch",
        constants.ND_SSH_PORT: 222,
        constants.ND_CPU_SPEED: 1.1,
        }
    fake_node = objects.Node(name="test",
                             ndparams=node_ndparams,
                             group="testgroup")
    fake_group = objects.NodeGroup(name="testgroup",
                                   ndparams={})
    self.assertEqual(node_ndparams,
                     self.fake_cl.FillND(fake_node, fake_group))

  def testFillNdParamsAll(self):
    node_ndparams = {
        constants.ND_OOB_PROGRAM: "/bin/node-oob",
        constants.ND_SPINDLE_COUNT: 5,
        constants.ND_EXCLUSIVE_STORAGE: True,
        constants.ND_OVS: True,
        constants.ND_OVS_LINK: "eth2",
        constants.ND_OVS_NAME: "openvswitch",
        constants.ND_SSH_PORT: 322,
        constants.ND_CPU_SPEED: 1.1,
        }
    fake_node = objects.Node(name="test",
                             ndparams=node_ndparams,
                             group="testgroup")
    group_ndparams = {
        constants.ND_OOB_PROGRAM: "/bin/group-oob",
        constants.ND_SPINDLE_COUNT: 4,
        constants.ND_SSH_PORT: 422,
        }
    fake_group = objects.NodeGroup(name="testgroup",
                                   ndparams=group_ndparams)
    self.assertEqual(node_ndparams,
                     self.fake_cl.FillND(fake_node, fake_group))

  def testPrimaryHypervisor(self):
    assert self.fake_cl.enabled_hypervisors is None
    self.fake_cl.enabled_hypervisors = [constants.HT_XEN_HVM]
    self.assertEqual(self.fake_cl.primary_hypervisor, constants.HT_XEN_HVM)

    self.fake_cl.enabled_hypervisors = [constants.HT_XEN_PVM, constants.HT_KVM]
    self.assertEqual(self.fake_cl.primary_hypervisor, constants.HT_XEN_PVM)

    self.fake_cl.enabled_hypervisors = sorted(constants.HYPER_TYPES)
    self.assertEqual(self.fake_cl.primary_hypervisor, constants.HT_CHROOT)

  def testUpgradeConfig(self):
    # FIXME: This test is incomplete
    cluster = objects.Cluster()
    cluster.UpgradeConfig()
    cluster = objects.Cluster(ipolicy={"unknown_key": None})
    self.assertRaises(errors.ConfigurationError, cluster.UpgradeConfig)

  def testUpgradeEnabledDiskTemplates(self):
    cfg = objects.ConfigData()
    cfg.cluster = objects.Cluster()
    cfg.cluster.volume_group_name = "myvg"
    instance1 = objects.Instance()
    instance1.disk_template = constants.DT_DISKLESS
    instance2 = objects.Instance()
    instance2.disk_template = constants.DT_RBD
    cfg.instances = { "myinstance1": instance1, "myinstance2": instance2 }
    disk2 = objects.Disk(dev_type=constants.DT_RBD)
    cfg.disks = { "pinkbunnydisk": disk2 }
    nodegroup = objects.NodeGroup()
    nodegroup.ipolicy = {}
    nodegroup.ipolicy[constants.IPOLICY_DTS] = [instance1.disk_template, \
      constants.DT_BLOCK]
    cfg.cluster.ipolicy = {}
    cfg.cluster.ipolicy[constants.IPOLICY_DTS] = \
      [constants.DT_EXT, constants.DT_DISKLESS]
    cfg.nodegroups = { "mynodegroup": nodegroup }
    cfg._UpgradeEnabledDiskTemplates()
    expected_disk_templates = [constants.DT_DRBD8,
                               constants.DT_PLAIN,
                               instance1.disk_template,
                               instance2.disk_template]
    self.assertEqual(set(expected_disk_templates),
                     set(cfg.cluster.enabled_disk_templates))
    self.assertEqual(set([instance1.disk_template]),
                     set(cfg.cluster.ipolicy[constants.IPOLICY_DTS]))


class TestClusterObjectTcpUdpPortPool(unittest.TestCase):
  def testNewCluster(self):
    self.assertTrue(objects.Cluster().tcpudp_port_pool is None)

  def testSerializingEmpty(self):
    self.assertEqual(objects.Cluster().ToDict(), {
      "tcpudp_port_pool": [],
      })

  def testSerializing(self):
    cluster = objects.Cluster.FromDict({})
    self.assertEqual(cluster.tcpudp_port_pool, set())

    cluster.tcpudp_port_pool.add(3546)
    cluster.tcpudp_port_pool.add(62511)

    data = cluster.ToDict()
    self.assertEqual(data.keys(), ["tcpudp_port_pool"])
    self.assertEqual(sorted(data["tcpudp_port_pool"]), sorted([3546, 62511]))

  def testDeserializingEmpty(self):
    cluster = objects.Cluster.FromDict({})
    self.assertEqual(cluster.tcpudp_port_pool, set())

  def testDeserialize(self):
    cluster = objects.Cluster.FromDict({
      "tcpudp_port_pool": [26214, 10039, 267],
      })
    self.assertEqual(cluster.tcpudp_port_pool, set([26214, 10039, 267]))


class TestOS(unittest.TestCase):
  ALL_DATA = [
    "debootstrap",
    "debootstrap+default",
    "debootstrap++default",
    ]

  def testSplitNameVariant(self):
    for name in self.ALL_DATA:
      self.assertEqual(len(objects.OS.SplitNameVariant(name)), 2)

  def testVariant(self):
    self.assertEqual(objects.OS.GetVariant("debootstrap"), "")
    self.assertEqual(objects.OS.GetVariant("debootstrap+default"), "default")


class TestInstance(unittest.TestCase):
  def testFindDisk(self):
    inst = objects.Instance(name="fakeinstdrbd.example.com",
      primary_node="node20.example.com",
      disks=[
        objects.Disk(dev_type=constants.DT_DRBD8, size=786432,
          logical_id=("node20.example.com", "node15.example.com",
                      12300, 0, 0, "secret"),
          children=[
            objects.Disk(dev_type=constants.DT_PLAIN, size=786432,
                         logical_id=("myxenvg", "disk0")),
            objects.Disk(dev_type=constants.DT_PLAIN, size=128,
                         logical_id=("myxenvg", "meta0"))
          ],
          iv_name="disk/0")
        ])

    self.assertEqual(inst.FindDisk(0), inst.disks[0])
    self.assertRaises(errors.OpPrereqError, inst.FindDisk, "hello")
    self.assertRaises(errors.OpPrereqError, inst.FindDisk, 100)
    self.assertRaises(errors.OpPrereqError, inst.FindDisk, 1)


class TestNode(unittest.TestCase):
  def testEmpty(self):
    self.assertEqual(objects.Node().ToDict(), {})
    self.assertTrue(isinstance(objects.Node.FromDict({}), objects.Node))

  def testHvState(self):
    node = objects.Node(name="node18157.example.com", hv_state={
      constants.HT_XEN_HVM: objects.NodeHvState(cpu_total=64),
      constants.HT_KVM: objects.NodeHvState(cpu_node=1),
      })

    node2 = objects.Node.FromDict(node.ToDict())

    # Make sure nothing can reference it anymore
    del node

    self.assertEqual(node2.name, "node18157.example.com")
    self.assertEqual(frozenset(node2.hv_state), frozenset([
      constants.HT_XEN_HVM,
      constants.HT_KVM,
      ]))
    self.assertEqual(node2.hv_state[constants.HT_KVM].cpu_node, 1)
    self.assertEqual(node2.hv_state[constants.HT_XEN_HVM].cpu_total, 64)

  def testDiskState(self):
    node = objects.Node(name="node32087.example.com", disk_state={
      constants.DT_PLAIN: {
        "lv32352": objects.NodeDiskState(total=128),
        "lv2082": objects.NodeDiskState(total=512),
        },
      })

    node2 = objects.Node.FromDict(node.ToDict())

    # Make sure nothing can reference it anymore
    del node

    self.assertEqual(node2.name, "node32087.example.com")
    self.assertEqual(frozenset(node2.disk_state), frozenset([
      constants.DT_PLAIN,
      ]))
    self.assertEqual(frozenset(node2.disk_state[constants.DT_PLAIN]),
                     frozenset(["lv32352", "lv2082"]))
    self.assertEqual(node2.disk_state[constants.DT_PLAIN]["lv2082"].total, 512)
    self.assertEqual(node2.disk_state[constants.DT_PLAIN]["lv32352"].total, 128)

  def testFilterEsNdp(self):
    node1 = objects.Node(name="node11673.example.com", ndparams={
      constants.ND_EXCLUSIVE_STORAGE: True,
      })
    node2 = objects.Node(name="node11674.example.com", ndparams={
      constants.ND_SPINDLE_COUNT: 3,
      constants.ND_EXCLUSIVE_STORAGE: False,
      })
    self.assertTrue(constants.ND_EXCLUSIVE_STORAGE in node1.ndparams)
    node1.UpgradeConfig()
    self.assertFalse(constants.ND_EXCLUSIVE_STORAGE in node1.ndparams)
    self.assertTrue(constants.ND_EXCLUSIVE_STORAGE in node2.ndparams)
    self.assertTrue(constants.ND_SPINDLE_COUNT in node2.ndparams)
    node2.UpgradeConfig()
    self.assertFalse(constants.ND_EXCLUSIVE_STORAGE in node2.ndparams)
    self.assertTrue(constants.ND_SPINDLE_COUNT in node2.ndparams)


class TestInstancePolicy(unittest.TestCase):
  def setUp(self):
    # Policies are big, and we want to see the difference in case of an error
    self.maxDiff = None

  def _AssertIPolicyIsFull(self, policy):
    self.assertEqual(frozenset(policy.keys()), constants.IPOLICY_ALL_KEYS)
    self.assertTrue(len(policy[constants.ISPECS_MINMAX]) > 0)
    for minmax in policy[constants.ISPECS_MINMAX]:
      self.assertEqual(frozenset(minmax.keys()), constants.ISPECS_MINMAX_KEYS)
      for key in constants.ISPECS_MINMAX_KEYS:
        self.assertEqual(frozenset(minmax[key].keys()),
                         constants.ISPECS_PARAMETERS)
    self.assertEqual(frozenset(policy[constants.ISPECS_STD].keys()),
                     constants.ISPECS_PARAMETERS)

  def testDefaultIPolicy(self):
    objects.InstancePolicy.CheckParameterSyntax(constants.IPOLICY_DEFAULTS,
                                                True)
    self._AssertIPolicyIsFull(constants.IPOLICY_DEFAULTS)

  def _AssertPolicyIsBad(self, ipolicy, do_check_std=None):
    if do_check_std is None:
      check_std_vals = [False, True]
    else:
      check_std_vals = [do_check_std]
    for check_std in check_std_vals:
      self.assertRaises(errors.ConfigurationError,
                        objects.InstancePolicy.CheckISpecSyntax,
                        ipolicy, check_std)

  def testCheckISpecSyntax(self):
    default_stdspec = constants.IPOLICY_DEFAULTS[constants.ISPECS_STD]
    incomplete_ipolicies = [
      {
         constants.ISPECS_MINMAX: [],
         constants.ISPECS_STD: default_stdspec,
         },
      {
         constants.ISPECS_MINMAX: [{}],
         constants.ISPECS_STD: default_stdspec,
         },
      {
        constants.ISPECS_MINMAX: [{
          constants.ISPECS_MIN: NotImplemented,
          }],
        constants.ISPECS_STD: default_stdspec,
        },
      {
        constants.ISPECS_MINMAX: [{
          constants.ISPECS_MAX: NotImplemented,
          }],
        constants.ISPECS_STD: default_stdspec,
        },
      {
        constants.ISPECS_MINMAX: [{
          constants.ISPECS_MIN: NotImplemented,
          constants.ISPECS_MAX: NotImplemented,
          }],
        },
      ]
    for ipol in incomplete_ipolicies:
      self.assertRaises(errors.ConfigurationError,
                        objects.InstancePolicy.CheckISpecSyntax,
                        ipol, True)
      oldminmax = ipol[constants.ISPECS_MINMAX]
      if oldminmax:
        # Prepending valid specs shouldn't change the error
        ipol[constants.ISPECS_MINMAX] = ([constants.ISPECS_MINMAX_DEFAULTS] +
                                         oldminmax)
        self.assertRaises(errors.ConfigurationError,
                          objects.InstancePolicy.CheckISpecSyntax,
                          ipol, True)

    good_ipolicy = {
      constants.ISPECS_MINMAX: [
        {
          constants.ISPECS_MIN: {
            constants.ISPEC_MEM_SIZE: 64,
            constants.ISPEC_CPU_COUNT: 1,
            constants.ISPEC_DISK_COUNT: 2,
            constants.ISPEC_DISK_SIZE: 64,
            constants.ISPEC_NIC_COUNT: 1,
            constants.ISPEC_SPINDLE_USE: 1,
            },
          constants.ISPECS_MAX: {
            constants.ISPEC_MEM_SIZE: 16384,
            constants.ISPEC_CPU_COUNT: 5,
            constants.ISPEC_DISK_COUNT: 12,
            constants.ISPEC_DISK_SIZE: 1024,
            constants.ISPEC_NIC_COUNT: 9,
            constants.ISPEC_SPINDLE_USE: 18,
            },
          },
        {
          constants.ISPECS_MIN: {
            constants.ISPEC_MEM_SIZE: 32768,
            constants.ISPEC_CPU_COUNT: 8,
            constants.ISPEC_DISK_COUNT: 1,
            constants.ISPEC_DISK_SIZE: 1024,
            constants.ISPEC_NIC_COUNT: 1,
            constants.ISPEC_SPINDLE_USE: 1,
            },
          constants.ISPECS_MAX: {
            constants.ISPEC_MEM_SIZE: 65536,
            constants.ISPEC_CPU_COUNT: 10,
            constants.ISPEC_DISK_COUNT: 5,
            constants.ISPEC_DISK_SIZE: 1024 * 1024,
            constants.ISPEC_NIC_COUNT: 3,
            constants.ISPEC_SPINDLE_USE: 12,
            },
          },
        ],
      }
    good_ipolicy[constants.ISPECS_STD] = copy.deepcopy(
      good_ipolicy[constants.ISPECS_MINMAX][0][constants.ISPECS_MAX])
    # Check that it's really good before making it bad
    objects.InstancePolicy.CheckISpecSyntax(good_ipolicy, True)

    bad_ipolicy = copy.deepcopy(good_ipolicy)
    for minmax in bad_ipolicy[constants.ISPECS_MINMAX]:
      for (key, spec) in minmax.items():
        for param in spec:
          oldv = spec[param]
          del spec[param]
          self._AssertPolicyIsBad(bad_ipolicy)
          if key == constants.ISPECS_MIN:
            spec[param] = minmax[constants.ISPECS_MAX][param] + 1
          self._AssertPolicyIsBad(bad_ipolicy)
          spec[param] = oldv
    assert bad_ipolicy == good_ipolicy

    stdspec = bad_ipolicy[constants.ISPECS_STD]
    for param in stdspec:
      oldv = stdspec[param]
      del stdspec[param]
      self._AssertPolicyIsBad(bad_ipolicy, True)
      # Note that std spec is the same as a max spec
      stdspec[param] = oldv + 1
      self._AssertPolicyIsBad(bad_ipolicy, True)
      stdspec[param] = oldv
    assert bad_ipolicy == good_ipolicy

    for minmax in good_ipolicy[constants.ISPECS_MINMAX]:
      for spec in minmax.values():
        good_ipolicy[constants.ISPECS_STD] = spec
        objects.InstancePolicy.CheckISpecSyntax(good_ipolicy, True)

  def testCheckISpecParamSyntax(self):
    par = "my_parameter"
    for check_std in [True, False]:
      # Min and max only
      good_values = [(11, 11), (11, 40), (0, 0)]
      for (mn, mx) in good_values:
        minmax = dict((k, {}) for k in constants.ISPECS_MINMAX_KEYS)
        minmax[constants.ISPECS_MIN][par] = mn
        minmax[constants.ISPECS_MAX][par] = mx
        objects.InstancePolicy._CheckISpecParamSyntax(minmax, {}, par,
                                                     check_std)
      minmax = dict((k, {}) for k in constants.ISPECS_MINMAX_KEYS)
      minmax[constants.ISPECS_MIN][par] = 11
      minmax[constants.ISPECS_MAX][par] = 5
      self.assertRaises(errors.ConfigurationError,
                        objects.InstancePolicy._CheckISpecParamSyntax,
                        minmax, {}, par, check_std)
    # Min, std, max
    good_values = [
      (11, 11, 11),
      (11, 11, 40),
      (11, 40, 40),
      ]
    for (mn, st, mx) in good_values:
      minmax = {
        constants.ISPECS_MIN: {par: mn},
        constants.ISPECS_MAX: {par: mx},
        }
      stdspec = {par: st}
      objects.InstancePolicy._CheckISpecParamSyntax(minmax, stdspec, par, True)
    bad_values = [
      (11, 11,  5, True),
      (40, 11, 11, True),
      (11, 80, 40, False),
      (11,  5, 40, False,),
      (11,  5,  5, True),
      (40, 40, 11, True),
      ]
    for (mn, st, mx, excp) in bad_values:
      minmax = {
        constants.ISPECS_MIN: {par: mn},
        constants.ISPECS_MAX: {par: mx},
        }
      stdspec = {par: st}
      if excp:
        self.assertRaises(errors.ConfigurationError,
                          objects.InstancePolicy._CheckISpecParamSyntax,
                          minmax, stdspec, par, True)
      else:
        ret = objects.InstancePolicy._CheckISpecParamSyntax(minmax, stdspec,
                                                            par, True)
        self.assertFalse(ret)

  def testCheckDiskTemplates(self):
    invalid = "this_is_not_a_good_template"
    for dt in constants.DISK_TEMPLATES:
      objects.InstancePolicy.CheckDiskTemplates([dt])
    objects.InstancePolicy.CheckDiskTemplates(list(constants.DISK_TEMPLATES))
    bad_examples = [
      [invalid],
      [constants.DT_DRBD8, invalid],
      list(constants.DISK_TEMPLATES) + [invalid],
      [],
      None,
      ]
    for dtl in bad_examples:
      self.assertRaises(errors.ConfigurationError,
                        objects.InstancePolicy.CheckDiskTemplates,
                        dtl)

  def testCheckParameterSyntax(self):
    invalid = "this_key_shouldnt_be_here"
    for check_std in [True, False]:
      objects.InstancePolicy.CheckParameterSyntax({}, check_std)
      policy = {invalid: None}
      self.assertRaises(errors.ConfigurationError,
                        objects.InstancePolicy.CheckParameterSyntax,
                        policy, check_std)
      for par in constants.IPOLICY_PARAMETERS:
        for val in ("blah", None, {}, [42]):
          policy = {par: val}
          self.assertRaises(errors.ConfigurationError,
                            objects.InstancePolicy.CheckParameterSyntax,
                            policy, check_std)

  def testFillIPolicyEmpty(self):
    policy = objects.FillIPolicy(constants.IPOLICY_DEFAULTS, {})
    objects.InstancePolicy.CheckParameterSyntax(policy, True)
    self.assertEqual(policy, constants.IPOLICY_DEFAULTS)

  def _AssertISpecsMerged(self, default_spec, diff_spec, merged_spec):
    for (param, value) in merged_spec.items():
      if param in diff_spec:
        self.assertEqual(value, diff_spec[param])
      else:
        self.assertEqual(value, default_spec[param])

  def _AssertIPolicyMerged(self, default_pol, diff_pol, merged_pol):
    for (key, value) in merged_pol.items():
      if key in diff_pol:
        if key == constants.ISPECS_STD:
          self._AssertISpecsMerged(default_pol[key], diff_pol[key], value)
        else:
          self.assertEqual(value, diff_pol[key])
      else:
        self.assertEqual(value, default_pol[key])

  def testFillIPolicy(self):
    partial_policies = [
      {constants.IPOLICY_VCPU_RATIO: 3.14},
      {constants.IPOLICY_SPINDLE_RATIO: 2.72},
      {constants.IPOLICY_DTS: [constants.DT_FILE]},
      {constants.ISPECS_STD: {constants.ISPEC_DISK_COUNT: 3}},
      {constants.ISPECS_MINMAX: [constants.ISPECS_MINMAX_DEFAULTS,
                                 constants.ISPECS_MINMAX_DEFAULTS]}
      ]
    for diff_pol in partial_policies:
      policy = objects.FillIPolicy(constants.IPOLICY_DEFAULTS, diff_pol)
      objects.InstancePolicy.CheckParameterSyntax(policy, True)
      self._AssertIPolicyIsFull(policy)
      self._AssertIPolicyMerged(constants.IPOLICY_DEFAULTS, diff_pol, policy)

  def testFillIPolicyKeepsUnknown(self):
    INVALID_KEY = "invalid_ipolicy_key"
    diff_pol = {
      INVALID_KEY: None,
      }
    policy = objects.FillIPolicy(constants.IPOLICY_DEFAULTS, diff_pol)
    self.assertTrue(INVALID_KEY in policy)


class TestDisk(unittest.TestCase):
  def addChild(self, disk):
    """Adds a child of the same device type as the parent."""
    disk.children = []
    child = objects.Disk()
    child.dev_type = disk.dev_type
    disk.children.append(child)

  def testUpgradeConfigDevTypeLegacy(self):
    for old, new in [("drbd8", constants.DT_DRBD8),
                     ("lvm", constants.DT_PLAIN)]:
      disk = objects.Disk()
      disk.dev_type = old
      self.addChild(disk)
      disk.UpgradeConfig()
      self.assertEqual(new, disk.dev_type)
      self.assertEqual(new, disk.children[0].dev_type)

  def testUpgradeConfigDevTypeLegacyUnchanged(self):
    dev_types = [constants.DT_FILE, constants.DT_SHARED_FILE,
                 constants.DT_BLOCK, constants.DT_EXT,
                 constants.DT_RBD, constants.DT_GLUSTER]
    for dev_type in dev_types:
      disk = objects.Disk()
      disk.dev_type = dev_type
      self.addChild(disk)
      disk.UpgradeConfig()
      self.assertEqual(dev_type, disk.dev_type)
      self.assertEqual(dev_type, disk.children[0].dev_type)


class TestSimpleFillOS(unittest.TestCase):
    # We have to make sure that:
    #  * From within the configuration, variants override defaults
    #  * Temporary values override configuration
    #  * No overlap between public, private and secret dicts is allowed
    #
    # As a result, here are the actors in this test:
    #
    # A:  temporary                 public
    # B:  temporary                        private
    # C:  temporary                                secret
    # X:  temporary                 public private secret
    # D:            configuration   public                       variant
    # E:            configuration   public                  base
    # F:            configuration          private               variant
    # G:            configuration          private          base
    #
    # Every time a param is assigned "ERROR", we expect FillOSParams to override
    # it. If it doesn't, it's an error.
    #
    # Every time a param is assigned itself as a value, it's the value we expect
    # FillOSParams to give us back.

    def setUp(self):
      self.fake_cl = objects.Cluster()
      self.fake_cl.UpgradeConfig()
      self.fake_cl.osparams = {"os": {"A": "ERROR",
                                      "D": "ERROR",
                                      "E": "E"},
                               "os+a": {"D": "D"}}
      self.fake_cl.osparams_private_cluster = {"os": {"B": "ERROR",
                                                      "F": "ERROR",
                                                      "G": "G"},
                                               "os+a": {"F": "F"}}

    def testConflictPublicPrivate(self):
      "Make sure we disallow attempts to override params based on visibility."
      public_dict = {"A": "A", "X": "X"}
      private_dict = {"B": "B", "X": "X"}
      secret_dict = {"C": "C"}
      dicts_pp = (public_dict, private_dict)
      dicts_pps = (public_dict, private_dict, secret_dict)

      # Without secret parameters
      self.assertRaises(errors.OpPrereqError,
                        lambda: self.fake_cl.SimpleFillOS("os+a", *dicts_pp))

      # but also with those.
      self.assertRaises(errors.OpPrereqError,
                        lambda: self.fake_cl.SimpleFillOS("os+a", *dicts_pps))

    def testConflictPublicSecret(self):
      "Make sure we disallow attempts to override params based on visibility."
      public_dict = {"A": "A", "X": "X"}
      private_dict = {"B": "B"}
      secret_dict = {"C": "C", "X": "X"}
      dicts_pps = (public_dict, private_dict, secret_dict)

      self.assertRaises(errors.OpPrereqError,
                        lambda: self.fake_cl.SimpleFillOS("os+a", *dicts_pps))

    def testConflictPrivateSecret(self):
      "Make sure we disallow attempts to override params based on visibility."
      public_dict = {"A": "A"}
      private_dict = {"B": "B", "X": "X"}
      secret_dict = {"C": "C", "X": "X"}
      dicts_pps = (public_dict, private_dict, secret_dict)

      self.assertRaises(errors.OpPrereqError,
                        lambda: self.fake_cl.SimpleFillOS("os+a", *dicts_pps))

    def testValidValues(self):
      "Make sure we handle all overriding we do allow correctly."
      public_dict = {"A": "A"}
      private_dict = {"B": "B"}
      secret_dict = {"C": "C"}
      dicts_p = (public_dict,)
      dicts_pp = (public_dict, private_dict)
      dicts_pps = (public_dict, private_dict, secret_dict)
      expected_keys_p = ("A", "D", "E") # nothing private, secret
      expected_keys_pp = ("A", "B", "D", "E", "F", "G") # nothing secret
      expected_keys_pps = ("A", "B", "C", "D", "E", "F", "G") # all of them

      for (dicts, expected_keys) in [(dicts_p, expected_keys_p),
                                     (dicts_pp, expected_keys_pp),
                                     (dicts_pps, expected_keys_pps)]:
        result = self.fake_cl.SimpleFillOS("os+a", *dicts)
        # Values
        for key in result:
          if not result[key] == key:
            self.fail("Invalid public-private fill with input:\n%s\n%s"
                      % (pprint.pformat(dicts), result))
        # Completeness
        if set(result) != set(expected_keys):
          self.fail("Problem with key %s from merge result of:\n%s\n%s"
                    % (set(expected_keys) ^ set(result), # symmetric difference
                       pprint.pformat(dicts),
                       result))

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