code.grnet.gr Git - ganeti-local/blob - test/py/cmdlib/test_unittest.py

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2013 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Tests for LUTest*"""
  23
  24 import mock
  25
  26 from ganeti import constants
  27 from ganeti import opcodes
  28
  29 from testsupport import *
  30
  31 import testutils
  32
  33 DELAY_DURATION = 0.01
  34
  35
  36 class TestLUTestDelay(CmdlibTestCase):
  37   def testRepeatedInvocation(self):
  38     op = opcodes.OpTestDelay(duration=DELAY_DURATION,
  39                              repeat=3)
  40     self.ExecOpCode(op)
  41
  42     self.assertLogContainsMessage(" - INFO: Test delay iteration 0/2")
  43     self.mcpu.assertLogContainsEntry(constants.ELOG_MESSAGE,
  44                                      " - INFO: Test delay iteration 1/2")
  45     self.assertLogContainsRegex("2/2$")
  46
  47   def testInvalidDuration(self):
  48     op = opcodes.OpTestDelay(duration=-1)
  49
  50     self.ExecOpCodeExpectOpExecError(op)
  51
  52   def testOnNodeUuid(self):
  53     node_uuids = [self.master_uuid]
  54     op = opcodes.OpTestDelay(duration=DELAY_DURATION,
  55                              on_node_uuids=node_uuids)
  56     self.ExecOpCode(op)
  57
  58     self.rpc.call_test_delay.assert_called_once_with(node_uuids, DELAY_DURATION)
  59
  60   def testOnNodeName(self):
  61     op = opcodes.OpTestDelay(duration=DELAY_DURATION,
  62                              on_nodes=[self.master.name])
  63     self.ExecOpCode(op)
  64
  65     self.rpc.call_test_delay.assert_called_once_with([self.master_uuid],
  66                                                      DELAY_DURATION)
  67
  68   def testSuccessfulRpc(self):
  69     op = opcodes.OpTestDelay(duration=DELAY_DURATION,
  70                              on_nodes=[self.master.name])
  71
  72     self.rpc.call_test_delay.return_value = \
  73       self.RpcResultsBuilder() \
  74         .AddSuccessfulNode(self.master) \
  75         .Build()
  76
  77     self.ExecOpCode(op)
  78
  79     self.rpc.call_test_delay.assert_called_once()
  80
  81   def testFailingRpc(self):
  82     op = opcodes.OpTestDelay(duration=DELAY_DURATION,
  83                              on_nodes=[self.master.name])
  84
  85     self.rpc.call_test_delay.return_value = \
  86       self.RpcResultsBuilder() \
  87         .AddFailedNode(self.master) \
  88         .Build()
  89
  90     self.ExecOpCodeExpectOpExecError(op)
  91
  92   def testMultipleNodes(self):
  93     node1 = self.cfg.AddNewNode()
  94     node2 = self.cfg.AddNewNode()
  95     op = opcodes.OpTestDelay(duration=DELAY_DURATION,
  96                              on_nodes=[node1.name, node2.name])
  97
  98     self.rpc.call_test_delay.return_value = \
  99       self.RpcResultsBuilder() \
 100         .AddSuccessfulNode(node1) \
 101         .AddSuccessfulNode(node2) \
 102         .Build()
 103
 104     self.ExecOpCode(op)
 105
 106     self.rpc.call_test_delay.assert_called_once_with([node1.uuid, node2.uuid],
 107                                                      DELAY_DURATION)
 108
 109
 110 class TestLUTestAllocator(CmdlibTestCase):
 111   def setUp(self):
 112     super(TestLUTestAllocator, self).setUp()
 113
 114     self.base_op = opcodes.OpTestAllocator(
 115                       name="new-instance.example.com",
 116                       nics=[],
 117                       disks=[],
 118                       disk_template=constants.DT_DISKLESS,
 119                       direction=constants.IALLOCATOR_DIR_OUT,
 120                       iallocator="test")
 121
 122     self.valid_alloc_op = \
 123       self.CopyOpCode(self.base_op,
 124                       mode=constants.IALLOCATOR_MODE_ALLOC,
 125                       memory=0,
 126                       disk_template=constants.DT_DISKLESS,
 127                       os="mock_os",
 128                       vcpus=1)
 129     self.valid_multi_alloc_op = \
 130       self.CopyOpCode(self.base_op,
 131                       mode=constants.IALLOCATOR_MODE_MULTI_ALLOC,
 132                       instances=["new-instance.example.com"],
 133                       memory=0,
 134                       disk_template=constants.DT_DISKLESS,
 135                       os="mock_os",
 136                       vcpus=1)
 137     self.valid_reloc_op = \
 138       self.CopyOpCode(self.base_op,
 139                       mode=constants.IALLOCATOR_MODE_RELOC)
 140     self.valid_chg_group_op = \
 141       self.CopyOpCode(self.base_op,
 142                       mode=constants.IALLOCATOR_MODE_CHG_GROUP,
 143                       instances=["new-instance.example.com"],
 144                       target_groups=["default"])
 145     self.valid_node_evac_op = \
 146       self.CopyOpCode(self.base_op,
 147                       mode=constants.IALLOCATOR_MODE_NODE_EVAC,
 148                       instances=["new-instance.example.com"],
 149                       evac_mode=constants.IALLOCATOR_NEVAC_PRI)
 150
 151     self.iallocator_cls.return_value.in_text = "mock in text"
 152     self.iallocator_cls.return_value.out_text = "mock out text"
 153
 154   def testMissingDirection(self):
 155     op = self.CopyOpCode(self.base_op,
 156                          direction=self.REMOVE)
 157
 158     self.ExecOpCodeExpectOpPrereqError(
 159       op, "'OP_TEST_ALLOCATOR.direction' fails validation")
 160
 161   def testAllocWrongDisks(self):
 162     op = self.CopyOpCode(self.valid_alloc_op,
 163                          disks=[0, "test"])
 164
 165     self.ExecOpCodeExpectOpPrereqError(op, "Invalid contents")
 166
 167   def testAllocWithExistingInstance(self):
 168     inst = self.cfg.AddNewInstance()
 169     op = self.CopyOpCode(self.valid_alloc_op, name=inst.name)
 170
 171     self.ExecOpCodeExpectOpPrereqError(op, "already in the cluster")
 172
 173   def testAllocMultiAllocMissingIAllocator(self):
 174     for mode in [constants.IALLOCATOR_MODE_ALLOC,
 175                  constants.IALLOCATOR_MODE_MULTI_ALLOC]:
 176       op = self.CopyOpCode(self.base_op,
 177                            mode=mode,
 178                            iallocator=None)
 179
 180       self.ResetMocks()
 181       self.ExecOpCodeExpectOpPrereqError(op, "Missing allocator name")
 182
 183   def testChgGroupNodeEvacMissingInstances(self):
 184     for mode in [constants.IALLOCATOR_MODE_CHG_GROUP,
 185                  constants.IALLOCATOR_MODE_NODE_EVAC]:
 186       op = self.CopyOpCode(self.base_op,
 187                            mode=mode)
 188
 189       self.ResetMocks()
 190       self.ExecOpCodeExpectOpPrereqError(op, "Missing instances")
 191
 192   def testAlloc(self):
 193     op = self.valid_alloc_op
 194
 195     self.ExecOpCode(op)
 196
 197     assert self.iallocator_cls.call_count == 1
 198     self.iallocator_cls.return_value.Run \
 199       .assert_called_once_with("test", validate=False)
 200
 201   def testReloc(self):
 202     op = self.valid_reloc_op
 203     self.cfg.AddNewInstance(name=op.name)
 204
 205     self.ExecOpCode(op)
 206
 207     assert self.iallocator_cls.call_count == 1
 208     self.iallocator_cls.return_value.Run \
 209       .assert_called_once_with("test", validate=False)
 210
 211   def testChgGroup(self):
 212     op = self.valid_chg_group_op
 213     for inst_name in op.instances:
 214       self.cfg.AddNewInstance(name=inst_name)
 215
 216     self.ExecOpCode(op)
 217
 218     assert self.iallocator_cls.call_count == 1
 219     self.iallocator_cls.return_value.Run \
 220       .assert_called_once_with("test", validate=False)
 221
 222   def testNodeEvac(self):
 223     op = self.valid_node_evac_op
 224     for inst_name in op.instances:
 225       self.cfg.AddNewInstance(name=inst_name)
 226
 227     self.ExecOpCode(op)
 228
 229     assert self.iallocator_cls.call_count == 1
 230     self.iallocator_cls.return_value.Run \
 231       .assert_called_once_with("test", validate=False)
 232
 233   def testMultiAlloc(self):
 234     op = self.valid_multi_alloc_op
 235
 236     self.ExecOpCode(op)
 237
 238     assert self.iallocator_cls.call_count == 1
 239     self.iallocator_cls.return_value.Run \
 240       .assert_called_once_with("test", validate=False)
 241
 242   def testAllocDirectionIn(self):
 243     op = self.CopyOpCode(self.valid_alloc_op,
 244                          direction=constants.IALLOCATOR_DIR_IN)
 245
 246     self.ExecOpCode(op)
 247
 248
 249 if __name__ == "__main__":
 250   testutils.GanetiTestProgram()