4 # Copyright (C) 2011 Google Inc.
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 # General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
22 """Script for testing ganeti.utils.text"""
30 from cStringIO import StringIO
32 from ganeti import constants
33 from ganeti import utils
34 from ganeti import errors
39 class TestMatchNameComponent(unittest.TestCase):
40 """Test case for the MatchNameComponent function"""
42 def testEmptyList(self):
43 """Test that there is no match against an empty list"""
44 self.failUnlessEqual(utils.MatchNameComponent("", []), None)
45 self.failUnlessEqual(utils.MatchNameComponent("test", []), None)
47 def testSingleMatch(self):
48 """Test that a single match is performed correctly"""
49 mlist = ["test1.example.com", "test2.example.com", "test3.example.com"]
50 for key in "test2", "test2.example", "test2.example.com":
51 self.failUnlessEqual(utils.MatchNameComponent(key, mlist), mlist[1])
53 def testMultipleMatches(self):
54 """Test that a multiple match is returned as None"""
55 mlist = ["test1.example.com", "test1.example.org", "test1.example.net"]
56 for key in "test1", "test1.example":
57 self.failUnlessEqual(utils.MatchNameComponent(key, mlist), None)
59 def testFullMatch(self):
60 """Test that a full match is returned correctly"""
62 key2 = "test1.example"
63 mlist = [key2, key2 + ".com"]
64 self.failUnlessEqual(utils.MatchNameComponent(key1, mlist), None)
65 self.failUnlessEqual(utils.MatchNameComponent(key2, mlist), key2)
67 def testCaseInsensitivePartialMatch(self):
68 """Test for the case_insensitive keyword"""
69 mlist = ["test1.example.com", "test2.example.net"]
70 self.assertEqual(utils.MatchNameComponent("test2", mlist,
71 case_sensitive=False),
73 self.assertEqual(utils.MatchNameComponent("Test2", mlist,
74 case_sensitive=False),
76 self.assertEqual(utils.MatchNameComponent("teSt2", mlist,
77 case_sensitive=False),
79 self.assertEqual(utils.MatchNameComponent("TeSt2", mlist,
80 case_sensitive=False),
83 def testCaseInsensitiveFullMatch(self):
84 mlist = ["ts1.ex", "ts1.ex.org", "ts2.ex", "Ts2.ex"]
86 # Between the two ts1 a full string match non-case insensitive should work
87 self.assertEqual(utils.MatchNameComponent("Ts1", mlist,
88 case_sensitive=False),
90 self.assertEqual(utils.MatchNameComponent("Ts1.ex", mlist,
91 case_sensitive=False),
93 self.assertEqual(utils.MatchNameComponent("ts1.ex", mlist,
94 case_sensitive=False),
97 # Between the two ts2 only case differs, so only case-match works
98 self.assertEqual(utils.MatchNameComponent("ts2.ex", mlist,
99 case_sensitive=False),
101 self.assertEqual(utils.MatchNameComponent("Ts2.ex", mlist,
102 case_sensitive=False),
104 self.assertEqual(utils.MatchNameComponent("TS2.ex", mlist,
105 case_sensitive=False),
109 class TestDnsNameGlobPattern(unittest.TestCase):
113 "node2-0.example.com",
114 "node2-1.example.com",
118 "sub.site.example.com",
121 def _Test(self, pattern):
122 re_pat = utils.DnsNameGlobPattern(pattern)
124 return filter(re.compile(re_pat).match, self.names)
127 for pattern in ["xyz", "node", " ", "example.net", "x*.example.*",
129 self.assertEqual(self._Test(pattern), [])
131 for pattern in ["*", "???*"]:
132 self.assertEqual(self._Test(pattern), self.names)
134 self.assertEqual(self._Test("node1.*.net"), ["node1.example.net"])
135 self.assertEqual(self._Test("*.example.net"), ["node1.example.net"])
136 self.assertEqual(self._Test("web1.example.com"), ["web1.example.com"])
138 for pattern in ["*.*.*.*", "???", "*.site"]:
139 self.assertEqual(self._Test(pattern), ["sub.site.example.com"])
141 self.assertEqual(self._Test("node1"), [
145 self.assertEqual(self._Test("node?*.example.*"), [
147 "node2-0.example.com",
148 "node2-1.example.com",
151 self.assertEqual(self._Test("*-?"), [
152 "node2-0.example.com",
153 "node2-1.example.com",
155 self.assertEqual(self._Test("node2-?.example.com"), [
156 "node2-0.example.com",
157 "node2-1.example.com",
161 class TestFormatUnit(unittest.TestCase):
162 """Test case for the FormatUnit function"""
165 self.assertEqual(utils.FormatUnit(1, "h"), "1M")
166 self.assertEqual(utils.FormatUnit(100, "h"), "100M")
167 self.assertEqual(utils.FormatUnit(1023, "h"), "1023M")
169 self.assertEqual(utils.FormatUnit(1, "m"), "1")
170 self.assertEqual(utils.FormatUnit(100, "m"), "100")
171 self.assertEqual(utils.FormatUnit(1023, "m"), "1023")
173 self.assertEqual(utils.FormatUnit(1024, "m"), "1024")
174 self.assertEqual(utils.FormatUnit(1536, "m"), "1536")
175 self.assertEqual(utils.FormatUnit(17133, "m"), "17133")
176 self.assertEqual(utils.FormatUnit(1024 * 1024 - 1, "m"), "1048575")
179 self.assertEqual(utils.FormatUnit(1024, "h"), "1.0G")
180 self.assertEqual(utils.FormatUnit(1536, "h"), "1.5G")
181 self.assertEqual(utils.FormatUnit(17133, "h"), "16.7G")
182 self.assertEqual(utils.FormatUnit(1024 * 1024 - 1, "h"), "1024.0G")
184 self.assertEqual(utils.FormatUnit(1024, "g"), "1.0")
185 self.assertEqual(utils.FormatUnit(1536, "g"), "1.5")
186 self.assertEqual(utils.FormatUnit(17133, "g"), "16.7")
187 self.assertEqual(utils.FormatUnit(1024 * 1024 - 1, "g"), "1024.0")
189 self.assertEqual(utils.FormatUnit(1024 * 1024, "g"), "1024.0")
190 self.assertEqual(utils.FormatUnit(5120 * 1024, "g"), "5120.0")
191 self.assertEqual(utils.FormatUnit(29829 * 1024, "g"), "29829.0")
194 self.assertEqual(utils.FormatUnit(1024 * 1024, "h"), "1.0T")
195 self.assertEqual(utils.FormatUnit(5120 * 1024, "h"), "5.0T")
196 self.assertEqual(utils.FormatUnit(29829 * 1024, "h"), "29.1T")
198 self.assertEqual(utils.FormatUnit(1024 * 1024, "t"), "1.0")
199 self.assertEqual(utils.FormatUnit(5120 * 1024, "t"), "5.0")
200 self.assertEqual(utils.FormatUnit(29829 * 1024, "t"), "29.1")
202 def testErrors(self):
203 self.assertRaises(errors.ProgrammerError, utils.FormatUnit, 1, "a")
206 class TestParseUnit(unittest.TestCase):
207 """Test case for the ParseUnit function"""
210 ("M", 1), ("G", 1024), ("T", 1024 * 1024),
211 ("MB", 1), ("GB", 1024), ("TB", 1024 * 1024),
212 ("MiB", 1), ("GiB", 1024), ("TiB", 1024 * 1024))
214 def testRounding(self):
215 self.assertEqual(utils.ParseUnit("0"), 0)
216 self.assertEqual(utils.ParseUnit("1"), 4)
217 self.assertEqual(utils.ParseUnit("2"), 4)
218 self.assertEqual(utils.ParseUnit("3"), 4)
220 self.assertEqual(utils.ParseUnit("124"), 124)
221 self.assertEqual(utils.ParseUnit("125"), 128)
222 self.assertEqual(utils.ParseUnit("126"), 128)
223 self.assertEqual(utils.ParseUnit("127"), 128)
224 self.assertEqual(utils.ParseUnit("128"), 128)
225 self.assertEqual(utils.ParseUnit("129"), 132)
226 self.assertEqual(utils.ParseUnit("130"), 132)
228 def testFloating(self):
229 self.assertEqual(utils.ParseUnit("0"), 0)
230 self.assertEqual(utils.ParseUnit("0.5"), 4)
231 self.assertEqual(utils.ParseUnit("1.75"), 4)
232 self.assertEqual(utils.ParseUnit("1.99"), 4)
233 self.assertEqual(utils.ParseUnit("2.00"), 4)
234 self.assertEqual(utils.ParseUnit("2.01"), 4)
235 self.assertEqual(utils.ParseUnit("3.99"), 4)
236 self.assertEqual(utils.ParseUnit("4.00"), 4)
237 self.assertEqual(utils.ParseUnit("4.01"), 8)
238 self.assertEqual(utils.ParseUnit("1.5G"), 1536)
239 self.assertEqual(utils.ParseUnit("1.8G"), 1844)
240 self.assertEqual(utils.ParseUnit("8.28T"), 8682212)
242 def testSuffixes(self):
243 for sep in ("", " ", " ", "\t", "\t "):
244 for suffix, scale in self.SCALES:
245 for func in (lambda x: x, str.lower, str.upper):
246 self.assertEqual(utils.ParseUnit("1024" + sep + func(suffix)),
249 def testInvalidInput(self):
250 for sep in ("-", "_", ",", "a"):
251 for suffix, _ in self.SCALES:
252 self.assertRaises(errors.UnitParseError, utils.ParseUnit,
255 for suffix, _ in self.SCALES:
256 self.assertRaises(errors.UnitParseError, utils.ParseUnit,
260 class TestShellQuoting(unittest.TestCase):
261 """Test case for shell quoting functions"""
263 def testShellQuote(self):
264 self.assertEqual(utils.ShellQuote("abc"), "abc")
265 self.assertEqual(utils.ShellQuote('ab"c'), "'ab\"c'")
266 self.assertEqual(utils.ShellQuote("a'bc"), "'a'\\''bc'")
267 self.assertEqual(utils.ShellQuote("a b c"), "'a b c'")
268 self.assertEqual(utils.ShellQuote("a b\\ c"), "'a b\\ c'")
270 def testShellQuoteArgs(self):
271 self.assertEqual(utils.ShellQuoteArgs(["a", "b", "c"]), "a b c")
272 self.assertEqual(utils.ShellQuoteArgs(['a', 'b"', 'c']), "a 'b\"' c")
273 self.assertEqual(utils.ShellQuoteArgs(['a', 'b\'', 'c']), "a 'b'\\\''' c")
276 class TestShellWriter(unittest.TestCase):
279 sw = utils.ShellWriter(buf)
280 sw.Write("#!/bin/bash")
281 sw.Write("if true; then")
284 sw.Write("echo true")
286 sw.Write("for i in 1 2 3")
290 self.assertEqual(sw._indent, 2)
297 sw.Write("echo %s", utils.ShellQuote("Hello World"))
300 self.assertEqual(sw._indent, 0)
302 output = buf.getvalue()
304 self.assert_(output.endswith("\n"))
306 lines = output.splitlines()
307 self.assertEqual(len(lines), 9)
308 self.assertEqual(lines[0], "#!/bin/bash")
309 self.assert_(re.match(r"^\s+date$", lines[5]))
310 self.assertEqual(lines[7], "echo 'Hello World'")
314 sw = utils.ShellWriter(buf)
316 self.assertEqual(buf.getvalue(), "")
318 def testEmptyNoIndent(self):
320 sw = utils.ShellWriter(buf, indent=False)
322 self.assertEqual(buf.getvalue(), "")
325 def _AddLevel(cls, sw, level):
331 # Add empty line, it should not be indented
334 cls._AddLevel(sw, level + 1)
338 def testEmptyLines(self):
340 sw = utils.ShellWriter(buf)
342 self._AddLevel(sw, 1)
344 self.assertEqual(buf.getvalue(),
345 "".join("\n%s%s\n" % (i * " ", i) for i in range(1, 6)))
347 def testEmptyLinesNoIndent(self):
349 sw = utils.ShellWriter(buf, indent=False)
351 self._AddLevel(sw, 1)
353 self.assertEqual(buf.getvalue(),
354 "".join("\n%s\n" % i for i in range(1, 6)))
357 class TestNormalizeAndValidateMac(unittest.TestCase):
358 def testInvalid(self):
359 for i in ["xxx", "00:11:22:33:44:55:66", "zz:zz:zz:zz:zz:zz"]:
360 self.assertRaises(errors.OpPrereqError, utils.NormalizeAndValidateMac, i)
362 def testNormalization(self):
363 for mac in ["aa:bb:cc:dd:ee:ff", "00:AA:11:bB:22:cc"]:
364 self.assertEqual(utils.NormalizeAndValidateMac(mac), mac.lower())
367 class TestNormalizeAndValidateThreeOctetMacPrefix(unittest.TestCase):
368 def testInvalid(self):
369 for i in ["xxx", "00:11:22:33:44:55:66", "zz:zz:zz:zz:zz:zz",
370 "aa:bb:cc:dd:ee:ff", "00:AA:11:bB:22:cc",
372 self.assertRaises(errors.OpPrereqError,
373 utils.NormalizeAndValidateThreeOctetMacPrefix, i)
375 def testNormalization(self):
376 for mac in ["aa:bb:cc", "00:AA:11"]:
377 self.assertEqual(utils.NormalizeAndValidateThreeOctetMacPrefix(mac),
381 class TestSafeEncode(unittest.TestCase):
382 """Test case for SafeEncode"""
385 for txt in [string.digits, string.letters, string.punctuation]:
386 self.failUnlessEqual(txt, utils.SafeEncode(txt))
388 def testDoubleEncode(self):
390 txt = utils.SafeEncode(chr(i))
391 self.failUnlessEqual(txt, utils.SafeEncode(txt))
393 def testUnicode(self):
394 # 1024 is high enough to catch non-direct ASCII mappings
395 for i in range(1024):
396 txt = utils.SafeEncode(unichr(i))
397 self.failUnlessEqual(txt, utils.SafeEncode(txt))
400 class TestUnescapeAndSplit(unittest.TestCase):
401 """Testing case for UnescapeAndSplit"""
404 # testing more that one separator for regexp safety
405 self._seps = [",", "+", ".", ":"]
407 def testSimple(self):
408 a = ["a", "b", "c", "d"]
409 for sep in self._seps:
410 self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), a)
412 def testEscape(self):
413 for sep in self._seps:
414 a = ["a", "b\\" + sep + "c", "d"]
415 b = ["a", "b" + sep + "c", "d"]
416 self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
418 def testDoubleEscape(self):
419 for sep in self._seps:
420 a = ["a", "b\\\\", "c", "d"]
421 b = ["a", "b\\", "c", "d"]
422 self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
424 def testThreeEscape(self):
425 for sep in self._seps:
426 a = ["a", "b\\\\\\" + sep + "c", "d"]
427 b = ["a", "b\\" + sep + "c", "d"]
428 self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
430 def testEscapeAtEnd(self):
431 for sep in self._seps:
432 self.assertEqual(utils.UnescapeAndSplit("\\", sep=sep), ["\\"])
434 a = ["a", "b\\", "c"]
435 b = ["a", "b" + sep + "c\\"]
436 self.assertEqual(utils.UnescapeAndSplit("%s\\" % sep.join(a), sep=sep), b)
438 a = ["\\" + sep, "\\" + sep, "c", "d\\.moo"]
439 b = [sep, sep, "c", "d.moo\\"]
440 self.assertEqual(utils.UnescapeAndSplit("%s\\" % sep.join(a), sep=sep), b)
442 def testMultipleEscapes(self):
443 for sep in self._seps:
444 a = ["a", "b\\" + sep + "c", "d\\" + sep + "e\\" + sep + "f", "g"]
445 b = ["a", "b" + sep + "c", "d" + sep + "e" + sep + "f", "g"]
446 self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
449 class TestCommaJoin(unittest.TestCase):
451 self.assertEqual(utils.CommaJoin([]), "")
452 self.assertEqual(utils.CommaJoin([1, 2, 3]), "1, 2, 3")
453 self.assertEqual(utils.CommaJoin(["Hello"]), "Hello")
454 self.assertEqual(utils.CommaJoin(["Hello", "World"]), "Hello, World")
455 self.assertEqual(utils.CommaJoin(["Hello", "World", 99]),
459 class TestFormatTime(unittest.TestCase):
460 """Testing case for FormatTime"""
463 def _TestInProcess(tz, timestamp, usecs, expected):
464 os.environ["TZ"] = tz
466 return utils.FormatTime(timestamp, usecs=usecs) == expected
468 def _Test(self, *args):
469 # Need to use separate process as we want to change TZ
470 self.assert_(utils.RunInSeparateProcess(self._TestInProcess, *args))
473 self._Test("UTC", 0, None, "1970-01-01 00:00:00")
474 self._Test("America/Sao_Paulo", 1292606926, None, "2010-12-17 15:28:46")
475 self._Test("Europe/London", 1292606926, None, "2010-12-17 17:28:46")
476 self._Test("Europe/Zurich", 1292606926, None, "2010-12-17 18:28:46")
477 self._Test("Europe/Zurich", 1332944288, 8787, "2012-03-28 16:18:08.008787")
478 self._Test("Australia/Sydney", 1292606926, None, "2010-12-18 04:28:46")
479 self._Test("Australia/Sydney", 1292606926, None, "2010-12-18 04:28:46")
480 self._Test("Australia/Sydney", 1292606926, 999999,
481 "2010-12-18 04:28:46.999999")
484 self.failUnlessEqual(utils.FormatTime(None), "N/A")
486 def testInvalid(self):
487 self.failUnlessEqual(utils.FormatTime(()), "N/A")
490 # tests that we accept time.time input
491 utils.FormatTime(time.time())
492 # tests that we accept int input
493 utils.FormatTime(int(time.time()))
496 class TestFormatSeconds(unittest.TestCase):
498 self.assertEqual(utils.FormatSeconds(1), "1s")
499 self.assertEqual(utils.FormatSeconds(3600), "1h 0m 0s")
500 self.assertEqual(utils.FormatSeconds(3599), "59m 59s")
501 self.assertEqual(utils.FormatSeconds(7200), "2h 0m 0s")
502 self.assertEqual(utils.FormatSeconds(7201), "2h 0m 1s")
503 self.assertEqual(utils.FormatSeconds(7281), "2h 1m 21s")
504 self.assertEqual(utils.FormatSeconds(29119), "8h 5m 19s")
505 self.assertEqual(utils.FormatSeconds(19431228), "224d 21h 33m 48s")
506 self.assertEqual(utils.FormatSeconds(-1), "-1s")
507 self.assertEqual(utils.FormatSeconds(-282), "-282s")
508 self.assertEqual(utils.FormatSeconds(-29119), "-29119s")
511 self.assertEqual(utils.FormatSeconds(1.3), "1s")
512 self.assertEqual(utils.FormatSeconds(1.9), "2s")
513 self.assertEqual(utils.FormatSeconds(3912.12311), "1h 5m 12s")
514 self.assertEqual(utils.FormatSeconds(3912.8), "1h 5m 13s")
517 class TestLineSplitter(unittest.TestCase):
520 ls = utils.LineSplitter(lines.append)
521 ls.write("Hello World\n")
522 self.assertEqual(lines, [])
523 ls.write("Foo\n Bar\r\n ")
526 self.assertEqual(lines, [])
528 self.assertEqual(lines, ["Hello World", "Foo", " Bar"])
530 self.assertEqual(lines, ["Hello World", "Foo", " Bar", " BazMoo"])
532 def _testExtra(self, line, all_lines, p1, p2):
533 self.assertEqual(p1, 999)
534 self.assertEqual(p2, "extra")
535 all_lines.append(line)
537 def testExtraArgsNoFlush(self):
539 ls = utils.LineSplitter(self._testExtra, lines, 999, "extra")
540 ls.write("\n\nHello World\n")
541 ls.write("Foo\n Bar\r\n ")
544 ls.write("Moo\n\nx\n")
545 self.assertEqual(lines, [])
547 self.assertEqual(lines, ["", "", "Hello World", "Foo", " Bar", " BazMoo",
551 class TestIsValidShellParam(unittest.TestCase):
557 self.assertEqual(utils.IsValidShellParam(val), result)
560 class TestBuildShellCmd(unittest.TestCase):
562 self.assertRaises(errors.ProgrammerError, utils.BuildShellCmd,
564 self.assertEqual(utils.BuildShellCmd("ls %s", "ab"), "ls ab")
567 class TestOrdinal(unittest.TestCase):
570 0: "0th", 1: "1st", 2: "2nd", 3: "3rd", 4: "4th", 5: "5th", 6: "6th",
571 7: "7th", 8: "8th", 9: "9th", 10: "10th", 11: "11th", 12: "12th",
572 13: "13th", 14: "14th", 15: "15th", 16: "16th", 17: "17th",
573 18: "18th", 19: "19th", 20: "20th", 21: "21st", 25: "25th", 30: "30th",
574 32: "32nd", 40: "40th", 50: "50th", 55: "55th", 60: "60th", 62: "62nd",
575 70: "70th", 80: "80th", 83: "83rd", 90: "90th", 91: "91st",
576 582: "582nd", 999: "999th",
579 for value, ordinal in checks.items():
580 self.assertEqual(utils.FormatOrdinal(value), ordinal)
583 class TestTruncate(unittest.TestCase):
584 def _Test(self, text, length):
585 result = utils.Truncate(text, length)
586 self.assertTrue(len(result) <= length)
590 self.assertEqual(self._Test("", 80), "")
591 self.assertEqual(self._Test("abc", 4), "abc")
592 self.assertEqual(self._Test("Hello World", 80), "Hello World")
593 self.assertEqual(self._Test("Hello World", 4), "H...")
594 self.assertEqual(self._Test("Hello World", 5), "He...")
596 for i in [4, 10, 100]:
597 data = i * "FooBarBaz"
598 self.assertEqual(self._Test(data, len(data)), data)
600 for (length, exp) in [(8, u"T\u00e4st\u2026xyz"), (7, u"T\u00e4st...")]:
601 self.assertEqual(self._Test(u"T\u00e4st\u2026xyz", length), exp)
603 self.assertEqual(self._Test(range(100), 20), "[0, 1, 2, 3, 4, 5...")
607 self.assertRaises(AssertionError, utils.Truncate, "", i)
610 class TestFilterEmptyLinesAndComments(unittest.TestCase):
612 self.assertEqual(utils.FilterEmptyLinesAndComments(""), [])
613 self.assertEqual(utils.FilterEmptyLinesAndComments("\n"), [])
614 self.assertEqual(utils.FilterEmptyLinesAndComments("\n" * 100), [])
615 self.assertEqual(utils.FilterEmptyLinesAndComments("\n \n\t \n"), [])
631 self.assertEqual(utils.FilterEmptyLinesAndComments(text), [
642 if __name__ == "__main__":
643 testutils.GanetiTestProgram()