Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (18.3 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
  def testEscapeAtEnd(self):
379
    for sep in self._seps:
380
      self.assertEqual(utils.UnescapeAndSplit("\\", sep=sep), ["\\"])
381

    
382
      a = ["a", "b\\", "c"]
383
      b = ["a", "b" + sep + "c\\"]
384
      self.assertEqual(utils.UnescapeAndSplit("%s\\" % sep.join(a), sep=sep), b)
385

    
386
      a = ["\\" + sep, "\\" + sep, "c", "d\\.moo"]
387
      b = [sep, sep, "c", "d.moo\\"]
388
      self.assertEqual(utils.UnescapeAndSplit("%s\\" % sep.join(a), sep=sep), b)
389

    
390
  def testMultipleEscapes(self):
391
    for sep in self._seps:
392
      a = ["a", "b\\" + sep + "c", "d\\" + sep + "e\\" + sep + "f", "g"]
393
      b = ["a", "b" + sep + "c", "d" + sep + "e" + sep + "f", "g"]
394
      self.failUnlessEqual(utils.UnescapeAndSplit(sep.join(a), sep=sep), b)
395

    
396

    
397
class TestCommaJoin(unittest.TestCase):
398
  def test(self):
399
    self.assertEqual(utils.CommaJoin([]), "")
400
    self.assertEqual(utils.CommaJoin([1, 2, 3]), "1, 2, 3")
401
    self.assertEqual(utils.CommaJoin(["Hello"]), "Hello")
402
    self.assertEqual(utils.CommaJoin(["Hello", "World"]), "Hello, World")
403
    self.assertEqual(utils.CommaJoin(["Hello", "World", 99]),
404
                     "Hello, World, 99")
405

    
406

    
407
class TestFormatTime(unittest.TestCase):
408
  """Testing case for FormatTime"""
409

    
410
  @staticmethod
411
  def _TestInProcess(tz, timestamp, expected):
412
    os.environ["TZ"] = tz
413
    time.tzset()
414
    return utils.FormatTime(timestamp) == expected
415

    
416
  def _Test(self, *args):
417
    # Need to use separate process as we want to change TZ
418
    self.assert_(utils.RunInSeparateProcess(self._TestInProcess, *args))
419

    
420
  def test(self):
421
    self._Test("UTC", 0, "1970-01-01 00:00:00")
422
    self._Test("America/Sao_Paulo", 1292606926, "2010-12-17 15:28:46")
423
    self._Test("Europe/London", 1292606926, "2010-12-17 17:28:46")
424
    self._Test("Europe/Zurich", 1292606926, "2010-12-17 18:28:46")
425
    self._Test("Australia/Sydney", 1292606926, "2010-12-18 04:28:46")
426

    
427
  def testNone(self):
428
    self.failUnlessEqual(utils.FormatTime(None), "N/A")
429

    
430
  def testInvalid(self):
431
    self.failUnlessEqual(utils.FormatTime(()), "N/A")
432

    
433
  def testNow(self):
434
    # tests that we accept time.time input
435
    utils.FormatTime(time.time())
436
    # tests that we accept int input
437
    utils.FormatTime(int(time.time()))
438

    
439

    
440
class TestFormatSeconds(unittest.TestCase):
441
  def test(self):
442
    self.assertEqual(utils.FormatSeconds(1), "1s")
443
    self.assertEqual(utils.FormatSeconds(3600), "1h 0m 0s")
444
    self.assertEqual(utils.FormatSeconds(3599), "59m 59s")
445
    self.assertEqual(utils.FormatSeconds(7200), "2h 0m 0s")
446
    self.assertEqual(utils.FormatSeconds(7201), "2h 0m 1s")
447
    self.assertEqual(utils.FormatSeconds(7281), "2h 1m 21s")
448
    self.assertEqual(utils.FormatSeconds(29119), "8h 5m 19s")
449
    self.assertEqual(utils.FormatSeconds(19431228), "224d 21h 33m 48s")
450
    self.assertEqual(utils.FormatSeconds(-1), "-1s")
451
    self.assertEqual(utils.FormatSeconds(-282), "-282s")
452
    self.assertEqual(utils.FormatSeconds(-29119), "-29119s")
453

    
454
  def testFloat(self):
455
    self.assertEqual(utils.FormatSeconds(1.3), "1s")
456
    self.assertEqual(utils.FormatSeconds(1.9), "2s")
457
    self.assertEqual(utils.FormatSeconds(3912.12311), "1h 5m 12s")
458
    self.assertEqual(utils.FormatSeconds(3912.8), "1h 5m 13s")
459

    
460

    
461
class TestLineSplitter(unittest.TestCase):
462
  def test(self):
463
    lines = []
464
    ls = utils.LineSplitter(lines.append)
465
    ls.write("Hello World\n")
466
    self.assertEqual(lines, [])
467
    ls.write("Foo\n Bar\r\n ")
468
    ls.write("Baz")
469
    ls.write("Moo")
470
    self.assertEqual(lines, [])
471
    ls.flush()
472
    self.assertEqual(lines, ["Hello World", "Foo", " Bar"])
473
    ls.close()
474
    self.assertEqual(lines, ["Hello World", "Foo", " Bar", " BazMoo"])
475

    
476
  def _testExtra(self, line, all_lines, p1, p2):
477
    self.assertEqual(p1, 999)
478
    self.assertEqual(p2, "extra")
479
    all_lines.append(line)
480

    
481
  def testExtraArgsNoFlush(self):
482
    lines = []
483
    ls = utils.LineSplitter(self._testExtra, lines, 999, "extra")
484
    ls.write("\n\nHello World\n")
485
    ls.write("Foo\n Bar\r\n ")
486
    ls.write("")
487
    ls.write("Baz")
488
    ls.write("Moo\n\nx\n")
489
    self.assertEqual(lines, [])
490
    ls.close()
491
    self.assertEqual(lines, ["", "", "Hello World", "Foo", " Bar", " BazMoo",
492
                             "", "x"])
493

    
494

    
495
class TestIsValidShellParam(unittest.TestCase):
496
  def test(self):
497
    for val, result in [
498
      ("abc", True),
499
      ("ab;cd", False),
500
      ]:
501
      self.assertEqual(utils.IsValidShellParam(val), result)
502

    
503

    
504
class TestBuildShellCmd(unittest.TestCase):
505
  def test(self):
506
    self.assertRaises(errors.ProgrammerError, utils.BuildShellCmd,
507
                      "ls %s", "ab;cd")
508
    self.assertEqual(utils.BuildShellCmd("ls %s", "ab"), "ls ab")
509

    
510

    
511
class TestOrdinal(unittest.TestCase):
512
  def test(self):
513
    checks = {
514
      0: "0th", 1: "1st", 2: "2nd", 3: "3rd", 4: "4th", 5: "5th", 6: "6th",
515
      7: "7th", 8: "8th", 9: "9th", 10: "10th", 11: "11th", 12: "12th",
516
      13: "13th", 14: "14th", 15: "15th", 16: "16th", 17: "17th",
517
      18: "18th", 19: "19th", 20: "20th", 21: "21st", 25: "25th", 30: "30th",
518
      32: "32nd", 40: "40th", 50: "50th", 55: "55th", 60: "60th", 62: "62nd",
519
      70: "70th", 80: "80th", 83: "83rd", 90: "90th", 91: "91st",
520
      582: "582nd", 999: "999th",
521
      }
522

    
523
    for value, ordinal in checks.items():
524
      self.assertEqual(utils.FormatOrdinal(value), ordinal)
525

    
526

    
527
if __name__ == "__main__":
528
  testutils.GanetiTestProgram()