"""Module implementing the parameter types code."""
import re
+import operator
from ganeti import compat
from ganeti import utils
+from ganeti import constants
_PAREN_RE = re.compile("^[a-zA-Z0-9_-]+$")
return "(%s)" % text
+class _WrapperBase(object):
+ __slots__ = [
+ "_fn",
+ "_text",
+ ]
+
+ def __init__(self, text, fn):
+ """Initializes this class.
+
+ @param text: Description
+ @param fn: Wrapped function
+
+ """
+ assert text.strip()
+
+ self._text = text
+ self._fn = fn
+
+ def __call__(self, *args):
+ return self._fn(*args)
+
+
+class _DescWrapper(_WrapperBase):
+ """Wrapper class for description text.
+
+ """
+ def __str__(self):
+ return self._text
+
+
+class _CommentWrapper(_WrapperBase):
+ """Wrapper class for comment.
+
+ """
+ def __str__(self):
+ return "%s [%s]" % (self._fn, self._text)
+
+
def WithDesc(text):
"""Builds wrapper class with description text.
"""
assert text[0] == text[0].upper()
- class wrapper(object): # pylint: disable-msg=C0103
- __slots__ = ["__call__"]
+ return compat.partial(_DescWrapper, text)
- def __init__(self, fn):
- """Initializes this class.
- @param fn: Wrapped function
+def Comment(text):
+ """Builds wrapper for adding comment to description text.
- """
- self.__call__ = fn
+ @type text: string
+ @param text: Comment text
+ @return: Callable class
- def __str__(self):
- return text
+ """
+ assert not frozenset(text).intersection("[]")
- return wrapper
+ return compat.partial(_CommentWrapper, text)
def CombinationDesc(op, args, fn):
# 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.
#
# >>> (isinstance(False, int), isinstance(True, int))
# (True, True)
- return isinstance(val, int) and not isinstance(val, bool)
+ return isinstance(val, (int, long)) and not isinstance(val, bool)
@WithDesc("Float")
(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
TStrictPositiveInt = \
TAnd(TInt, WithDesc("GreaterThanZero")(lambda v: v > 0))
+#: a strictly negative integer (0 > value)
+TStrictNegativeInt = \
+ TAnd(TInt, WithDesc("LessThanZero")(compat.partial(operator.gt, 0)))
+
+#: a positive float
+TPositiveFloat = \
+ TAnd(TFloat, WithDesc("EqualGreaterZero")(lambda v: v >= 0.0))
+
+#: Job ID
+TJobId = WithDesc("JobId")(TOr(TPositiveInt,
+ TRegex(re.compile("^%s$" %
+ constants.JOB_ID_TEMPLATE))))
+
+#: Number
+TNumber = TOr(TInt, TFloat)
+
+#: Relative job ID
+TRelativeJobId = WithDesc("RelativeJobId")(TStrictNegativeInt)
+
def TListOf(my_type):
"""Checks if a given value is a list with all elements of the same type.
compat.all(val_type(v) for v in container.values()))
return desc(TAnd(TDict, fn))
+
+
+def _TStrictDictCheck(require_all, exclusive, items, val):
+ """Helper function for L{TStrictDict}.
+
+ """
+ notfound_fn = lambda _: not exclusive
+
+ if require_all and not frozenset(val.keys()).issuperset(items.keys()):
+ # Requires items not found in value
+ return False
+
+ return compat.all(items.get(key, notfound_fn)(value)
+ for (key, value) in val.items())
+
+
+def TStrictDict(require_all, exclusive, items):
+ """Strict dictionary check with specific keys.
+
+ @type require_all: boolean
+ @param require_all: Whether all keys in L{items} are required
+ @type exclusive: boolean
+ @param exclusive: Whether only keys listed in L{items} should be accepted
+ @type items: dictionary
+ @param items: Mapping from key (string) to verification function
+
+ """
+ descparts = ["Dictionary containing"]
+
+ if exclusive:
+ descparts.append(" none but the")
+
+ if require_all:
+ descparts.append(" required")
+
+ if len(items) == 1:
+ descparts.append(" key ")
+ else:
+ descparts.append(" keys ")
+
+ descparts.append(utils.CommaJoin("\"%s\" (value %s)" % (key, value)
+ for (key, value) in items.items()))
+
+ desc = WithDesc("".join(descparts))
+
+ 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)))