Synnefo » snf-ganeti

    op = OpTest()

    lu = TestLU(op)

    # Neither node nor iallocator given

    for n in (None, []):

          assert iallocator._NEVAC_RESULT(alloc_result)

          lu = _FakeLU()

          if moved:

            (_, (info_args, )) = lu.info_log.pop(0)

    assert iallocator._NEVAC_RESULT(alloc_result)

    lu = _FakeLU()

                      lu, alloc_result, False, False)

    self.assertFalse(lu.info_log)

    (_, (args, )) = lu.warning_log.pop(0)

class TestHvStateHelper(unittest.TestCase):

  def testWithoutOpData(self):

                     None)

  def testWithoutOldData(self):

        constants.HVST_MEMORY_TOTAL: 4096,

        },

      }

  def testWithWrongHv(self):

    new = {

        constants.HVST_MEMORY_TOTAL: 4096,

        },

      }

                      new, None)

class TestDiskStateHelper(unittest.TestCase):

  def testWithoutOpData(self):

                     None)

  def testWithoutOldData(self):

          },

        },

      }

  def testWithWrongStorageType(self):

    new = {

          },

        },

      }

                      new, None)

  def test(self):

    compute_fn = _ValidateComputeMinMaxSpec

    self.assertEqual(ret, [])

  def testDiskFull(self):

    compute_fn = _NoDiskComputeMinMaxSpec

    self.assertEqual(ret, [constants.ISPEC_DISK_COUNT])

  def testDiskLess(self):

    compute_fn = _NoDiskComputeMinMaxSpec

    self.assertEqual(ret, [])

  def testWrongTemplates(self):

    compute_fn = _ValidateComputeMinMaxSpec

    self.assertEqual(len(ret), 1)

    self.assertTrue("Disk template" in ret[0])

  def testInvalidArguments(self):

                      self._MICRO_IPOL, 1024, 1, 1, 1, [], 1,

                      constants.DT_PLAIN,)

  def testInvalidSpec(self):

    spec = _SpecWrapper([None, False, "foo", None, "bar", None])

    compute_fn = spec.ComputeMinMaxSpec

    self.assertEqual(ret, ["foo", "bar"])

    self.assertFalse(spec.spec)

      constants.IPOLICY_DTS: [disk_template],

      }

    def AssertComputeViolation(ipolicy, violations):

      self.assertEqual(len(ret), violations)

    AssertComputeViolation(ipolicy1, 0)

                                disk_template=constants.DT_PLAIN)

    stub = _StubComputeIPolicySpecViolation(2048, 2, 1, 0, [512], 4,

                                            constants.DT_PLAIN)

    self.assertEqual(ret, [])

    instance2 = objects.Instance(beparams={}, disks=disks, nics=[],

                                 disk_template=constants.DT_PLAIN)

    self.assertEqual(ret, [])

  def testNoViolation(self):

    compute_recoder = _CallRecorder(return_value=[])

    self.assertTrue(compute_recoder.called)

    self.assertEqual(self.lu.warning_log, [])

  def testNoIgnore(self):

    compute_recoder = _CallRecorder(return_value=["mem_size not in range"])

                      self.lu, NotImplemented, self.instance,

                      self.target_node, NotImplemented,

                      _compute_fn=compute_recoder)

  def testIgnoreViolation(self):

    compute_recoder = _CallRecorder(return_value=["mem_size not in range"])

                                     self.target_node, NotImplemented,

                                     ignore=True, _compute_fn=compute_recoder)

    self.assertTrue(compute_recoder.called)

  def testEmptyContainer(self):

    container = []

    chgdesc = []

                                None)

    self.assertEqual(container, [])

    self.assertEqual(chgdesc, [])

  def testAdd(self):

    container = []

    chgdesc = []

      (constants.DDM_ADD, -1, "Hello"),

      (constants.DDM_ADD, -1, "World"),

      (constants.DDM_ADD, 0, "Start"),

      (constants.DDM_ADD, -1, "End"),

      ], None)

                                None, None, None)

    self.assertEqual(container, ["Start", "Hello", "World", "End"])

    self.assertEqual(chgdesc, [])

      (constants.DDM_ADD, 0, "zero"),

      (constants.DDM_ADD, 3, "Added"),

      (constants.DDM_ADD, 5, "four"),

      (constants.DDM_ADD, 7, "xyz"),

      ], None)

                                None, None, None)

    self.assertEqual(container,

                     ["zero", "Start", "Hello", "Added", "World", "four",

    self.assertEqual(chgdesc, [])

    for idx in [-2, len(container) + 1]:

        (constants.DDM_ADD, idx, "error"),

        ], None)

                        "test", container, None, mods, None, None, None)

  def testRemoveError(self):

    for idx in [0, 1, 2, 100, -1, -4]:

        (constants.DDM_REMOVE, idx, None),

        ], None)

                        "test", [], None, mods, None, None, None)

      (constants.DDM_REMOVE, 0, object()),

      ], None)

                      "test", [""], None, mods, None, None, None)

  def testAddError(self):

    for idx in range(-100, -1) + [100]:

        (constants.DDM_ADD, idx, None),

        ], None)

                        "test", [], None, mods, None, None, None)

  def testRemove(self):

    container = ["item 1", "item 2"]

      (constants.DDM_ADD, -1, "aaa"),

      (constants.DDM_REMOVE, -1, None),

      (constants.DDM_ADD, -1, "bbb"),

      ], None)

    chgdesc = []

                                None, None, None)

    self.assertEqual(container, ["item 1", "item 2", "bbb"])

    self.assertEqual(chgdesc, [

  def testModify(self):

    container = ["item 1", "item 2"]

      (constants.DDM_MODIFY, -1, "a"),

      (constants.DDM_MODIFY, 0, "b"),

      (constants.DDM_MODIFY, 1, "c"),

      ], None)

    chgdesc = []

                                None, None, None)

    self.assertEqual(container, ["item 1", "item 2"])

    self.assertEqual(chgdesc, [])

    for idx in [-2, len(container) + 1]:

        (constants.DDM_MODIFY, idx, "error"),

        ], None)

                        "test", container, None, mods, None, None, None)

  class _PrivateData:

  def testAddWithCreateFunction(self):

    container = []

    chgdesc = []

      (constants.DDM_ADD, -1, "Hello"),

      (constants.DDM_ADD, -1, "World"),

      (constants.DDM_ADD, 0, "Start"),

      (constants.DDM_REMOVE, 2, None),

      (constants.DDM_ADD, 1, "More"),

      ], self._PrivateData)

                                self._CreateTestFn, self._ModifyTestFn,

                                self._RemoveTestFn)

    self.assertEqual(container, [

    return copy.deepcopy(constants.DISK_DT_DEFAULTS)

  def testWrongDiskTemplate(self):

    disk_template = "##unknown##"

    assert disk_template not in constants.DISK_TEMPLATES

                      self.GetDiskParams())

  def testDiskless(self):

    result = gdt(self.lu, constants.DT_DISKLESS, "inst27734.example.com",

                 "node30113.example.com", [], [],

                       file_driver=NotImplemented,

                       req_file_storage=NotImplemented,

                       req_shr_file_storage=NotImplemented):

    map(lambda params: utils.ForceDictType(params,

                                           constants.IDISK_PARAMS_TYPES),

      ])

  def testDrbd8(self):

    drbd8_defaults = constants.DISK_LD_DEFAULTS[constants.LD_DRBD8]

    drbd8_default_metavg = drbd8_defaults[constants.LDP_DEFAULT_METAVG]

                            disk_template=constants.DT_PLAIN,

                            disks=disks)

  def _FailingWipeCb(self, (disk, _), offset, size):

    # This should only ever be called for the first disk

                            disks=disks)

    try:

    except errors.OpExecError, err:

      self.assertTrue(str(err), "Could not wipe disk 0 at offset 0 ")

    else:

    (lu, inst, pauset, progresst) = self._PrepareWipeTest(0, disks)

    self.assertEqual(pauset.history, [

      ("disk0", 1024, True),

        self._PrepareWipeTest(start_offset, disks)

      # Test start offset with only one disk

      # Only the second disk may have been paused and wiped

      self.assertEqual(pauset.history, [

class TestCopyLockList(unittest.TestCase):

  def test(self):

    names = ["foo", "bar"]

    self.assertEqual(names, output)

    self.assertNotEqual(id(names), id(output), msg="List was not copied")

      }

    if not isgroup:

      diff_policy[constants.ISPECS_STD] = diff_std

    self.assertTrue(constants.ISPECS_MINMAX in new_policy)

    self.assertEqual(new_policy[constants.ISPECS_MINMAX], diff_minmax)

          self.assertEqual(new_std[key], old_std[key])

  def _TestSet(self, old_policy, diff_policy, isgroup):

                                           group_policy=isgroup)

    for key in diff_policy:

      self.assertTrue(key in new_policy)

    diff_policy = {

      constants.IPOLICY_SPINDLE_RATIO: constants.VALUE_DEFAULT,

      }

    for key in diff_policy:

      self.assertFalse(key in new_policy)

    for key in old_policy:

        self.assertTrue(key in new_policy)

        self.assertEqual(new_policy[key], old_policy[key])

                      old_policy, diff_policy, group_policy=False)

  def testUnsetEmpty(self):

      diff_policy = {

        key: constants.VALUE_DEFAULT,

        }

    self.assertEqual(new_policy, old_policy)

  def _TestInvalidKeys(self, old_policy, isgroup):

      INVALID_KEY: 3,

      }

    invalid_policy = INVALID_DICT

                      old_policy, invalid_policy, group_policy=isgroup)

    invalid_ispecs = {

      constants.ISPECS_MINMAX: [INVALID_DICT],

      }

                      old_policy, invalid_ispecs, group_policy=isgroup)

    if isgroup:

      invalid_for_group = {

        constants.ISPECS_STD: constants.IPOLICY_DEFAULTS[constants.ISPECS_STD],

        }

                        old_policy, invalid_for_group, group_policy=isgroup)

    good_ispecs = self._OLD_CLUSTER_POLICY[constants.ISPECS_MINMAX]

    invalid_ispecs = copy.deepcopy(good_ispecs)

        ispec = minmax[key]

        ispec[INVALID_KEY] = None

        self.assertRaises(errors.TypeEnforcementError,

                          invalid_policy, group_policy=isgroup)

        del ispec[INVALID_KEY]

        for par in constants.ISPECS_PARAMETERS:

          oldv = ispec[par]

          ispec[par] = "this_is_not_good"

          self.assertRaises(errors.TypeEnforcementError,

                            old_policy, invalid_policy, group_policy=isgroup)

          ispec[par] = oldv

    # This is to make sure that no two errors were present during the tests

  def testInvalidKeys(self):

    self._TestInvalidKeys(self._OLD_GROUP_POLICY, True)

      bad_policy = {

        par: "invalid_value",

        }

                        bad_policy, group_policy=True)

if __name__ == "__main__":