import logging
import re
-from ganeti import compat
from ganeti import constants
from ganeti import errors
from ganeti import ht
_OPID_RE = re.compile("([a-z])([A-Z])")
#: Utility function for L{OpClusterSetParams}
-_TestClusterOsList = ht.TOr(ht.TNone,
- ht.TListOf(ht.TAnd(ht.TList, ht.TIsLength(2),
- ht.TMap(ht.WithDesc("GetFirstItem")(compat.fst),
- ht.TElemOf(constants.DDMS_VALUES)))))
+_TestClusterOsListItem = \
+ ht.TAnd(ht.TIsLength(2), ht.TItems([
+ ht.TElemOf(constants.DDMS_VALUES),
+ ht.TNonEmptyString,
+ ]))
+_TestClusterOsList = ht.TOr(ht.TNone, ht.TListOf(_TestClusterOsListItem))
# TODO: Generate check from constants.INIC_PARAMS_TYPES
#: Utility function for testing NIC definitions
}))
+class TestClusterOsList(unittest.TestCase):
+ def test(self):
+ good = [
+ None,
+ [],
+ [(constants.DDM_ADD, "dos"),
+ (constants.DDM_REMOVE, "linux")],
+ ]
+
+ for i in good:
+ self.assertTrue(opcodes._TestClusterOsList(i))
+
+ wrong = ["", 0, "xy", ["Hello World"], object(),
+ [("foo", "bar")],
+ [("", "")],
+ [[constants.DDM_ADD]],
+ [(constants.DDM_ADD, "")],
+ [(constants.DDM_REMOVE, "")],
+ [(constants.DDM_ADD, None)],
+ [(constants.DDM_REMOVE, None)],
+ ]
+
+ for i in wrong:
+ self.assertFalse(opcodes._TestClusterOsList(i))
+
+
if __name__ == "__main__":
testutils.GanetiTestProgram()