#!/usr/bin/python
#
# Copyright (C) 2011 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# 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
# 02110-1301, USA.
"""Script for testing ganeti.utils.text"""
import re
import string
import time
import unittest
import os
from cStringIO import StringIO
from ganeti import constants
from ganeti import utils
from ganeti import errors
import testutils
class TestMatchNameComponent(unittest.TestCase):
"""Test case for the MatchNameComponent function"""
def testEmptyList(self):
"""Test that there is no match against an empty list"""
self.failUnlessEqual(utils.MatchNameComponent("", []), None)
self.failUnlessEqual(utils.MatchNameComponent("test", []), None)
def testSingleMatch(self):
"""Test that a single match is performed correctly"""
mlist = ["test1.example.com", "test2.example.com", "test3.example.com"]
for key in "test2", "test2.example", "test2.example.com":
self.failUnlessEqual(utils.MatchNameComponent(key, mlist), mlist[1])
def testMultipleMatches(self):
"""Test that a multiple match is returned as None"""
mlist = ["test1.example.com", "test1.example.org", "test1.example.net"]
for key in "test1", "test1.example":
self.failUnlessEqual(utils.MatchNameComponent(key, mlist), None)
def testFullMatch(self):
"""Test that a full match is returned correctly"""
key1 = "test1"
key2 = "test1.example"
mlist = [key2, key2 + ".com"]
self.failUnlessEqual(utils.MatchNameComponent(key1, mlist), None)
self.failUnlessEqual(utils.MatchNameComponent(key2, mlist), key2)
def testCaseInsensitivePartialMatch(self):
"""Test for the case_insensitive keyword"""
mlist = ["test1.example.com", "test2.example.net"]
self.assertEqual(utils.MatchNameComponent("test2", mlist,
case_sensitive=False),
"test2.example.net")
self.assertEqual(utils.MatchNameComponent("Test2", mlist,
case_sensitive=False),
"test2.example.net")
self.assertEqual(utils.MatchNameComponent("teSt2", mlist,
case_sensitive=False),
"test2.example.net")
self.assertEqual(utils.MatchNameComponent("TeSt2", mlist,
case_sensitive=False),
"test2.example.net")
def testCaseInsensitiveFullMatch(self):
mlist = ["ts1.ex", "ts1.ex.org", "ts2.ex", "Ts2.ex"]
# Between the two ts1 a full string match non-case insensitive should work
self.assertEqual(utils.MatchNameComponent("Ts1", mlist,
case_sensitive=False),
None)
self.assertEqual(utils.MatchNameComponent("Ts1.ex", mlist,
case_sensitive=False),
"ts1.ex")
self.assertEqual(utils.MatchNameComponent("ts1.ex", mlist,
case_sensitive=False),
"ts1.ex")
# Between the two ts2 only case differs, so only case-match works
self.assertEqual(utils.MatchNameComponent("ts2.ex", mlist,
case_sensitive=False),
"ts2.ex")
self.assertEqual(utils.MatchNameComponent("Ts2.ex", mlist,
case_sensitive=False),
"Ts2.ex")
self.assertEqual(utils.MatchNameComponent("TS2.ex", mlist,
case_sensitive=False),
None)
class TestFormatUnit(unittest.TestCase):
"""Test case for the FormatUnit function"""
def testMiB(self):
self.assertEqual(utils.FormatUnit(1, "h"), "1M")
self.assertEqual(utils.FormatUnit(100, "h"), "100M")
self.assertEqual(utils.FormatUnit(1023, "h"), "1023M")
self.assertEqual(utils.FormatUnit(1, "m"), "1")
self.assertEqual(utils.FormatUnit(100, "m"), "100")
self.assertEqual(utils.FormatUnit(1023, "m"), "1023")
self.assertEqual(utils.FormatUnit(1024, "m"), "1024")
self.assertEqual(utils.FormatUnit(1536, "m"), "1536")
self.assertEqual(utils.FormatUnit(17133, "m"), "17133")
self.assertEqual(utils.FormatUnit(1024 * 1024 - 1, "m"), "1048575")
def testGiB(self):
self.assertEqual(utils.FormatUnit(1024, "h"), "1.0G")
self.assertEqual(utils.FormatUnit(1536, "h"), "1.5G")
self.assertEqual(utils.FormatUnit(17133, "h"), "16.7G")
self.assertEqual(utils.FormatUnit(1024 * 1024 - 1, "h"), "1024.0G")
self.assertEqual(utils.FormatUnit(1024, "g"), "1.0")
self.assertEqual(utils.FormatUnit(1536, "g"), "1.5")
self.assertEqual(utils.FormatUnit(17133, "g"), "16.7")
self.assertEqual(utils.FormatUnit(1024 * 1024 - 1, "g"), "1024.0")
self.assertEqual(utils.FormatUnit(1024 * 1024, "g"), "1024.0")
self.assertEqual(utils.FormatUnit(5120 * 1024, "g"), "5120.0")
self.assertEqual(utils.FormatUnit(29829 * 1024, "g"), "29829.0")
def testTiB(self):
self.assertEqual(utils.FormatUnit(1024 * 1024, "h"), "1.0T")
self.assertEqual(utils.FormatUnit(5120 * 1024, "h"), "5.0T")
self.assertEqual(utils.FormatUnit(29829 * 1024, "h"), "29.1T")
self.assertEqual(utils.FormatUnit(1024 * 1024, "t"), "1.0")
self.assertEqual(utils.FormatUnit(5120 * 1024, "t"), "5.0")
self.assertEqual(utils.FormatUnit(29829 * 1024, "t"), "29.1")
def testErrors(self):
self.assertRaises(errors.ProgrammerError, utils.FormatUnit, 1, "a")
class TestParseUnit(unittest.TestCase):
"""Test case for the ParseUnit function"""
SCALES = (("", 1),
("M", 1), ("G", 1024), ("T", 1024 * 1024),
("MB", 1), ("GB", 1024), ("TB", 1024 * 1024),
("MiB", 1), ("GiB", 1024), ("TiB", 1024 * 1024))
def testRounding(self):
self.assertEqual(utils.ParseUnit("0"), 0)
self.assertEqual(utils.ParseUnit("1"), 4)
self.assertEqual(utils.ParseUnit("2"), 4)
self.assertEqual(utils.ParseUnit("3"), 4)
self.assertEqual(utils.ParseUnit("124"), 124)
self.assertEqual(utils.ParseUnit("125"), 128)
self.assertEqual(utils.ParseUnit("126"), 128)
self.assertEqual(utils.ParseUnit("127"), 128)
self.assertEqual(utils.ParseUnit("128"), 128)
self.assertEqual(utils.ParseUnit("129"), 132)
self.assertEqual(utils.ParseUnit("130"), 132)
def testFloating(self):
self.assertEqual(utils.ParseUnit("0"), 0)
self.assertEqual(utils.ParseUnit("0.5"), 4)
self.assertEqual(utils.ParseUnit("1.75"), 4)
self.assertEqual(utils.ParseUnit("1.99"), 4)
self.assertEqual(utils.ParseUnit("2.00"), 4)
self.assertEqual(utils.ParseUnit("2.01"), 4)
self.assertEqual(utils.ParseUnit("3.99"), 4)
self.assertEqual(utils.ParseUnit("4.00"), 4)
self.assertEqual(utils.ParseUnit("4.01"), 8)
self.assertEqual(utils.ParseUnit("1.5G"), 1536)
self.assertEqual(utils.ParseUnit("1.8G"), 1844)
self.assertEqual(utils.ParseUnit("8.28T"), 8682212)
def testSuffixes(self):
for sep in ("", " ", " ", "\t", "\t "):
for suffix, scale in self.SCALES:
for func in (lambda x: x, str.lower, str.upper):
self.assertEqual(utils.ParseUnit("1024" + sep + func(suffix)),
1024 * scale)
def testInvalidInput(self):
for sep in ("-", "_", ",", "a"):
for suffix, _ in self.SCALES:
self.assertRaises(errors.UnitParseError, utils.ParseUnit,
"1" + sep + suffix)
for suffix, _ in self.SCALES:
self.assertRaises(errors.UnitParseError, utils.ParseUnit,
"1,3" + suffix)
class TestShellQuoting(unittest.TestCase):
"""Test case for shell quoting functions"""
def testShellQuote(self):
self.assertEqual(utils.ShellQuote('abc'), "abc")
self.assertEqual(utils.ShellQuote('ab"c'), "'ab\"c'")
self.assertEqual(utils.ShellQuote("a'bc"), "'a'\\''bc'")
self.assertEqual(utils.ShellQuote("a b c"), "'a b c'")
self.assertEqual(utils.ShellQuote("a b\\ c"), "'a b\\ c'")
def testShellQuoteArgs(self):
self.assertEqual(utils.ShellQuoteArgs(['a', 'b', 'c']), "a b c")
self.assertEqual(utils.ShellQuoteArgs(['a', 'b"', 'c']), "a 'b\"' c")
self.assertEqual(utils.ShellQuoteArgs(['a', 'b\'', 'c']), "a 'b'\\\''' c")
class TestShellWriter(unittest.TestCase):
def test(self):
buf = StringIO()
sw = utils.ShellWriter(buf)
sw.Write("#!/bin/bash")
sw.Write("if true; then")
sw.IncIndent()
try:
sw.Write("echo true")
sw.Write("for i in 1 2 3")
sw.Write("do")
sw.IncIndent()
try:
self.assertEqual(sw._indent, 2)
sw.Write("date")
finally:
sw.DecIndent()
sw.Write("done")
finally:
sw.DecIndent()
sw.Write("echo %s", utils.ShellQuote("Hello World"))
sw.Write("exit 0")
self.assertEqual(sw._indent, 0)
output = buf.getvalue()
self.assert_(output.endswith("\n"))
lines = output.splitlines()
self.assertEqual(len(lines), 9)
self.assertEqual(lines[0], "#!/bin/bash")
self.assert_(re.match(r"^\s+date$", lines[5]))
self.assertEqual(lines[7], "echo 'Hello World'")
def testEmpty(self):
buf = StringIO()
sw = utils.ShellWriter(buf)
sw = None
self.assertEqual(buf.getvalue(), "")
class TestNormalizeAndValidateMac(unittest.TestCase):
def testInvalid(self):
self.assertRaises(errors.OpPrereqError,
utils.NormalizeAndValidateMac, "xxx")
def testNormalization(self):
for mac in ["aa:bb:cc:dd:ee:ff", "00:AA:11:bB:22:cc"]:
self.assertEqual(utils.NormalizeAndValidateMac(mac), mac.lower())
class TestSafeEncode(unittest.TestCase):
"""Test case for SafeEncode"""
def testAscii(self):
for txt in [string.digits, string.letters, string.punctuation]:
self.failUnlessEqual(txt, utils.SafeEncode(txt))
def testDoubleEncode(self):
for i in range(255):
txt = utils.SafeEncode(chr(i))
self.failUnlessEqual(txt, utils.SafeEncode(txt))
def testUnicode(self):
# 1024 is high enough to catch non-direct ASCII mappings
for i in range(1024):
txt = utils.SafeEncode(unichr(i))
self.failUnlessEqual(txt, utils.SafeEncode(txt))
class TestUnescapeAndSplit(unittest.TestCase):
"""Testing case for UnescapeAndSplit"""
def setUp(self):
# testing more that one separator for regexp safety
self._seps = [",", "+", "."]
def testSimple(self):
a = ["a", "b", "c", "d"]
for sep in self._seps:
self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), a)
def testEscape(self):
for sep in self._seps:
a = ["a", "b\\" + sep + "c", "d"]
b = ["a", "b" + sep + "c", "d"]
self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
def testDoubleEscape(self):
for sep in self._seps:
a = ["a", "b\\\\", "c", "d"]
b = ["a", "b\\", "c", "d"]
self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
def testThreeEscape(self):
for sep in self._seps:
a = ["a", "b\\\\\\" + sep + "c", "d"]
b = ["a", "b\\" + sep + "c", "d"]
self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
class TestCommaJoin(unittest.TestCase):
def test(self):
self.assertEqual(utils.CommaJoin([]), "")
self.assertEqual(utils.CommaJoin([1, 2, 3]), "1, 2, 3")
self.assertEqual(utils.CommaJoin(["Hello"]), "Hello")
self.assertEqual(utils.CommaJoin(["Hello", "World"]), "Hello, World")
self.assertEqual(utils.CommaJoin(["Hello", "World", 99]),
"Hello, World, 99")
class TestFormatTime(unittest.TestCase):
"""Testing case for FormatTime"""
@staticmethod
def _TestInProcess(tz, timestamp, expected):
os.environ["TZ"] = tz
time.tzset()
return utils.FormatTime(timestamp) == expected
def _Test(self, *args):
# Need to use separate process as we want to change TZ
self.assert_(utils.RunInSeparateProcess(self._TestInProcess, *args))
def test(self):
self._Test("UTC", 0, "1970-01-01 00:00:00")
self._Test("America/Sao_Paulo", 1292606926, "2010-12-17 15:28:46")
self._Test("Europe/London", 1292606926, "2010-12-17 17:28:46")
self._Test("Europe/Zurich", 1292606926, "2010-12-17 18:28:46")
self._Test("Australia/Sydney", 1292606926, "2010-12-18 04:28:46")
def testNone(self):
self.failUnlessEqual(utils.FormatTime(None), "N/A")
def testInvalid(self):
self.failUnlessEqual(utils.FormatTime(()), "N/A")
def testNow(self):
# tests that we accept time.time input
utils.FormatTime(time.time())
# tests that we accept int input
utils.FormatTime(int(time.time()))
class TestFormatSeconds(unittest.TestCase):
def test(self):
self.assertEqual(utils.FormatSeconds(1), "1s")
self.assertEqual(utils.FormatSeconds(3600), "1h 0m 0s")
self.assertEqual(utils.FormatSeconds(3599), "59m 59s")
self.assertEqual(utils.FormatSeconds(7200), "2h 0m 0s")
self.assertEqual(utils.FormatSeconds(7201), "2h 0m 1s")
self.assertEqual(utils.FormatSeconds(7281), "2h 1m 21s")
self.assertEqual(utils.FormatSeconds(29119), "8h 5m 19s")
self.assertEqual(utils.FormatSeconds(19431228), "224d 21h 33m 48s")
self.assertEqual(utils.FormatSeconds(-1), "-1s")
self.assertEqual(utils.FormatSeconds(-282), "-282s")
self.assertEqual(utils.FormatSeconds(-29119), "-29119s")
def testFloat(self):
self.assertEqual(utils.FormatSeconds(1.3), "1s")
self.assertEqual(utils.FormatSeconds(1.9), "2s")
self.assertEqual(utils.FormatSeconds(3912.12311), "1h 5m 12s")
self.assertEqual(utils.FormatSeconds(3912.8), "1h 5m 13s")
class TestLineSplitter(unittest.TestCase):
def test(self):
lines = []
ls = utils.LineSplitter(lines.append)
ls.write("Hello World\n")
self.assertEqual(lines, [])
ls.write("Foo\n Bar\r\n ")
ls.write("Baz")
ls.write("Moo")
self.assertEqual(lines, [])
ls.flush()
self.assertEqual(lines, ["Hello World", "Foo", " Bar"])
ls.close()
self.assertEqual(lines, ["Hello World", "Foo", " Bar", " BazMoo"])
def _testExtra(self, line, all_lines, p1, p2):
self.assertEqual(p1, 999)
self.assertEqual(p2, "extra")
all_lines.append(line)
def testExtraArgsNoFlush(self):
lines = []
ls = utils.LineSplitter(self._testExtra, lines, 999, "extra")
ls.write("\n\nHello World\n")
ls.write("Foo\n Bar\r\n ")
ls.write("")
ls.write("Baz")
ls.write("Moo\n\nx\n")
self.assertEqual(lines, [])
ls.close()
self.assertEqual(lines, ["", "", "Hello World", "Foo", " Bar", " BazMoo",
"", "x"])
class TestIsValidShellParam(unittest.TestCase):
def test(self):
for val, result in [
("abc", True),
("ab;cd", False),
]:
self.assertEqual(utils.IsValidShellParam(val), result)
class TestBuildShellCmd(unittest.TestCase):
def test(self):
self.assertRaises(errors.ProgrammerError, utils.BuildShellCmd,
"ls %s", "ab;cd")
self.assertEqual(utils.BuildShellCmd("ls %s", "ab"), "ls ab")
if __name__ == "__main__":
testutils.GanetiTestProgram()