#!/usr/bin/python
#

# Copyright (C) 2006, 2007, 2008, 2010, 2012, 2013 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU 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.


"""Script for unittesting the objects module"""


import copy
import unittest

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

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,
        }
    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,
        }
    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,
        }
    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,
        }
    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 }
    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 _GenericCheck(self, inst):
    for i in [inst.all_nodes, inst.secondary_nodes]:
      self.assertTrue(isinstance(inst.all_nodes, (list, tuple)),
                      msg="Data type doesn't guarantee order")

    self.assertTrue(inst.primary_node not in inst.secondary_nodes)
    self.assertEqual(inst.all_nodes[0], inst.primary_node,
                     msg="Primary node not first node in list")

  def testNodesNoDisks(self):
    inst = objects.Instance(name="fakeinst.example.com",
      primary_node="pnode.example.com",
      disks=[
        ])

    self._GenericCheck(inst)
    self.assertEqual(len(inst.secondary_nodes), 0)
    self.assertEqual(set(inst.all_nodes), set([inst.primary_node]))
    self.assertEqual(inst.MapLVsByNode(), {
      inst.primary_node: [],
      })

  def testNodesPlainDisks(self):
    inst = objects.Instance(name="fakeinstplain.example.com",
      primary_node="node3.example.com",
      disks=[
        objects.Disk(dev_type=constants.DT_PLAIN, size=128,
                     logical_id=("myxenvg", "disk25494")),
        objects.Disk(dev_type=constants.DT_PLAIN, size=512,
                     logical_id=("myxenvg", "disk29071")),
        ])

    self._GenericCheck(inst)
    self.assertEqual(len(inst.secondary_nodes), 0)
    self.assertEqual(set(inst.all_nodes), set([inst.primary_node]))
    self.assertEqual(inst.MapLVsByNode(), {
      inst.primary_node: ["myxenvg/disk25494", "myxenvg/disk29071"],
      })

  def testNodesDrbdDisks(self):
    inst = objects.Instance(name="fakeinstdrbd.example.com",
      primary_node="node10.example.com",
      disks=[
        objects.Disk(dev_type=constants.DT_DRBD8, size=786432,
          logical_id=("node10.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._GenericCheck(inst)
    self.assertEqual(set(inst.secondary_nodes), set(["node15.example.com"]))
    self.assertEqual(set(inst.all_nodes),
                     set([inst.primary_node, "node15.example.com"]))
    self.assertEqual(inst.MapLVsByNode(), {
      inst.primary_node: ["myxenvg/disk0", "myxenvg/meta0"],
      "node15.example.com": ["myxenvg/disk0", "myxenvg/meta0"],
      })

    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]
    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)


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