| #!/usr/bin/python |
| # |
| |
| # Copyright (C) 2011, 2012 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.algo""" |
| |
| import unittest |
| import random |
| import operator |
| |
| from ganeti import constants |
| from ganeti import compat |
| from ganeti.utils import algo |
| |
| import testutils |
| |
| |
| class TestUniqueSequence(unittest.TestCase): |
| """Test case for UniqueSequence""" |
| |
| def _test(self, input, expected): |
| self.assertEqual(algo.UniqueSequence(input), expected) |
| |
| def runTest(self): |
| # Ordered input |
| self._test([1, 2, 3], [1, 2, 3]) |
| self._test([1, 1, 2, 2, 3, 3], [1, 2, 3]) |
| self._test([1, 2, 2, 3], [1, 2, 3]) |
| self._test([1, 2, 3, 3], [1, 2, 3]) |
| |
| # Unordered input |
| self._test([1, 2, 3, 1, 2, 3], [1, 2, 3]) |
| self._test([1, 1, 2, 3, 3, 1, 2], [1, 2, 3]) |
| |
| # Strings |
| self._test(["a", "a"], ["a"]) |
| self._test(["a", "b"], ["a", "b"]) |
| self._test(["a", "b", "a"], ["a", "b"]) |
| |
| |
| class TestFindDuplicates(unittest.TestCase): |
| """Test case for FindDuplicates""" |
| |
| def _Test(self, seq, expected): |
| result = algo.FindDuplicates(seq) |
| self.assertEqual(result, algo.UniqueSequence(result)) |
| self.assertEqual(set(result), set(expected)) |
| |
| def test(self): |
| self._Test([], []) |
| self._Test([1, 2, 3], []) |
| self._Test([9, 8, 8, 0, 5, 1, 7, 0, 6, 7], [8, 0, 7]) |
| for exp in [[1, 2, 3], [3, 2, 1]]: |
| self._Test([1, 1, 2, 2, 3, 3], exp) |
| |
| self._Test(["A", "a", "B"], []) |
| self._Test(["a", "A", "a", "B"], ["a"]) |
| self._Test("Hello World out there!", ["e", " ", "o", "r", "t", "l"]) |
| |
| self._Test(self._Gen(False), []) |
| self._Test(self._Gen(True), range(1, 10)) |
| |
| @staticmethod |
| def _Gen(dup): |
| for i in range(10): |
| yield i |
| if dup: |
| for _ in range(i): |
| yield i |
| |
| |
| class TestNiceSort(unittest.TestCase): |
| def test(self): |
| self.assertEqual(algo.NiceSort([]), []) |
| self.assertEqual(algo.NiceSort(["foo"]), ["foo"]) |
| self.assertEqual(algo.NiceSort(["bar", ""]), ["", "bar"]) |
| self.assertEqual(algo.NiceSort([",", "."]), [",", "."]) |
| self.assertEqual(algo.NiceSort(["0.1", "0.2"]), ["0.1", "0.2"]) |
| self.assertEqual(algo.NiceSort(["0;099", "0,099", "0.1", "0.2"]), |
| ["0,099", "0.1", "0.2", "0;099"]) |
| |
| data = ["a0", "a1", "a99", "a20", "a2", "b10", "b70", "b00", "0000"] |
| self.assertEqual(algo.NiceSort(data), |
| ["0000", "a0", "a1", "a2", "a20", "a99", |
| "b00", "b10", "b70"]) |
| |
| data = ["a0-0", "a1-0", "a99-10", "a20-3", "a0-4", "a99-3", "a09-2", |
| "Z", "a9-1", "A", "b"] |
| self.assertEqual(algo.NiceSort(data), |
| ["A", "Z", "a0-0", "a0-4", "a1-0", "a9-1", "a09-2", |
| "a20-3", "a99-3", "a99-10", "b"]) |
| self.assertEqual(algo.NiceSort(data, key=str.lower), |
| ["A", "a0-0", "a0-4", "a1-0", "a9-1", "a09-2", |
| "a20-3", "a99-3", "a99-10", "b", "Z"]) |
| self.assertEqual(algo.NiceSort(data, key=str.upper), |
| ["A", "a0-0", "a0-4", "a1-0", "a9-1", "a09-2", |
| "a20-3", "a99-3", "a99-10", "b", "Z"]) |
| |
| def testLargeA(self): |
| data = [ |
| "Eegah9ei", "xij88brTulHYAv8IEOyU", "3jTwJPtrXOY22bwL2YoW", |
| "Z8Ljf1Pf5eBfNg171wJR", "WvNJd91OoXvLzdEiEXa6", "uHXAyYYftCSG1o7qcCqe", |
| "xpIUJeVT1Rp", "KOt7vn1dWXi", "a07h8feON165N67PIE", "bH4Q7aCu3PUPjK3JtH", |
| "cPRi0lM7HLnSuWA2G9", "KVQqLPDjcPjf8T3oyzjcOsfkb", |
| "guKJkXnkULealVC8CyF1xefym", "pqF8dkU5B1cMnyZuREaSOADYx", |
| ] |
| self.assertEqual(algo.NiceSort(data), [ |
| "3jTwJPtrXOY22bwL2YoW", "Eegah9ei", "KOt7vn1dWXi", |
| "KVQqLPDjcPjf8T3oyzjcOsfkb", "WvNJd91OoXvLzdEiEXa6", |
| "Z8Ljf1Pf5eBfNg171wJR", "a07h8feON165N67PIE", "bH4Q7aCu3PUPjK3JtH", |
| "cPRi0lM7HLnSuWA2G9", "guKJkXnkULealVC8CyF1xefym", |
| "pqF8dkU5B1cMnyZuREaSOADYx", "uHXAyYYftCSG1o7qcCqe", |
| "xij88brTulHYAv8IEOyU", "xpIUJeVT1Rp" |
| ]) |
| |
| def testLargeB(self): |
| data = [ |
| "inst-0.0.0.0-0.0.0.0", |
| "inst-0.1.0.0-0.0.0.0", |
| "inst-0.2.0.0-0.0.0.0", |
| "inst-0.2.1.0-0.0.0.0", |
| "inst-0.2.2.0-0.0.0.0", |
| "inst-0.2.2.0-0.0.0.9", |
| "inst-0.2.2.0-0.0.3.9", |
| "inst-0.2.2.0-0.2.0.9", |
| "inst-0.2.2.0-0.9.0.9", |
| "inst-0.20.2.0-0.0.0.0", |
| "inst-0.20.2.0-0.9.0.9", |
| "inst-10.020.2.0-0.9.0.10", |
| "inst-15.020.2.0-0.9.1.00", |
| "inst-100.020.2.0-0.9.0.9", |
| |
| # Only the last group, not converted to a number anymore, differs |
| "inst-100.020.2.0a999", |
| "inst-100.020.2.0b000", |
| "inst-100.020.2.0c10", |
| "inst-100.020.2.0c101", |
| "inst-100.020.2.0c2", |
| "inst-100.020.2.0c20", |
| "inst-100.020.2.0c3", |
| "inst-100.020.2.0c39123", |
| ] |
| |
| rnd = random.Random(16205) |
| for _ in range(10): |
| testdata = data[:] |
| rnd.shuffle(testdata) |
| assert testdata != data |
| self.assertEqual(algo.NiceSort(testdata), data) |
| |
| class _CallCount: |
| def __init__(self, fn): |
| self.count = 0 |
| self.fn = fn |
| |
| def __call__(self, *args): |
| self.count += 1 |
| return self.fn(*args) |
| |
| def testKeyfuncA(self): |
| # Generate some random numbers |
| rnd = random.Random(21131) |
| numbers = [rnd.randint(0, 10000) for _ in range(999)] |
| assert numbers != sorted(numbers) |
| |
| # Convert to hex |
| data = [hex(i) for i in numbers] |
| datacopy = data[:] |
| |
| keyfn = self._CallCount(lambda value: str(int(value, 16))) |
| |
| # Sort with key function converting hex to decimal |
| result = algo.NiceSort(data, key=keyfn) |
| |
| self.assertEqual([hex(i) for i in sorted(numbers)], result) |
| self.assertEqual(data, datacopy, msg="Input data was modified in NiceSort") |
| self.assertEqual(keyfn.count, len(numbers), |
| msg="Key function was not called once per value") |
| |
| class _TestData: |
| def __init__(self, name, value): |
| self.name = name |
| self.value = value |
| |
| def testKeyfuncB(self): |
| rnd = random.Random(27396) |
| data = [] |
| for i in range(123): |
| v1 = rnd.randint(0, 5) |
| v2 = rnd.randint(0, 5) |
| data.append(self._TestData("inst-%s-%s-%s" % (v1, v2, i), |
| (v1, v2, i))) |
| rnd.shuffle(data) |
| assert data != sorted(data, key=operator.attrgetter("name")) |
| |
| keyfn = self._CallCount(operator.attrgetter("name")) |
| |
| # Sort by name |
| result = algo.NiceSort(data, key=keyfn) |
| |
| self.assertEqual(result, sorted(data, key=operator.attrgetter("value"))) |
| self.assertEqual(keyfn.count, len(data), |
| msg="Key function was not called once per value") |
| |
| def testNiceSortKey(self): |
| self.assertEqual(algo.NiceSortKey(""), |
| ([None] * algo._SORTER_GROUPS) + [""]) |
| self.assertEqual(algo.NiceSortKey("Hello World"), |
| ["Hello World"] + |
| ([None] * int(algo._SORTER_GROUPS - 1)) + [""]) |
| self.assertEqual(algo.NiceSortKey("node1.net75.bld3.example.com"), |
| ["node", 1, ".net", 75, ".bld", 3, ".example.com", |
| None, ""]) |
| |
| |
| class TestInvertDict(unittest.TestCase): |
| def testInvertDict(self): |
| test_dict = { "foo": 1, "bar": 2, "baz": 5 } |
| self.assertEqual(algo.InvertDict(test_dict), |
| { 1: "foo", 2: "bar", 5: "baz"}) |
| |
| |
| class TestInsertAtPos(unittest.TestCase): |
| def test(self): |
| a = [1, 5, 6] |
| b = [2, 3, 4] |
| self.assertEqual(algo.InsertAtPos(a, 1, b), [1, 2, 3, 4, 5, 6]) |
| self.assertEqual(algo.InsertAtPos(a, 0, b), b + a) |
| self.assertEqual(algo.InsertAtPos(a, len(a), b), a + b) |
| self.assertEqual(algo.InsertAtPos(a, 2, b), [1, 5, 2, 3, 4, 6]) |
| |
| |
| class TimeMock: |
| def __init__(self, values): |
| self.values = values |
| |
| def __call__(self): |
| return self.values.pop(0) |
| |
| |
| class TestRunningTimeout(unittest.TestCase): |
| def setUp(self): |
| self.time_fn = TimeMock([0.0, 0.3, 4.6, 6.5]) |
| |
| def testRemainingFloat(self): |
| timeout = algo.RunningTimeout(5.0, True, _time_fn=self.time_fn) |
| self.assertAlmostEqual(timeout.Remaining(), 4.7) |
| self.assertAlmostEqual(timeout.Remaining(), 0.4) |
| self.assertAlmostEqual(timeout.Remaining(), -1.5) |
| |
| def testRemaining(self): |
| self.time_fn = TimeMock([0, 2, 4, 5, 6]) |
| timeout = algo.RunningTimeout(5, True, _time_fn=self.time_fn) |
| self.assertEqual(timeout.Remaining(), 3) |
| self.assertEqual(timeout.Remaining(), 1) |
| self.assertEqual(timeout.Remaining(), 0) |
| self.assertEqual(timeout.Remaining(), -1) |
| |
| def testRemainingNonNegative(self): |
| timeout = algo.RunningTimeout(5.0, False, _time_fn=self.time_fn) |
| self.assertAlmostEqual(timeout.Remaining(), 4.7) |
| self.assertAlmostEqual(timeout.Remaining(), 0.4) |
| self.assertEqual(timeout.Remaining(), 0.0) |
| |
| def testNegativeTimeout(self): |
| self.assertRaises(ValueError, algo.RunningTimeout, -1.0, True) |
| |
| |
| class TestJoinDisjointDicts(unittest.TestCase): |
| def setUp(self): |
| self.non_empty_dict = {"a": 1, "b": 2} |
| self.empty_dict = {} |
| |
| def testWithEmptyDicts(self): |
| self.assertEqual(self.empty_dict, algo.JoinDisjointDicts(self.empty_dict, |
| self.empty_dict)) |
| self.assertEqual(self.non_empty_dict, algo.JoinDisjointDicts( |
| self.empty_dict, self.non_empty_dict)) |
| self.assertEqual(self.non_empty_dict, algo.JoinDisjointDicts( |
| self.non_empty_dict, self.empty_dict)) |
| |
| def testNonDisjoint(self): |
| self.assertRaises(AssertionError, algo.JoinDisjointDicts, |
| self.non_empty_dict, self.non_empty_dict) |
| |
| def testCommonCase(self): |
| dict_a = {"TEST1": 1, "TEST2": 2} |
| dict_b = {"TEST3": 3, "TEST4": 4} |
| |
| result = dict_a.copy() |
| result.update(dict_b) |
| |
| self.assertEqual(result, algo.JoinDisjointDicts(dict_a, dict_b)) |
| self.assertEqual(result, algo.JoinDisjointDicts(dict_b, dict_a)) |
| |
| |
| class TestSequenceToDict(unittest.TestCase): |
| def testEmpty(self): |
| self.assertEqual(algo.SequenceToDict([]), {}) |
| self.assertEqual(algo.SequenceToDict({}), {}) |
| |
| def testSimple(self): |
| data = [(i, str(i), "test%s" % i) for i in range(391)] |
| self.assertEqual(algo.SequenceToDict(data), |
| dict((i, (i, str(i), "test%s" % i)) |
| for i in range(391))) |
| |
| def testCustomKey(self): |
| data = [(i, hex(i), "test%s" % i) for i in range(100)] |
| self.assertEqual(algo.SequenceToDict(data, key=compat.snd), |
| dict((hex(i), (i, hex(i), "test%s" % i)) |
| for i in range(100))) |
| self.assertEqual(algo.SequenceToDict(data, |
| key=lambda (a, b, val): hash(val)), |
| dict((hash("test%s" % i), (i, hex(i), "test%s" % i)) |
| for i in range(100))) |
| |
| def testDuplicate(self): |
| self.assertRaises(ValueError, algo.SequenceToDict, |
| [(0, 0), (0, 0)]) |
| self.assertRaises(ValueError, algo.SequenceToDict, |
| [(i, ) for i in range(200)] + [(10, )]) |
| |
| |
| class TestFlatToDict(unittest.TestCase): |
| def testNormal(self): |
| data = [ |
| ("lv/xenvg", {"foo": "bar", "bar": "baz"}), |
| ("lv/xenfoo", {"foo": "bar", "baz": "blubb"}), |
| ("san/foo", {"ip": "127.0.0.1", "port": 1337}), |
| ("san/blubb/blibb", 54), |
| ] |
| reference = { |
| "lv": { |
| "xenvg": {"foo": "bar", "bar": "baz"}, |
| "xenfoo": {"foo": "bar", "baz": "blubb"}, |
| }, |
| "san": { |
| "foo": {"ip": "127.0.0.1", "port": 1337}, |
| "blubb": {"blibb": 54}, |
| }, |
| } |
| self.assertEqual(algo.FlatToDict(data), reference) |
| |
| def testUnlikeDepth(self): |
| data = [ |
| ("san/foo", {"ip": "127.0.0.1", "port": 1337}), |
| ("san/foo/blubb", 23), # Another foo entry under san |
| ("san/blubb/blibb", 54), |
| ] |
| self.assertRaises(AssertionError, algo.FlatToDict, data) |
| |
| |
| if __name__ == "__main__": |
| testutils.GanetiTestProgram() |
