ht: Add checks for anything, regexp, job ID, container items

The check for container items is useful for tuples and/or lists with
non-uniform values. The “anything” check can be used when any value
should be accepted for an item.

The job ID check, which uses the regexp check, will be used for
expressing opcode dependencies on other jobs.

Signed-off-by: Michael Hanselmann <hansmi@google.com>
Reviewed-by: Iustin Pop <iustin@google.com>

lib/ht.py

@@ -25,6 +25,7 @@ import re
 
 from ganeti import compat
 from ganeti import utils
+from ganeti import constants
 
 
 _PAREN_RE = re.compile("^[a-zA-Z0-9_-]+$")
@@ -118,6 +119,14 @@ NoType = object()
 
 
 # Some basic types
+@WithDesc("Anything")
+def TAny(_):
+  """Accepts any value.
+
+  """
+  return True
+
+
 @WithDesc("NotNone")
 def TNotNone(val):
   """Checks if the given value is not None.
@@ -245,6 +254,18 @@ def TMap(fn, test):
                   (Parens(fn), Parens(test)))(lambda val: test(fn(val)))
 
 
+def TRegex(pobj):
+  """Checks whether a string matches a specific regular expression.
+
+  @param pobj: Compiled regular expression as returned by C{re.compile}
+
+  """
+  desc = WithDesc("String matching regex \"%s\"" %
+                  pobj.pattern.encode("string_escape"))
+
+  return desc(TAnd(TString, pobj.match))
+
+
 # Type aliases
 
 #: a non-empty string
@@ -271,6 +292,10 @@ TStrictPositiveInt = \
 TPositiveFloat = \
   TAnd(TFloat, WithDesc("EqualGreaterZero")(lambda v: v >= 0.0))
 
+#: Job ID
+TJobId = TOr(TPositiveInt,
+             TRegex(re.compile("^%s$" % constants.JOB_ID_TEMPLATE)))
+
 
 def TListOf(my_type):
   """Checks if a given value is a list with all elements of the same type.
@@ -340,3 +365,25 @@ def TStrictDict(require_all, exclusive, items):
   return desc(TAnd(TDict,
                    compat.partial(_TStrictDictCheck, require_all, exclusive,
                                   items)))
+
+
+def TItems(items):
+  """Checks individual items of a container.
+
+  If the verified value and the list of expected items differ in length, this
+  check considers only as many items as are contained in the shorter list. Use
+  L{TIsLength} to enforce a certain length.
+
+  @type items: list
+  @param items: List of checks
+
+  """
+  assert items, "Need items"
+
+  text = ["Item", "item"]
+  desc = WithDesc(utils.CommaJoin("%s %s is %s" %
+                                  (text[int(idx > 0)], idx, Parens(check))
+                                  for (idx, check) in enumerate(items)))
+
+  return desc(lambda value: compat.all(check(i)
+                                       for (check, i) in zip(items, value)))

test/ganeti.ht_unittest.py

@@ -230,6 +230,32 @@ class TestTypeChecks(unittest.TestCase):
     self.assertTrue(fn({"other": 11}))
     self.assertTrue(fn({"other": object()}))
 
+  def testJobId(self):
+    for i in [0, 1, 4395, 2347625220]:
+      self.assertTrue(ht.TJobId(i))
+      self.assertTrue(ht.TJobId(str(i)))
+      self.assertFalse(ht.TJobId(-(i + 1)))
+
+    for i in ["", "-", ".", ",", "a", "99j", "job-123", "\t", " 83 ",
+              None, [], {}, object()]:
+      self.assertFalse(ht.TJobId(i))
+
+  def testItems(self):
+    self.assertRaises(AssertionError, ht.TItems, [])
+
+    fn = ht.TItems([ht.TString])
+    self.assertFalse(fn([0]))
+    self.assertFalse(fn([None]))
+    self.assertTrue(fn(["Hello"]))
+    self.assertTrue(fn(["Hello", "World"]))
+    self.assertTrue(fn(["Hello", 0, 1, 2, "anything"]))
+
+    fn = ht.TItems([ht.TAny, ht.TInt, ht.TAny])
+    self.assertTrue(fn(["Hello", 0, []]))
+    self.assertTrue(fn(["Hello", 893782]))
+    self.assertTrue(fn([{}, -938210858947, None]))
+    self.assertFalse(fn(["Hello", []]))
+
 
 if __name__ == "__main__":
   testutils.GanetiTestProgram()