Statistics
| Branch: | Tag: | Revision:

root / test / ganeti.utils.text_unittest.py @ bbfed756

History | View | Annotate | Download (17.6 kB)

1
#!/usr/bin/python
2
#
3

    
4
# Copyright (C) 2011 Google Inc.
5
#
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.
10
#
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.
15
#
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
19
# 02110-1301, USA.
20

    
21

    
22
"""Script for testing ganeti.utils.text"""
23

    
24
import re
25
import string
26
import time
27
import unittest
28
import os
29

    
30
from cStringIO import StringIO
31

    
32
from ganeti import constants
33
from ganeti import utils
34
from ganeti import errors
35

    
36
import testutils
37

    
38

    
39
class TestMatchNameComponent(unittest.TestCase):
40
  """Test case for the MatchNameComponent function"""
41

    
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)
46

    
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])
52

    
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)
58

    
59
  def testFullMatch(self):
60
    """Test that a full match is returned correctly"""
61
    key1 = "test1"
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)
66

    
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),
72
                     "test2.example.net")
73
    self.assertEqual(utils.MatchNameComponent("Test2", mlist,
74
                                              case_sensitive=False),
75
                     "test2.example.net")
76
    self.assertEqual(utils.MatchNameComponent("teSt2", mlist,
77
                                              case_sensitive=False),
78
                     "test2.example.net")
79
    self.assertEqual(utils.MatchNameComponent("TeSt2", mlist,
80
                                              case_sensitive=False),
81
                     "test2.example.net")
82

    
83
  def testCaseInsensitiveFullMatch(self):
84
    mlist = ["ts1.ex", "ts1.ex.org", "ts2.ex", "Ts2.ex"]
85

    
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),
89
                     None)
90
    self.assertEqual(utils.MatchNameComponent("Ts1.ex", mlist,
91
                                              case_sensitive=False),
92
                     "ts1.ex")
93
    self.assertEqual(utils.MatchNameComponent("ts1.ex", mlist,
94
                                              case_sensitive=False),
95
                     "ts1.ex")
96

    
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),
100
                     "ts2.ex")
101
    self.assertEqual(utils.MatchNameComponent("Ts2.ex", mlist,
102
                                              case_sensitive=False),
103
                     "Ts2.ex")
104
    self.assertEqual(utils.MatchNameComponent("TS2.ex", mlist,
105
                                              case_sensitive=False),
106
                     None)
107

    
108

    
109
class TestDnsNameGlobPattern(unittest.TestCase):
110
  def setUp(self):
111
    self.names = [
112
      "node1.example.com",
113
      "node2-0.example.com",
114
      "node2-1.example.com",
115
      "node1.example.net",
116
      "web1.example.com",
117
      "web2.example.com",
118
      "sub.site.example.com",
119
      ]
120

    
121
  def _Test(self, pattern):
122
    re_pat = utils.DnsNameGlobPattern(pattern)
123

    
124
    return filter(re.compile(re_pat).match, self.names)
125

    
126
  def test(self):
127
    for pattern in ["xyz", "node", " ", "example.net", "x*.example.*",
128
                    "x*.example.com"]:
129
      self.assertEqual(self._Test(pattern), [])
130

    
131
    for pattern in ["*", "???*"]:
132
      self.assertEqual(self._Test(pattern), self.names)
133

    
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"])
137

    
138
    for pattern in ["*.*.*.*", "???", "*.site"]:
139
      self.assertEqual(self._Test(pattern), ["sub.site.example.com"])
140

    
141
    self.assertEqual(self._Test("node1"), [
142
      "node1.example.com",
143
      "node1.example.net",
144
      ])
145
    self.assertEqual(self._Test("node?*.example.*"), [
146
      "node1.example.com",
147
      "node2-0.example.com",
148
      "node2-1.example.com",
149
      "node1.example.net",
150
      ])
151
    self.assertEqual(self._Test("*-?"), [
152
      "node2-0.example.com",
153
      "node2-1.example.com",
154
      ])
155
    self.assertEqual(self._Test("node2-?.example.com"), [
156
      "node2-0.example.com",
157
      "node2-1.example.com",
158
      ])
159

    
160

    
161
class TestFormatUnit(unittest.TestCase):
162
  """Test case for the FormatUnit function"""
163

    
164
  def testMiB(self):
165
    self.assertEqual(utils.FormatUnit(1, "h"), "1M")
166
    self.assertEqual(utils.FormatUnit(100, "h"), "100M")
167
    self.assertEqual(utils.FormatUnit(1023, "h"), "1023M")
168

    
169
    self.assertEqual(utils.FormatUnit(1, "m"), "1")
170
    self.assertEqual(utils.FormatUnit(100, "m"), "100")
171
    self.assertEqual(utils.FormatUnit(1023, "m"), "1023")
172

    
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")
177

    
178
  def testGiB(self):
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")
183

    
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")
188

    
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")
192

    
193
  def testTiB(self):
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")
197

    
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")
201

    
202
  def testErrors(self):
203
    self.assertRaises(errors.ProgrammerError, utils.FormatUnit, 1, "a")
204

    
205

    
206
class TestParseUnit(unittest.TestCase):
207
  """Test case for the ParseUnit function"""
208

    
209
  SCALES = (("", 1),
210
            ("M", 1), ("G", 1024), ("T", 1024 * 1024),
211
            ("MB", 1), ("GB", 1024), ("TB", 1024 * 1024),
212
            ("MiB", 1), ("GiB", 1024), ("TiB", 1024 * 1024))
213

    
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)
219

    
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)
227

    
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)
241

    
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)),
247
                           1024 * scale)
248

    
249
  def testInvalidInput(self):
250
    for sep in ("-", "_", ",", "a"):
251
      for suffix, _ in self.SCALES:
252
        self.assertRaises(errors.UnitParseError, utils.ParseUnit,
253
                          "1" + sep + suffix)
254

    
255
    for suffix, _ in self.SCALES:
256
      self.assertRaises(errors.UnitParseError, utils.ParseUnit,
257
                        "1,3" + suffix)
258

    
259

    
260
class TestShellQuoting(unittest.TestCase):
261
  """Test case for shell quoting functions"""
262

    
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'")
269

    
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")
274

    
275

    
276
class TestShellWriter(unittest.TestCase):
277
  def test(self):
278
    buf = StringIO()
279
    sw = utils.ShellWriter(buf)
280
    sw.Write("#!/bin/bash")
281
    sw.Write("if true; then")
282
    sw.IncIndent()
283
    try:
284
      sw.Write("echo true")
285

    
286
      sw.Write("for i in 1 2 3")
287
      sw.Write("do")
288
      sw.IncIndent()
289
      try:
290
        self.assertEqual(sw._indent, 2)
291
        sw.Write("date")
292
      finally:
293
        sw.DecIndent()
294
      sw.Write("done")
295
    finally:
296
      sw.DecIndent()
297
    sw.Write("echo %s", utils.ShellQuote("Hello World"))
298
    sw.Write("exit 0")
299

    
300
    self.assertEqual(sw._indent, 0)
301

    
302
    output = buf.getvalue()
303

    
304
    self.assert_(output.endswith("\n"))
305

    
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'")
311

    
312
  def testEmpty(self):
313
    buf = StringIO()
314
    sw = utils.ShellWriter(buf)
315
    sw = None
316
    self.assertEqual(buf.getvalue(), "")
317

    
318

    
319
class TestNormalizeAndValidateMac(unittest.TestCase):
320
  def testInvalid(self):
321
    self.assertRaises(errors.OpPrereqError,
322
                      utils.NormalizeAndValidateMac, "xxx")
323

    
324
  def testNormalization(self):
325
    for mac in ["aa:bb:cc:dd:ee:ff", "00:AA:11:bB:22:cc"]:
326
      self.assertEqual(utils.NormalizeAndValidateMac(mac), mac.lower())
327

    
328

    
329
class TestSafeEncode(unittest.TestCase):
330
  """Test case for SafeEncode"""
331

    
332
  def testAscii(self):
333
    for txt in [string.digits, string.letters, string.punctuation]:
334
      self.failUnlessEqual(txt, utils.SafeEncode(txt))
335

    
336
  def testDoubleEncode(self):
337
    for i in range(255):
338
      txt = utils.SafeEncode(chr(i))
339
      self.failUnlessEqual(txt, utils.SafeEncode(txt))
340

    
341
  def testUnicode(self):
342
    # 1024 is high enough to catch non-direct ASCII mappings
343
    for i in range(1024):
344
      txt = utils.SafeEncode(unichr(i))
345
      self.failUnlessEqual(txt, utils.SafeEncode(txt))
346

    
347

    
348
class TestUnescapeAndSplit(unittest.TestCase):
349
  """Testing case for UnescapeAndSplit"""
350

    
351
  def setUp(self):
352
    # testing more that one separator for regexp safety
353
    self._seps = [",", "+", "."]
354

    
355
  def testSimple(self):
356
    a = ["a", "b", "c", "d"]
357
    for sep in self._seps:
358
      self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), a)
359

    
360
  def testEscape(self):
361
    for sep in self._seps:
362
      a = ["a", "b\\" + sep + "c", "d"]
363
      b = ["a", "b" + sep + "c", "d"]
364
      self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
365

    
366
  def testDoubleEscape(self):
367
    for sep in self._seps:
368
      a = ["a", "b\\\\", "c", "d"]
369
      b = ["a", "b\\", "c", "d"]
370
      self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
371

    
372
  def testThreeEscape(self):
373
    for sep in self._seps:
374
      a = ["a", "b\\\\\\" + sep + "c", "d"]
375
      b = ["a", "b\\" + sep + "c", "d"]
376
      self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
377

    
378

    
379
class TestCommaJoin(unittest.TestCase):
380
  def test(self):
381
    self.assertEqual(utils.CommaJoin([]), "")
382
    self.assertEqual(utils.CommaJoin([1, 2, 3]), "1, 2, 3")
383
    self.assertEqual(utils.CommaJoin(["Hello"]), "Hello")
384
    self.assertEqual(utils.CommaJoin(["Hello", "World"]), "Hello, World")
385
    self.assertEqual(utils.CommaJoin(["Hello", "World", 99]),
386
                     "Hello, World, 99")
387

    
388

    
389
class TestFormatTime(unittest.TestCase):
390
  """Testing case for FormatTime"""
391

    
392
  @staticmethod
393
  def _TestInProcess(tz, timestamp, expected):
394
    os.environ["TZ"] = tz
395
    time.tzset()
396
    return utils.FormatTime(timestamp) == expected
397

    
398
  def _Test(self, *args):
399
    # Need to use separate process as we want to change TZ
400
    self.assert_(utils.RunInSeparateProcess(self._TestInProcess, *args))
401

    
402
  def test(self):
403
    self._Test("UTC", 0, "1970-01-01 00:00:00")
404
    self._Test("America/Sao_Paulo", 1292606926, "2010-12-17 15:28:46")
405
    self._Test("Europe/London", 1292606926, "2010-12-17 17:28:46")
406
    self._Test("Europe/Zurich", 1292606926, "2010-12-17 18:28:46")
407
    self._Test("Australia/Sydney", 1292606926, "2010-12-18 04:28:46")
408

    
409
  def testNone(self):
410
    self.failUnlessEqual(utils.FormatTime(None), "N/A")
411

    
412
  def testInvalid(self):
413
    self.failUnlessEqual(utils.FormatTime(()), "N/A")
414

    
415
  def testNow(self):
416
    # tests that we accept time.time input
417
    utils.FormatTime(time.time())
418
    # tests that we accept int input
419
    utils.FormatTime(int(time.time()))
420

    
421

    
422
class TestFormatSeconds(unittest.TestCase):
423
  def test(self):
424
    self.assertEqual(utils.FormatSeconds(1), "1s")
425
    self.assertEqual(utils.FormatSeconds(3600), "1h 0m 0s")
426
    self.assertEqual(utils.FormatSeconds(3599), "59m 59s")
427
    self.assertEqual(utils.FormatSeconds(7200), "2h 0m 0s")
428
    self.assertEqual(utils.FormatSeconds(7201), "2h 0m 1s")
429
    self.assertEqual(utils.FormatSeconds(7281), "2h 1m 21s")
430
    self.assertEqual(utils.FormatSeconds(29119), "8h 5m 19s")
431
    self.assertEqual(utils.FormatSeconds(19431228), "224d 21h 33m 48s")
432
    self.assertEqual(utils.FormatSeconds(-1), "-1s")
433
    self.assertEqual(utils.FormatSeconds(-282), "-282s")
434
    self.assertEqual(utils.FormatSeconds(-29119), "-29119s")
435

    
436
  def testFloat(self):
437
    self.assertEqual(utils.FormatSeconds(1.3), "1s")
438
    self.assertEqual(utils.FormatSeconds(1.9), "2s")
439
    self.assertEqual(utils.FormatSeconds(3912.12311), "1h 5m 12s")
440
    self.assertEqual(utils.FormatSeconds(3912.8), "1h 5m 13s")
441

    
442

    
443
class TestLineSplitter(unittest.TestCase):
444
  def test(self):
445
    lines = []
446
    ls = utils.LineSplitter(lines.append)
447
    ls.write("Hello World\n")
448
    self.assertEqual(lines, [])
449
    ls.write("Foo\n Bar\r\n ")
450
    ls.write("Baz")
451
    ls.write("Moo")
452
    self.assertEqual(lines, [])
453
    ls.flush()
454
    self.assertEqual(lines, ["Hello World", "Foo", " Bar"])
455
    ls.close()
456
    self.assertEqual(lines, ["Hello World", "Foo", " Bar", " BazMoo"])
457

    
458
  def _testExtra(self, line, all_lines, p1, p2):
459
    self.assertEqual(p1, 999)
460
    self.assertEqual(p2, "extra")
461
    all_lines.append(line)
462

    
463
  def testExtraArgsNoFlush(self):
464
    lines = []
465
    ls = utils.LineSplitter(self._testExtra, lines, 999, "extra")
466
    ls.write("\n\nHello World\n")
467
    ls.write("Foo\n Bar\r\n ")
468
    ls.write("")
469
    ls.write("Baz")
470
    ls.write("Moo\n\nx\n")
471
    self.assertEqual(lines, [])
472
    ls.close()
473
    self.assertEqual(lines, ["", "", "Hello World", "Foo", " Bar", " BazMoo",
474
                             "", "x"])
475

    
476

    
477
class TestIsValidShellParam(unittest.TestCase):
478
  def test(self):
479
    for val, result in [
480
      ("abc", True),
481
      ("ab;cd", False),
482
      ]:
483
      self.assertEqual(utils.IsValidShellParam(val), result)
484

    
485

    
486
class TestBuildShellCmd(unittest.TestCase):
487
  def test(self):
488
    self.assertRaises(errors.ProgrammerError, utils.BuildShellCmd,
489
                      "ls %s", "ab;cd")
490
    self.assertEqual(utils.BuildShellCmd("ls %s", "ab"), "ls ab")
491

    
492

    
493
class TestOrdinal(unittest.TestCase):
494
  def test(self):
495
    checks = {
496
      0: "0th", 1: "1st", 2: "2nd", 3: "3rd", 4: "4th", 5: "5th", 6: "6th",
497
      7: "7th", 8: "8th", 9: "9th", 10: "10th", 11: "11th", 12: "12th",
498
      13: "13th", 14: "14th", 15: "15th", 16: "16th", 17: "17th",
499
      18: "18th", 19: "19th", 20: "20th", 21: "21st", 25: "25th", 30: "30th",
500
      32: "32nd", 40: "40th", 50: "50th", 55: "55th", 60: "60th", 62: "62nd",
501
      70: "70th", 80: "80th", 83: "83rd", 90: "90th", 91: "91st",
502
      582: "582nd", 999: "999th",
503
      }
504

    
505
    for value, ordinal in checks.items():
506
      self.assertEqual(utils.FormatOrdinal(value), ordinal)
507

    
508

    
509
if __name__ == "__main__":
510
  testutils.GanetiTestProgram()