Add the gnt-storage client
[ganeti-local] / test / ganeti.constants_unittest.py
index ddc6814..3723e04 100755 (executable)
@@ -16,7 +16,7 @@
 # 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
-# 0.0510-1301, USA.
+# 02110-1301, USA.
 
 
 """Script for unittesting the constants module"""
 
 import unittest
 import re
+import itertools
 
 from ganeti import constants
+from ganeti import locking
+from ganeti import utils
+
+import testutils
 
 
 class TestConstants(unittest.TestCase):
@@ -58,15 +63,63 @@ class TestConstants(unittest.TestCase):
     self.failUnless(constants.LDS_OKAY < constants.LDS_UNKNOWN)
     self.failUnless(constants.LDS_UNKNOWN < constants.LDS_FAULTY)
 
+  def testClockSkew(self):
+    self.failUnless(constants.NODE_MAX_CLOCK_SKEW <
+                    (0.8 * constants.CONFD_MAX_CLOCK_SKEW))
+
+  def testSslCertExpiration(self):
+    self.failUnless(constants.SSL_CERT_EXPIRATION_ERROR <
+                    constants.SSL_CERT_EXPIRATION_WARN)
+
+  def testOpCodePriority(self):
+    self.failUnless(constants.OP_PRIO_LOWEST > constants.OP_PRIO_LOW)
+    self.failUnless(constants.OP_PRIO_LOW > constants.OP_PRIO_NORMAL)
+    self.failUnlessEqual(constants.OP_PRIO_NORMAL, locking._DEFAULT_PRIORITY)
+    self.failUnlessEqual(constants.OP_PRIO_DEFAULT, locking._DEFAULT_PRIORITY)
+    self.failUnless(constants.OP_PRIO_NORMAL > constants.OP_PRIO_HIGH)
+    self.failUnless(constants.OP_PRIO_HIGH > constants.OP_PRIO_HIGHEST)
+
+  def testDiskDefaults(self):
+    self.failUnless(set(constants.DISK_LD_DEFAULTS.keys()) ==
+                    constants.LOGICAL_DISK_TYPES)
+    self.failUnless(set(constants.DISK_DT_DEFAULTS.keys()) ==
+                    constants.DISK_TEMPLATES)
+
+  def testJobStatus(self):
+    self.assertFalse(constants.JOBS_PENDING & constants.JOBS_FINALIZED)
+    self.assertFalse(constants.JOBS_PENDING - constants.JOB_STATUS_ALL)
+    self.assertFalse(constants.JOBS_FINALIZED - constants.JOB_STATUS_ALL)
+
+
+class TestExportedNames(unittest.TestCase):
+  _VALID_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]+$")
+  _BUILTIN_NAME_RE = re.compile(r"^__\w+__$")
+  _EXCEPTIONS = frozenset([
+    "SplitVersion",
+    "BuildVersion",
+    ])
+
+  def test(self):
+    wrong = \
+      set(itertools.ifilterfalse(self._BUILTIN_NAME_RE.match,
+            itertools.ifilterfalse(self._VALID_NAME_RE.match,
+                                   dir(constants))))
+    wrong -= self._EXCEPTIONS
+    self.assertFalse(wrong,
+                     msg=("Invalid names exported from constants module: %s" %
+                          utils.CommaJoin(sorted(wrong))))
+
 
 class TestParameterNames(unittest.TestCase):
   """HV/BE parameter tests"""
   VALID_NAME = re.compile("^[a-zA-Z_][a-zA-Z0-9_]*$")
 
   def testNoDashes(self):
-    for kind, source in [('hypervisor', constants.HVS_PARAMETER_TYPES),
-                         ('backend', constants.BES_PARAMETER_TYPES),
-                         ('nic', constants.NICS_PARAMETER_TYPES),
+    for kind, source in [("hypervisor", constants.HVS_PARAMETER_TYPES),
+                         ("backend", constants.BES_PARAMETER_TYPES),
+                         ("nic", constants.NICS_PARAMETER_TYPES),
+                         ("instdisk", constants.IDISK_PARAMS_TYPES),
+                         ("instnic", constants.INIC_PARAMS_TYPES),
                         ]:
       for key in source:
         self.failUnless(self.VALID_NAME.match(key),
@@ -78,25 +131,17 @@ class TestConfdConstants(unittest.TestCase):
   """Test the confd constants"""
 
   def testFourCc(self):
-    self.failUnlessEqual(len(constants.CONFD_MAGIC_FOURCC), 4,
-                    "Invalid fourcc len, should be 4")
-
-  def _IsUniqueSequence(self, sequence):
-    seen = set()
-    for member in sequence:
-      if member in seen:
-        return False
-      seen.add(member)
-    return True
+    self.assertEqual(len(constants.CONFD_MAGIC_FOURCC), 4,
+                     msg="Invalid fourcc len, should be 4")
 
   def testReqs(self):
-    self.failUnless(self._IsUniqueSequence(constants.CONFD_REQS),
-                    "Duplicated confd request code")
+    self.assertFalse(utils.FindDuplicates(constants.CONFD_REQS),
+                     msg="Duplicated confd request code")
 
   def testReplStatuses(self):
-    self.failUnless(self._IsUniqueSequence(constants.CONFD_REPL_STATUSES),
-                    "Duplicated confd reply status code")
+    self.assertFalse(utils.FindDuplicates(constants.CONFD_REPL_STATUSES),
+                     msg="Duplicated confd reply status code")
 
 
-if __name__ == '__main__':
-  unittest.main()
+if __name__ == "__main__":
+  testutils.GanetiTestProgram()