Statistics
| Branch: | Tag: | Revision:

root / test / ganeti.mcpu_unittest.py @ c9c33a28

History | View | Annotate | Download (2.8 kB)

1
#!/usr/bin/python
2
#
3

    
4
# Copyright (C) 2009, 2011 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
"""Script for unittesting the mcpu module"""
23

    
24

    
25
import unittest
26

    
27
from ganeti import mcpu
28
from ganeti import opcodes
29
from ganeti.constants import \
30
    LOCK_ATTEMPTS_TIMEOUT, \
31
    LOCK_ATTEMPTS_MAXWAIT, \
32
    LOCK_ATTEMPTS_MINWAIT
33

    
34
import testutils
35

    
36

    
37
REQ_BGL_WHITELIST = frozenset([
38
  opcodes.OpClusterActivateMasterIp,
39
  opcodes.OpClusterDeactivateMasterIp,
40
  opcodes.OpClusterDestroy,
41
  opcodes.OpClusterPostInit,
42
  opcodes.OpClusterRename,
43
  opcodes.OpInstanceRename,
44
  opcodes.OpNodeAdd,
45
  opcodes.OpNodeRemove,
46
  opcodes.OpTestAllocator,
47
  ])
48

    
49

    
50
class TestLockAttemptTimeoutStrategy(unittest.TestCase):
51
  def testConstants(self):
52
    tpa = mcpu.LockAttemptTimeoutStrategy._TIMEOUT_PER_ATTEMPT
53
    self.assert_(len(tpa) > LOCK_ATTEMPTS_TIMEOUT / LOCK_ATTEMPTS_MAXWAIT)
54
    self.assert_(sum(tpa) >= LOCK_ATTEMPTS_TIMEOUT)
55

    
56
  def testSimple(self):
57
    strat = mcpu.LockAttemptTimeoutStrategy(_random_fn=lambda: 0.5,
58
                                            _time_fn=lambda: 0.0)
59

    
60
    prev = None
61
    for i in range(len(strat._TIMEOUT_PER_ATTEMPT)):
62
      timeout = strat.NextAttempt()
63
      self.assert_(timeout is not None)
64

    
65
      self.assert_(timeout <= LOCK_ATTEMPTS_MAXWAIT)
66
      self.assert_(timeout >= LOCK_ATTEMPTS_MINWAIT)
67
      self.assert_(prev is None or timeout >= prev)
68

    
69
      prev = timeout
70

    
71
    for _ in range(10):
72
      self.assert_(strat.NextAttempt() is None)
73

    
74

    
75
class TestDispatchTable(unittest.TestCase):
76
  def test(self):
77
    for opcls in opcodes.OP_MAPPING.values():
78
      if not opcls.WITH_LU:
79
        continue
80
      self.assertTrue(opcls in mcpu.Processor.DISPATCH_TABLE,
81
                      msg="%s missing handler class" % opcls)
82

    
83
      # Check against BGL whitelist
84
      lucls = mcpu.Processor.DISPATCH_TABLE[opcls]
85
      if lucls.REQ_BGL:
86
        self.assertTrue(opcls in REQ_BGL_WHITELIST,
87
                        msg=("%s not whitelisted for BGL" % opcls.OP_ID))
88
      else:
89
        self.assertFalse(opcls in REQ_BGL_WHITELIST,
90
                         msg=("%s whitelisted for BGL, but doesn't use it" %
91
                              opcls.OP_ID))
92

    
93

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