4 # Copyright (C) 2006, 2007 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 unittesting the utils module"""
40 from ganeti import constants
41 from ganeti import utils
42 from ganeti import errors
43 from ganeti.utils import IsProcessAlive, RunCmd, \
44 RemoveFile, MatchNameComponent, FormatUnit, \
45 ParseUnit, AddAuthorizedKey, RemoveAuthorizedKey, \
46 ShellQuote, ShellQuoteArgs, TcpPing, ListVisibleFiles, \
47 SetEtcHostsEntry, RemoveEtcHostsEntry, FirstFree, OwnIpAddress, \
48 TailFile, ForceDictType, SafeEncode, IsNormAbsPath
50 from ganeti.errors import LockError, UnitParseError, GenericError, \
54 class TestIsProcessAlive(unittest.TestCase):
55 """Testing case for IsProcessAlive"""
59 self.assert_(IsProcessAlive(mypid),
60 "can't find myself running")
62 def testNotExisting(self):
63 pid_non_existing = os.fork()
64 if pid_non_existing == 0:
66 elif pid_non_existing < 0:
67 raise SystemError("can't fork")
68 os.waitpid(pid_non_existing, 0)
69 self.assert_(not IsProcessAlive(pid_non_existing),
70 "nonexisting process detected")
73 class TestPidFileFunctions(unittest.TestCase):
74 """Tests for WritePidFile, RemovePidFile and ReadPidFile"""
77 self.dir = tempfile.mkdtemp()
78 self.f_dpn = lambda name: os.path.join(self.dir, "%s.pid" % name)
79 utils.DaemonPidFileName = self.f_dpn
81 def testPidFileFunctions(self):
82 pid_file = self.f_dpn('test')
83 utils.WritePidFile('test')
84 self.failUnless(os.path.exists(pid_file),
85 "PID file should have been created")
86 read_pid = utils.ReadPidFile(pid_file)
87 self.failUnlessEqual(read_pid, os.getpid())
88 self.failUnless(utils.IsProcessAlive(read_pid))
89 self.failUnlessRaises(GenericError, utils.WritePidFile, 'test')
90 utils.RemovePidFile('test')
91 self.failIf(os.path.exists(pid_file),
92 "PID file should not exist anymore")
93 self.failUnlessEqual(utils.ReadPidFile(pid_file), 0,
94 "ReadPidFile should return 0 for missing pid file")
95 fh = open(pid_file, "w")
98 self.failUnlessEqual(utils.ReadPidFile(pid_file), 0,
99 "ReadPidFile should return 0 for invalid pid file")
100 utils.RemovePidFile('test')
101 self.failIf(os.path.exists(pid_file),
102 "PID file should not exist anymore")
105 pid_file = self.f_dpn('child')
106 r_fd, w_fd = os.pipe()
108 if new_pid == 0: #child
109 utils.WritePidFile('child')
114 # else we are in the parent
115 # wait until the child has written the pid file
117 read_pid = utils.ReadPidFile(pid_file)
118 self.failUnlessEqual(read_pid, new_pid)
119 self.failUnless(utils.IsProcessAlive(new_pid))
120 utils.KillProcess(new_pid, waitpid=True)
121 self.failIf(utils.IsProcessAlive(new_pid))
122 utils.RemovePidFile('child')
123 self.failUnlessRaises(ProgrammerError, utils.KillProcess, 0)
126 for name in os.listdir(self.dir):
127 os.unlink(os.path.join(self.dir, name))
131 class TestRunCmd(testutils.GanetiTestCase):
132 """Testing case for the RunCmd function"""
135 testutils.GanetiTestCase.setUp(self)
136 self.magic = time.ctime() + " ganeti test"
137 self.fname = self._CreateTempFile()
140 """Test successful exit code"""
141 result = RunCmd("/bin/sh -c 'exit 0'")
142 self.assertEqual(result.exit_code, 0)
143 self.assertEqual(result.output, "")
146 """Test fail exit code"""
147 result = RunCmd("/bin/sh -c 'exit 1'")
148 self.assertEqual(result.exit_code, 1)
149 self.assertEqual(result.output, "")
151 def testStdout(self):
152 """Test standard output"""
153 cmd = 'echo -n "%s"' % self.magic
154 result = RunCmd("/bin/sh -c '%s'" % cmd)
155 self.assertEqual(result.stdout, self.magic)
156 result = RunCmd("/bin/sh -c '%s'" % cmd, output=self.fname)
157 self.assertEqual(result.output, "")
158 self.assertFileContent(self.fname, self.magic)
160 def testStderr(self):
161 """Test standard error"""
162 cmd = 'echo -n "%s"' % self.magic
163 result = RunCmd("/bin/sh -c '%s' 1>&2" % cmd)
164 self.assertEqual(result.stderr, self.magic)
165 result = RunCmd("/bin/sh -c '%s' 1>&2" % cmd, output=self.fname)
166 self.assertEqual(result.output, "")
167 self.assertFileContent(self.fname, self.magic)
169 def testCombined(self):
170 """Test combined output"""
171 cmd = 'echo -n "A%s"; echo -n "B%s" 1>&2' % (self.magic, self.magic)
172 expected = "A" + self.magic + "B" + self.magic
173 result = RunCmd("/bin/sh -c '%s'" % cmd)
174 self.assertEqual(result.output, expected)
175 result = RunCmd("/bin/sh -c '%s'" % cmd, output=self.fname)
176 self.assertEqual(result.output, "")
177 self.assertFileContent(self.fname, expected)
179 def testSignal(self):
181 result = RunCmd(["python", "-c", "import os; os.kill(os.getpid(), 15)"])
182 self.assertEqual(result.signal, 15)
183 self.assertEqual(result.output, "")
185 def testListRun(self):
187 result = RunCmd(["true"])
188 self.assertEqual(result.signal, None)
189 self.assertEqual(result.exit_code, 0)
190 result = RunCmd(["/bin/sh", "-c", "exit 1"])
191 self.assertEqual(result.signal, None)
192 self.assertEqual(result.exit_code, 1)
193 result = RunCmd(["echo", "-n", self.magic])
194 self.assertEqual(result.signal, None)
195 self.assertEqual(result.exit_code, 0)
196 self.assertEqual(result.stdout, self.magic)
198 def testFileEmptyOutput(self):
199 """Test file output"""
200 result = RunCmd(["true"], output=self.fname)
201 self.assertEqual(result.signal, None)
202 self.assertEqual(result.exit_code, 0)
203 self.assertFileContent(self.fname, "")
206 """Test locale environment"""
207 old_env = os.environ.copy()
209 os.environ["LANG"] = "en_US.UTF-8"
210 os.environ["LC_ALL"] = "en_US.UTF-8"
211 result = RunCmd(["locale"])
212 for line in result.output.splitlines():
213 key, value = line.split("=", 1)
214 # Ignore these variables, they're overridden by LC_ALL
215 if key == "LANG" or key == "LANGUAGE":
217 self.failIf(value and value != "C" and value != '"C"',
218 "Variable %s is set to the invalid value '%s'" % (key, value))
222 def testDefaultCwd(self):
223 """Test default working directory"""
224 self.failUnlessEqual(RunCmd(["pwd"]).stdout.strip(), "/")
227 """Test default working directory"""
228 self.failUnlessEqual(RunCmd(["pwd"], cwd="/").stdout.strip(), "/")
229 self.failUnlessEqual(RunCmd(["pwd"], cwd="/tmp").stdout.strip(), "/tmp")
231 self.failUnlessEqual(RunCmd(["pwd"], cwd=cwd).stdout.strip(), cwd)
234 class TestRemoveFile(unittest.TestCase):
235 """Test case for the RemoveFile function"""
238 """Create a temp dir and file for each case"""
239 self.tmpdir = tempfile.mkdtemp('', 'ganeti-unittest-')
240 fd, self.tmpfile = tempfile.mkstemp('', '', self.tmpdir)
244 if os.path.exists(self.tmpfile):
245 os.unlink(self.tmpfile)
246 os.rmdir(self.tmpdir)
249 def testIgnoreDirs(self):
250 """Test that RemoveFile() ignores directories"""
251 self.assertEqual(None, RemoveFile(self.tmpdir))
254 def testIgnoreNotExisting(self):
255 """Test that RemoveFile() ignores non-existing files"""
256 RemoveFile(self.tmpfile)
257 RemoveFile(self.tmpfile)
260 def testRemoveFile(self):
261 """Test that RemoveFile does remove a file"""
262 RemoveFile(self.tmpfile)
263 if os.path.exists(self.tmpfile):
264 self.fail("File '%s' not removed" % self.tmpfile)
267 def testRemoveSymlink(self):
268 """Test that RemoveFile does remove symlinks"""
269 symlink = self.tmpdir + "/symlink"
270 os.symlink("no-such-file", symlink)
272 if os.path.exists(symlink):
273 self.fail("File '%s' not removed" % symlink)
274 os.symlink(self.tmpfile, symlink)
276 if os.path.exists(symlink):
277 self.fail("File '%s' not removed" % symlink)
280 class TestRename(unittest.TestCase):
281 """Test case for RenameFile"""
284 """Create a temporary directory"""
285 self.tmpdir = tempfile.mkdtemp()
286 self.tmpfile = os.path.join(self.tmpdir, "test1")
289 open(self.tmpfile, "w").close()
292 """Remove temporary directory"""
293 shutil.rmtree(self.tmpdir)
295 def testSimpleRename1(self):
296 """Simple rename 1"""
297 utils.RenameFile(self.tmpfile, os.path.join(self.tmpdir, "xyz"))
299 def testSimpleRename2(self):
300 """Simple rename 2"""
301 utils.RenameFile(self.tmpfile, os.path.join(self.tmpdir, "xyz"),
304 def testRenameMkdir(self):
305 """Rename with mkdir"""
306 utils.RenameFile(self.tmpfile, os.path.join(self.tmpdir, "test/xyz"),
310 class TestMatchNameComponent(unittest.TestCase):
311 """Test case for the MatchNameComponent function"""
313 def testEmptyList(self):
314 """Test that there is no match against an empty list"""
316 self.failUnlessEqual(MatchNameComponent("", []), None)
317 self.failUnlessEqual(MatchNameComponent("test", []), None)
319 def testSingleMatch(self):
320 """Test that a single match is performed correctly"""
321 mlist = ["test1.example.com", "test2.example.com", "test3.example.com"]
322 for key in "test2", "test2.example", "test2.example.com":
323 self.failUnlessEqual(MatchNameComponent(key, mlist), mlist[1])
325 def testMultipleMatches(self):
326 """Test that a multiple match is returned as None"""
327 mlist = ["test1.example.com", "test1.example.org", "test1.example.net"]
328 for key in "test1", "test1.example":
329 self.failUnlessEqual(MatchNameComponent(key, mlist), None)
332 class TestFormatUnit(unittest.TestCase):
333 """Test case for the FormatUnit function"""
336 self.assertEqual(FormatUnit(1, 'h'), '1M')
337 self.assertEqual(FormatUnit(100, 'h'), '100M')
338 self.assertEqual(FormatUnit(1023, 'h'), '1023M')
340 self.assertEqual(FormatUnit(1, 'm'), '1')
341 self.assertEqual(FormatUnit(100, 'm'), '100')
342 self.assertEqual(FormatUnit(1023, 'm'), '1023')
344 self.assertEqual(FormatUnit(1024, 'm'), '1024')
345 self.assertEqual(FormatUnit(1536, 'm'), '1536')
346 self.assertEqual(FormatUnit(17133, 'm'), '17133')
347 self.assertEqual(FormatUnit(1024 * 1024 - 1, 'm'), '1048575')
350 self.assertEqual(FormatUnit(1024, 'h'), '1.0G')
351 self.assertEqual(FormatUnit(1536, 'h'), '1.5G')
352 self.assertEqual(FormatUnit(17133, 'h'), '16.7G')
353 self.assertEqual(FormatUnit(1024 * 1024 - 1, 'h'), '1024.0G')
355 self.assertEqual(FormatUnit(1024, 'g'), '1.0')
356 self.assertEqual(FormatUnit(1536, 'g'), '1.5')
357 self.assertEqual(FormatUnit(17133, 'g'), '16.7')
358 self.assertEqual(FormatUnit(1024 * 1024 - 1, 'g'), '1024.0')
360 self.assertEqual(FormatUnit(1024 * 1024, 'g'), '1024.0')
361 self.assertEqual(FormatUnit(5120 * 1024, 'g'), '5120.0')
362 self.assertEqual(FormatUnit(29829 * 1024, 'g'), '29829.0')
365 self.assertEqual(FormatUnit(1024 * 1024, 'h'), '1.0T')
366 self.assertEqual(FormatUnit(5120 * 1024, 'h'), '5.0T')
367 self.assertEqual(FormatUnit(29829 * 1024, 'h'), '29.1T')
369 self.assertEqual(FormatUnit(1024 * 1024, 't'), '1.0')
370 self.assertEqual(FormatUnit(5120 * 1024, 't'), '5.0')
371 self.assertEqual(FormatUnit(29829 * 1024, 't'), '29.1')
373 class TestParseUnit(unittest.TestCase):
374 """Test case for the ParseUnit function"""
377 ('M', 1), ('G', 1024), ('T', 1024 * 1024),
378 ('MB', 1), ('GB', 1024), ('TB', 1024 * 1024),
379 ('MiB', 1), ('GiB', 1024), ('TiB', 1024 * 1024))
381 def testRounding(self):
382 self.assertEqual(ParseUnit('0'), 0)
383 self.assertEqual(ParseUnit('1'), 4)
384 self.assertEqual(ParseUnit('2'), 4)
385 self.assertEqual(ParseUnit('3'), 4)
387 self.assertEqual(ParseUnit('124'), 124)
388 self.assertEqual(ParseUnit('125'), 128)
389 self.assertEqual(ParseUnit('126'), 128)
390 self.assertEqual(ParseUnit('127'), 128)
391 self.assertEqual(ParseUnit('128'), 128)
392 self.assertEqual(ParseUnit('129'), 132)
393 self.assertEqual(ParseUnit('130'), 132)
395 def testFloating(self):
396 self.assertEqual(ParseUnit('0'), 0)
397 self.assertEqual(ParseUnit('0.5'), 4)
398 self.assertEqual(ParseUnit('1.75'), 4)
399 self.assertEqual(ParseUnit('1.99'), 4)
400 self.assertEqual(ParseUnit('2.00'), 4)
401 self.assertEqual(ParseUnit('2.01'), 4)
402 self.assertEqual(ParseUnit('3.99'), 4)
403 self.assertEqual(ParseUnit('4.00'), 4)
404 self.assertEqual(ParseUnit('4.01'), 8)
405 self.assertEqual(ParseUnit('1.5G'), 1536)
406 self.assertEqual(ParseUnit('1.8G'), 1844)
407 self.assertEqual(ParseUnit('8.28T'), 8682212)
409 def testSuffixes(self):
410 for sep in ('', ' ', ' ', "\t", "\t "):
411 for suffix, scale in TestParseUnit.SCALES:
412 for func in (lambda x: x, str.lower, str.upper):
413 self.assertEqual(ParseUnit('1024' + sep + func(suffix)),
416 def testInvalidInput(self):
417 for sep in ('-', '_', ',', 'a'):
418 for suffix, _ in TestParseUnit.SCALES:
419 self.assertRaises(UnitParseError, ParseUnit, '1' + sep + suffix)
421 for suffix, _ in TestParseUnit.SCALES:
422 self.assertRaises(UnitParseError, ParseUnit, '1,3' + suffix)
425 class TestSshKeys(testutils.GanetiTestCase):
426 """Test case for the AddAuthorizedKey function"""
428 KEY_A = 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a'
429 KEY_B = ('command="/usr/bin/fooserver -t --verbose",from="1.2.3.4" '
430 'ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b')
433 testutils.GanetiTestCase.setUp(self)
434 self.tmpname = self._CreateTempFile()
435 handle = open(self.tmpname, 'w')
437 handle.write("%s\n" % TestSshKeys.KEY_A)
438 handle.write("%s\n" % TestSshKeys.KEY_B)
442 def testAddingNewKey(self):
443 AddAuthorizedKey(self.tmpname, 'ssh-dss AAAAB3NzaC1kc3MAAACB root@test')
445 self.assertFileContent(self.tmpname,
446 "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a\n"
447 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
448 " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n"
449 "ssh-dss AAAAB3NzaC1kc3MAAACB root@test\n")
451 def testAddingAlmostButNotCompletelyTheSameKey(self):
452 AddAuthorizedKey(self.tmpname,
453 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@test')
455 self.assertFileContent(self.tmpname,
456 "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a\n"
457 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
458 " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n"
459 "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@test\n")
461 def testAddingExistingKeyWithSomeMoreSpaces(self):
462 AddAuthorizedKey(self.tmpname,
463 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a')
465 self.assertFileContent(self.tmpname,
466 "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a\n"
467 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
468 " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n")
470 def testRemovingExistingKeyWithSomeMoreSpaces(self):
471 RemoveAuthorizedKey(self.tmpname,
472 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a')
474 self.assertFileContent(self.tmpname,
475 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
476 " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n")
478 def testRemovingNonExistingKey(self):
479 RemoveAuthorizedKey(self.tmpname,
480 'ssh-dss AAAAB3Nsdfj230xxjxJjsjwjsjdjU root@test')
482 self.assertFileContent(self.tmpname,
483 "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a\n"
484 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
485 " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n")
488 class TestEtcHosts(testutils.GanetiTestCase):
489 """Test functions modifying /etc/hosts"""
492 testutils.GanetiTestCase.setUp(self)
493 self.tmpname = self._CreateTempFile()
494 handle = open(self.tmpname, 'w')
496 handle.write('# This is a test file for /etc/hosts\n')
497 handle.write('127.0.0.1\tlocalhost\n')
498 handle.write('192.168.1.1 router gw\n')
502 def testSettingNewIp(self):
503 SetEtcHostsEntry(self.tmpname, '1.2.3.4', 'myhost.domain.tld', ['myhost'])
505 self.assertFileContent(self.tmpname,
506 "# This is a test file for /etc/hosts\n"
507 "127.0.0.1\tlocalhost\n"
508 "192.168.1.1 router gw\n"
509 "1.2.3.4\tmyhost.domain.tld myhost\n")
510 self.assertFileMode(self.tmpname, 0644)
512 def testSettingExistingIp(self):
513 SetEtcHostsEntry(self.tmpname, '192.168.1.1', 'myhost.domain.tld',
516 self.assertFileContent(self.tmpname,
517 "# This is a test file for /etc/hosts\n"
518 "127.0.0.1\tlocalhost\n"
519 "192.168.1.1\tmyhost.domain.tld myhost\n")
520 self.assertFileMode(self.tmpname, 0644)
522 def testSettingDuplicateName(self):
523 SetEtcHostsEntry(self.tmpname, '1.2.3.4', 'myhost', ['myhost'])
525 self.assertFileContent(self.tmpname,
526 "# This is a test file for /etc/hosts\n"
527 "127.0.0.1\tlocalhost\n"
528 "192.168.1.1 router gw\n"
530 self.assertFileMode(self.tmpname, 0644)
532 def testRemovingExistingHost(self):
533 RemoveEtcHostsEntry(self.tmpname, 'router')
535 self.assertFileContent(self.tmpname,
536 "# This is a test file for /etc/hosts\n"
537 "127.0.0.1\tlocalhost\n"
539 self.assertFileMode(self.tmpname, 0644)
541 def testRemovingSingleExistingHost(self):
542 RemoveEtcHostsEntry(self.tmpname, 'localhost')
544 self.assertFileContent(self.tmpname,
545 "# This is a test file for /etc/hosts\n"
546 "192.168.1.1 router gw\n")
547 self.assertFileMode(self.tmpname, 0644)
549 def testRemovingNonExistingHost(self):
550 RemoveEtcHostsEntry(self.tmpname, 'myhost')
552 self.assertFileContent(self.tmpname,
553 "# This is a test file for /etc/hosts\n"
554 "127.0.0.1\tlocalhost\n"
555 "192.168.1.1 router gw\n")
556 self.assertFileMode(self.tmpname, 0644)
558 def testRemovingAlias(self):
559 RemoveEtcHostsEntry(self.tmpname, 'gw')
561 self.assertFileContent(self.tmpname,
562 "# This is a test file for /etc/hosts\n"
563 "127.0.0.1\tlocalhost\n"
564 "192.168.1.1 router\n")
565 self.assertFileMode(self.tmpname, 0644)
568 class TestShellQuoting(unittest.TestCase):
569 """Test case for shell quoting functions"""
571 def testShellQuote(self):
572 self.assertEqual(ShellQuote('abc'), "abc")
573 self.assertEqual(ShellQuote('ab"c'), "'ab\"c'")
574 self.assertEqual(ShellQuote("a'bc"), "'a'\\''bc'")
575 self.assertEqual(ShellQuote("a b c"), "'a b c'")
576 self.assertEqual(ShellQuote("a b\\ c"), "'a b\\ c'")
578 def testShellQuoteArgs(self):
579 self.assertEqual(ShellQuoteArgs(['a', 'b', 'c']), "a b c")
580 self.assertEqual(ShellQuoteArgs(['a', 'b"', 'c']), "a 'b\"' c")
581 self.assertEqual(ShellQuoteArgs(['a', 'b\'', 'c']), "a 'b'\\\''' c")
584 class TestTcpPing(unittest.TestCase):
585 """Testcase for TCP version of ping - against listen(2)ing port"""
588 self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
589 self.listener.bind((constants.LOCALHOST_IP_ADDRESS, 0))
590 self.listenerport = self.listener.getsockname()[1]
591 self.listener.listen(1)
594 self.listener.shutdown(socket.SHUT_RDWR)
596 del self.listenerport
598 def testTcpPingToLocalHostAccept(self):
599 self.assert_(TcpPing(constants.LOCALHOST_IP_ADDRESS,
602 live_port_needed=True,
603 source=constants.LOCALHOST_IP_ADDRESS,
605 "failed to connect to test listener")
607 self.assert_(TcpPing(constants.LOCALHOST_IP_ADDRESS,
610 live_port_needed=True,
612 "failed to connect to test listener (no source)")
615 class TestTcpPingDeaf(unittest.TestCase):
616 """Testcase for TCP version of ping - against non listen(2)ing port"""
619 self.deaflistener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
620 self.deaflistener.bind((constants.LOCALHOST_IP_ADDRESS, 0))
621 self.deaflistenerport = self.deaflistener.getsockname()[1]
624 del self.deaflistener
625 del self.deaflistenerport
627 def testTcpPingToLocalHostAcceptDeaf(self):
628 self.failIf(TcpPing(constants.LOCALHOST_IP_ADDRESS,
629 self.deaflistenerport,
630 timeout=constants.TCP_PING_TIMEOUT,
631 live_port_needed=True,
632 source=constants.LOCALHOST_IP_ADDRESS,
633 ), # need successful connect(2)
634 "successfully connected to deaf listener")
636 self.failIf(TcpPing(constants.LOCALHOST_IP_ADDRESS,
637 self.deaflistenerport,
638 timeout=constants.TCP_PING_TIMEOUT,
639 live_port_needed=True,
640 ), # need successful connect(2)
641 "successfully connected to deaf listener (no source addr)")
643 def testTcpPingToLocalHostNoAccept(self):
644 self.assert_(TcpPing(constants.LOCALHOST_IP_ADDRESS,
645 self.deaflistenerport,
646 timeout=constants.TCP_PING_TIMEOUT,
647 live_port_needed=False,
648 source=constants.LOCALHOST_IP_ADDRESS,
649 ), # ECONNREFUSED is OK
650 "failed to ping alive host on deaf port")
652 self.assert_(TcpPing(constants.LOCALHOST_IP_ADDRESS,
653 self.deaflistenerport,
654 timeout=constants.TCP_PING_TIMEOUT,
655 live_port_needed=False,
656 ), # ECONNREFUSED is OK
657 "failed to ping alive host on deaf port (no source addr)")
660 class TestOwnIpAddress(unittest.TestCase):
661 """Testcase for OwnIpAddress"""
663 def testOwnLoopback(self):
664 """check having the loopback ip"""
665 self.failUnless(OwnIpAddress(constants.LOCALHOST_IP_ADDRESS),
666 "Should own the loopback address")
668 def testNowOwnAddress(self):
669 """check that I don't own an address"""
671 # network 192.0.2.0/24 is reserved for test/documentation as per
672 # rfc 3330, so we *should* not have an address of this range... if
673 # this fails, we should extend the test to multiple addresses
675 self.failIf(OwnIpAddress(DST_IP), "Should not own IP address %s" % DST_IP)
678 class TestListVisibleFiles(unittest.TestCase):
679 """Test case for ListVisibleFiles"""
682 self.path = tempfile.mkdtemp()
685 shutil.rmtree(self.path)
687 def _test(self, files, expected):
689 expected = expected[:]
693 f = open(os.path.join(self.path, name), 'w')
699 found = ListVisibleFiles(self.path)
702 self.assertEqual(found, expected)
704 def testAllVisible(self):
705 files = ["a", "b", "c"]
707 self._test(files, expected)
709 def testNoneVisible(self):
710 files = [".a", ".b", ".c"]
712 self._test(files, expected)
714 def testSomeVisible(self):
715 files = ["a", "b", ".c"]
716 expected = ["a", "b"]
717 self._test(files, expected)
720 class TestNewUUID(unittest.TestCase):
721 """Test case for NewUUID"""
723 _re_uuid = re.compile('^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-'
724 '[a-f0-9]{4}-[a-f0-9]{12}$')
727 self.failUnless(self._re_uuid.match(utils.NewUUID()))
730 class TestUniqueSequence(unittest.TestCase):
731 """Test case for UniqueSequence"""
733 def _test(self, input, expected):
734 self.assertEqual(utils.UniqueSequence(input), expected)
738 self._test([1, 2, 3], [1, 2, 3])
739 self._test([1, 1, 2, 2, 3, 3], [1, 2, 3])
740 self._test([1, 2, 2, 3], [1, 2, 3])
741 self._test([1, 2, 3, 3], [1, 2, 3])
744 self._test([1, 2, 3, 1, 2, 3], [1, 2, 3])
745 self._test([1, 1, 2, 3, 3, 1, 2], [1, 2, 3])
748 self._test(["a", "a"], ["a"])
749 self._test(["a", "b"], ["a", "b"])
750 self._test(["a", "b", "a"], ["a", "b"])
753 class TestFirstFree(unittest.TestCase):
754 """Test case for the FirstFree function"""
758 self.failUnlessEqual(FirstFree([0, 1, 3]), 2)
759 self.failUnlessEqual(FirstFree([]), None)
760 self.failUnlessEqual(FirstFree([3, 4, 6]), 0)
761 self.failUnlessEqual(FirstFree([3, 4, 6], base=3), 5)
762 self.failUnlessRaises(AssertionError, FirstFree, [0, 3, 4, 6], base=3)
765 class TestTailFile(testutils.GanetiTestCase):
766 """Test case for the TailFile function"""
769 fname = self._CreateTempFile()
770 self.failUnlessEqual(TailFile(fname), [])
771 self.failUnlessEqual(TailFile(fname, lines=25), [])
773 def testAllLines(self):
774 data = ["test %d" % i for i in range(30)]
776 fname = self._CreateTempFile()
777 fd = open(fname, "w")
778 fd.write("\n".join(data[:i]))
782 self.failUnlessEqual(TailFile(fname, lines=i), data[:i])
784 def testPartialLines(self):
785 data = ["test %d" % i for i in range(30)]
786 fname = self._CreateTempFile()
787 fd = open(fname, "w")
788 fd.write("\n".join(data))
791 for i in range(1, 30):
792 self.failUnlessEqual(TailFile(fname, lines=i), data[-i:])
794 def testBigFile(self):
795 data = ["test %d" % i for i in range(30)]
796 fname = self._CreateTempFile()
797 fd = open(fname, "w")
798 fd.write("X" * 1048576)
800 fd.write("\n".join(data))
803 for i in range(1, 30):
804 self.failUnlessEqual(TailFile(fname, lines=i), data[-i:])
807 class TestFileLock(unittest.TestCase):
808 """Test case for the FileLock class"""
811 self.tmpfile = tempfile.NamedTemporaryFile()
812 self.lock = utils.FileLock(self.tmpfile.name)
814 def testSharedNonblocking(self):
815 self.lock.Shared(blocking=False)
818 def testExclusiveNonblocking(self):
819 self.lock.Exclusive(blocking=False)
822 def testUnlockNonblocking(self):
823 self.lock.Unlock(blocking=False)
826 def testSharedBlocking(self):
827 self.lock.Shared(blocking=True)
830 def testExclusiveBlocking(self):
831 self.lock.Exclusive(blocking=True)
834 def testUnlockBlocking(self):
835 self.lock.Unlock(blocking=True)
838 def testSharedExclusiveUnlock(self):
839 self.lock.Shared(blocking=False)
840 self.lock.Exclusive(blocking=False)
841 self.lock.Unlock(blocking=False)
844 def testExclusiveSharedUnlock(self):
845 self.lock.Exclusive(blocking=False)
846 self.lock.Shared(blocking=False)
847 self.lock.Unlock(blocking=False)
850 def testCloseShared(self):
852 self.assertRaises(AssertionError, self.lock.Shared, blocking=False)
854 def testCloseExclusive(self):
856 self.assertRaises(AssertionError, self.lock.Exclusive, blocking=False)
858 def testCloseUnlock(self):
860 self.assertRaises(AssertionError, self.lock.Unlock, blocking=False)
863 class TestTimeFunctions(unittest.TestCase):
864 """Test case for time functions"""
867 self.assertEqual(utils.SplitTime(1), (1, 0))
868 self.assertEqual(utils.SplitTime(1.5), (1, 500000))
869 self.assertEqual(utils.SplitTime(1218448917.4809151), (1218448917, 480915))
870 self.assertEqual(utils.SplitTime(123.48012), (123, 480120))
871 self.assertEqual(utils.SplitTime(123.9996), (123, 999600))
872 self.assertEqual(utils.SplitTime(123.9995), (123, 999500))
873 self.assertEqual(utils.SplitTime(123.9994), (123, 999400))
874 self.assertEqual(utils.SplitTime(123.999999999), (123, 999999))
876 self.assertRaises(AssertionError, utils.SplitTime, -1)
878 self.assertEqual(utils.MergeTime((1, 0)), 1.0)
879 self.assertEqual(utils.MergeTime((1, 500000)), 1.5)
880 self.assertEqual(utils.MergeTime((1218448917, 500000)), 1218448917.5)
882 self.assertEqual(round(utils.MergeTime((1218448917, 481000)), 3), 1218448917.481)
883 self.assertEqual(round(utils.MergeTime((1, 801000)), 3), 1.801)
885 self.assertRaises(AssertionError, utils.MergeTime, (0, -1))
886 self.assertRaises(AssertionError, utils.MergeTime, (0, 1000000))
887 self.assertRaises(AssertionError, utils.MergeTime, (0, 9999999))
888 self.assertRaises(AssertionError, utils.MergeTime, (-1, 0))
889 self.assertRaises(AssertionError, utils.MergeTime, (-9999, 0))
892 class FieldSetTestCase(unittest.TestCase):
893 """Test case for FieldSets"""
895 def testSimpleMatch(self):
896 f = utils.FieldSet("a", "b", "c", "def")
897 self.failUnless(f.Matches("a"))
898 self.failIf(f.Matches("d"), "Substring matched")
899 self.failIf(f.Matches("defghi"), "Prefix string matched")
900 self.failIf(f.NonMatching(["b", "c"]))
901 self.failIf(f.NonMatching(["a", "b", "c", "def"]))
902 self.failUnless(f.NonMatching(["a", "d"]))
904 def testRegexMatch(self):
905 f = utils.FieldSet("a", "b([0-9]+)", "c")
906 self.failUnless(f.Matches("b1"))
907 self.failUnless(f.Matches("b99"))
908 self.failIf(f.Matches("b/1"))
909 self.failIf(f.NonMatching(["b12", "c"]))
910 self.failUnless(f.NonMatching(["a", "1"]))
912 class TestForceDictType(unittest.TestCase):
913 """Test case for ForceDictType"""
917 'a': constants.VTYPE_INT,
918 'b': constants.VTYPE_BOOL,
919 'c': constants.VTYPE_STRING,
920 'd': constants.VTYPE_SIZE,
923 def _fdt(self, dict, allowed_values=None):
924 if allowed_values is None:
925 ForceDictType(dict, self.key_types)
927 ForceDictType(dict, self.key_types, allowed_values=allowed_values)
931 def testSimpleDict(self):
932 self.assertEqual(self._fdt({}), {})
933 self.assertEqual(self._fdt({'a': 1}), {'a': 1})
934 self.assertEqual(self._fdt({'a': '1'}), {'a': 1})
935 self.assertEqual(self._fdt({'a': 1, 'b': 1}), {'a':1, 'b': True})
936 self.assertEqual(self._fdt({'b': 1, 'c': 'foo'}), {'b': True, 'c': 'foo'})
937 self.assertEqual(self._fdt({'b': 1, 'c': False}), {'b': True, 'c': ''})
938 self.assertEqual(self._fdt({'b': 'false'}), {'b': False})
939 self.assertEqual(self._fdt({'b': 'False'}), {'b': False})
940 self.assertEqual(self._fdt({'b': 'true'}), {'b': True})
941 self.assertEqual(self._fdt({'b': 'True'}), {'b': True})
942 self.assertEqual(self._fdt({'d': '4'}), {'d': 4})
943 self.assertEqual(self._fdt({'d': '4M'}), {'d': 4})
945 def testErrors(self):
946 self.assertRaises(errors.TypeEnforcementError, self._fdt, {'a': 'astring'})
947 self.assertRaises(errors.TypeEnforcementError, self._fdt, {'c': True})
948 self.assertRaises(errors.TypeEnforcementError, self._fdt, {'d': 'astring'})
949 self.assertRaises(errors.TypeEnforcementError, self._fdt, {'d': '4 L'})
952 class TestIsAbsNormPath(unittest.TestCase):
953 """Testing case for IsProcessAlive"""
955 def _pathTestHelper(self, path, result):
957 self.assert_(IsNormAbsPath(path),
958 "Path %s should result absolute and normalized" % path)
960 self.assert_(not IsNormAbsPath(path),
961 "Path %s should not result absolute and normalized" % path)
964 self._pathTestHelper('/etc', True)
965 self._pathTestHelper('/srv', True)
966 self._pathTestHelper('etc', False)
967 self._pathTestHelper('/etc/../root', False)
968 self._pathTestHelper('/etc/', False)
971 class TestSafeEncode(unittest.TestCase):
972 """Test case for SafeEncode"""
975 for txt in [string.digits, string.letters, string.punctuation]:
976 self.failUnlessEqual(txt, SafeEncode(txt))
978 def testDoubleEncode(self):
980 txt = SafeEncode(chr(i))
981 self.failUnlessEqual(txt, SafeEncode(txt))
983 def testUnicode(self):
984 # 1024 is high enough to catch non-direct ASCII mappings
985 for i in range(1024):
986 txt = SafeEncode(unichr(i))
987 self.failUnlessEqual(txt, SafeEncode(txt))
990 if __name__ == '__main__':