Merge branch 'devel-2.5'

[ganeti-local] / test / ganeti.mcpu_unittest.py

diff --git a/test/ganeti.mcpu_unittest.py b/test/ganeti.mcpu_unittest.py
index a60ebac..2070762 100755 (executable)
--- a/test/ganeti.mcpu_unittest.py
+++ b/test/ganeti.mcpu_unittest.py
@@ -1,7 +1,7 @@
 #!/usr/bin/python
 #
 
-# Copyright (C) 2009 Google Inc.
+# Copyright (C) 2009, 2011 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
@@ -25,37 +25,48 @@
 import unittest
 
 from ganeti import mcpu
+from ganeti import opcodes
+from ganeti.constants import \
+    LOCK_ATTEMPTS_TIMEOUT, \
+    LOCK_ATTEMPTS_MAXWAIT, \
+    LOCK_ATTEMPTS_MINWAIT
+
+import testutils
 
 
 class TestLockAttemptTimeoutStrategy(unittest.TestCase):
   def testConstants(self):
-    tpa = mcpu._LockAttemptTimeoutStrategy._TIMEOUT_PER_ATTEMPT
-    self.assert_(len(tpa) > 10)
-    self.assert_(sum(tpa) >= 150.0)
+    tpa = mcpu.LockAttemptTimeoutStrategy._TIMEOUT_PER_ATTEMPT
+    self.assert_(len(tpa) > LOCK_ATTEMPTS_TIMEOUT / LOCK_ATTEMPTS_MAXWAIT)
+    self.assert_(sum(tpa) >= LOCK_ATTEMPTS_TIMEOUT)
 
   def testSimple(self):
-    strat = mcpu._LockAttemptTimeoutStrategy(_random_fn=lambda: 0.5,
-                                             _time_fn=lambda: 0.0)
-
-    self.assertEqual(strat._attempt, 0)
+    strat = mcpu.LockAttemptTimeoutStrategy(_random_fn=lambda: 0.5,
+                                            _time_fn=lambda: 0.0)
 
     prev = None
-    for i in range(len(mcpu._LockAttemptTimeoutStrategy._TIMEOUT_PER_ATTEMPT)):
-      timeout = strat.CalcRemainingTimeout()
+    for i in range(len(strat._TIMEOUT_PER_ATTEMPT)):
+      timeout = strat.NextAttempt()
       self.assert_(timeout is not None)
 
-      self.assert_(timeout <= 10.0)
-      self.assert_(timeout >= 0.0)
+      self.assert_(timeout <= LOCK_ATTEMPTS_MAXWAIT)
+      self.assert_(timeout >= LOCK_ATTEMPTS_MINWAIT)
       self.assert_(prev is None or timeout >= prev)
 
-      strat = strat.NextAttempt()
-      self.assertEqual(strat._attempt, i + 1)
-
       prev = timeout
 
     for _ in range(10):
-      self.assert_(strat.CalcRemainingTimeout() is None)
+      self.assert_(strat.NextAttempt() is None)
+
+
+class TestDispatchTable(unittest.TestCase):
+  def test(self):
+    for opcls in opcodes.OP_MAPPING.values():
+      if not opcls.WITH_LU:
+        continue
+      self.assertTrue(opcls in mcpu.Processor.DISPATCH_TABLE,
+                      msg="%s missing handler class" % opcls)
 
 
 if __name__ == "__main__":
-  unittest.main()
+  testutils.GanetiTestProgram()