import testutils
from ganeti import constants
from ganeti import utils
+from ganeti import errors
from ganeti.utils import IsProcessAlive, RunCmd, \
RemoveFile, CheckDict, MatchNameComponent, FormatUnit, \
ParseUnit, AddAuthorizedKey, RemoveAuthorizedKey, \
ShellQuote, ShellQuoteArgs, TcpPing, ListVisibleFiles, \
- SetEtcHostsEntry, RemoveEtcHostsEntry, FirstFree, OwnIpAddress
+ SetEtcHostsEntry, RemoveEtcHostsEntry, FirstFree, OwnIpAddress, \
+ TailFile, ForceDictType, IsNormAbsPath
+
from ganeti.errors import LockError, UnitParseError, GenericError, \
ProgrammerError
class TestIsProcessAlive(unittest.TestCase):
"""Testing case for IsProcessAlive"""
- def _CreateZombie(self):
- # create a zombie
- r_fd, w_fd = os.pipe()
- pid_zombie = os.fork()
- if pid_zombie == 0:
- # explicit close of read, write end will be closed only due to exit
- os.close(r_fd)
- os._exit(0)
- elif pid_zombie < 0:
- raise SystemError("can't fork")
- # parent: we close our end of the w_fd, so reads will error as
- # soon as the OS cleans the child's filedescriptors on its exit
- os.close(w_fd)
- # wait for 60 seconds at max for the exit (pathological case, just
- # so that the test doesn't hang indefinitely)
- r_set, w_set, e_set = select.select([r_fd], [], [], 60)
- if not r_set and not e_set:
- self.fail("Timeout exceeded in zombie creation")
- return pid_zombie
-
def testExists(self):
mypid = os.getpid()
self.assert_(IsProcessAlive(mypid),
"can't find myself running")
- def testZombie(self):
- pid_zombie = self._CreateZombie()
- is_zombie = not IsProcessAlive(pid_zombie)
- self.assert_(is_zombie, "zombie not detected as zombie")
- os.waitpid(pid_zombie, os.WNOHANG)
-
def testNotExisting(self):
pid_non_existing = os.fork()
if pid_non_existing == 0:
read_pid = utils.ReadPidFile(pid_file)
self.failUnlessEqual(read_pid, new_pid)
self.failUnless(utils.IsProcessAlive(new_pid))
- utils.KillProcess(new_pid)
+ utils.KillProcess(new_pid, waitpid=True)
self.failIf(utils.IsProcessAlive(new_pid))
utils.RemovePidFile('child')
self.failUnlessRaises(ProgrammerError, utils.KillProcess, 0)
os.rmdir(self.dir)
-class TestRunCmd(unittest.TestCase):
+class TestRunCmd(testutils.GanetiTestCase):
"""Testing case for the RunCmd function"""
def setUp(self):
+ testutils.GanetiTestCase.setUp(self)
self.magic = time.ctime() + " ganeti test"
+ self.fname = self._CreateTempFile()
def testOk(self):
"""Test successful exit code"""
result = RunCmd("/bin/sh -c 'exit 0'")
self.assertEqual(result.exit_code, 0)
+ self.assertEqual(result.output, "")
def testFail(self):
"""Test fail exit code"""
result = RunCmd("/bin/sh -c 'exit 1'")
self.assertEqual(result.exit_code, 1)
-
+ self.assertEqual(result.output, "")
def testStdout(self):
"""Test standard output"""
cmd = 'echo -n "%s"' % self.magic
result = RunCmd("/bin/sh -c '%s'" % cmd)
self.assertEqual(result.stdout, self.magic)
-
+ result = RunCmd("/bin/sh -c '%s'" % cmd, output=self.fname)
+ self.assertEqual(result.output, "")
+ self.assertFileContent(self.fname, self.magic)
def testStderr(self):
"""Test standard error"""
cmd = 'echo -n "%s"' % self.magic
result = RunCmd("/bin/sh -c '%s' 1>&2" % cmd)
self.assertEqual(result.stderr, self.magic)
-
+ result = RunCmd("/bin/sh -c '%s' 1>&2" % cmd, output=self.fname)
+ self.assertEqual(result.output, "")
+ self.assertFileContent(self.fname, self.magic)
def testCombined(self):
"""Test combined output"""
cmd = 'echo -n "A%s"; echo -n "B%s" 1>&2' % (self.magic, self.magic)
+ expected = "A" + self.magic + "B" + self.magic
result = RunCmd("/bin/sh -c '%s'" % cmd)
- self.assertEqual(result.output, "A" + self.magic + "B" + self.magic)
+ self.assertEqual(result.output, expected)
+ result = RunCmd("/bin/sh -c '%s'" % cmd, output=self.fname)
+ self.assertEqual(result.output, "")
+ self.assertFileContent(self.fname, expected)
def testSignal(self):
"""Test signal"""
result = RunCmd(["python", "-c", "import os; os.kill(os.getpid(), 15)"])
self.assertEqual(result.signal, 15)
+ self.assertEqual(result.output, "")
def testListRun(self):
"""Test list runs"""
self.assertEqual(result.exit_code, 0)
self.assertEqual(result.stdout, self.magic)
+ def testFileEmptyOutput(self):
+ """Test file output"""
+ result = RunCmd(["true"], output=self.fname)
+ self.assertEqual(result.signal, None)
+ self.assertEqual(result.exit_code, 0)
+ self.assertFileContent(self.fname, "")
+
def testLang(self):
"""Test locale environment"""
old_env = os.environ.copy()
finally:
os.environ = old_env
+ def testDefaultCwd(self):
+ """Test default working directory"""
+ self.failUnlessEqual(RunCmd(["pwd"]).stdout.strip(), "/")
+
+ def testCwd(self):
+ """Test default working directory"""
+ self.failUnlessEqual(RunCmd(["pwd"], cwd="/").stdout.strip(), "/")
+ self.failUnlessEqual(RunCmd(["pwd"], cwd="/tmp").stdout.strip(), "/tmp")
+ cwd = os.getcwd()
+ self.failUnlessEqual(RunCmd(["pwd"], cwd=cwd).stdout.strip(), cwd)
+
class TestRemoveFile(unittest.TestCase):
"""Test case for the RemoveFile function"""
self.fail("File '%s' not removed" % symlink)
+class TestRename(unittest.TestCase):
+ """Test case for RenameFile"""
+
+ def setUp(self):
+ """Create a temporary directory"""
+ self.tmpdir = tempfile.mkdtemp()
+ self.tmpfile = os.path.join(self.tmpdir, "test1")
+
+ # Touch the file
+ open(self.tmpfile, "w").close()
+
+ def tearDown(self):
+ """Remove temporary directory"""
+ shutil.rmtree(self.tmpdir)
+
+ def testSimpleRename1(self):
+ """Simple rename 1"""
+ utils.RenameFile(self.tmpfile, os.path.join(self.tmpdir, "xyz"))
+
+ def testSimpleRename2(self):
+ """Simple rename 2"""
+ utils.RenameFile(self.tmpfile, os.path.join(self.tmpdir, "xyz"),
+ mkdir=True)
+
+ def testRenameMkdir(self):
+ """Rename with mkdir"""
+ utils.RenameFile(self.tmpfile, os.path.join(self.tmpdir, "test/xyz"),
+ mkdir=True)
+
+
class TestCheckdict(unittest.TestCase):
"""Test case for the CheckDict function"""
"""Test case for the FormatUnit function"""
def testMiB(self):
- self.assertEqual(FormatUnit(1), '1M')
- self.assertEqual(FormatUnit(100), '100M')
- self.assertEqual(FormatUnit(1023), '1023M')
+ self.assertEqual(FormatUnit(1, 'h'), '1M')
+ self.assertEqual(FormatUnit(100, 'h'), '100M')
+ self.assertEqual(FormatUnit(1023, 'h'), '1023M')
+
+ self.assertEqual(FormatUnit(1, 'm'), '1')
+ self.assertEqual(FormatUnit(100, 'm'), '100')
+ self.assertEqual(FormatUnit(1023, 'm'), '1023')
+
+ self.assertEqual(FormatUnit(1024, 'm'), '1024')
+ self.assertEqual(FormatUnit(1536, 'm'), '1536')
+ self.assertEqual(FormatUnit(17133, 'm'), '17133')
+ self.assertEqual(FormatUnit(1024 * 1024 - 1, 'm'), '1048575')
def testGiB(self):
- self.assertEqual(FormatUnit(1024), '1.0G')
- self.assertEqual(FormatUnit(1536), '1.5G')
- self.assertEqual(FormatUnit(17133), '16.7G')
- self.assertEqual(FormatUnit(1024 * 1024 - 1), '1024.0G')
+ self.assertEqual(FormatUnit(1024, 'h'), '1.0G')
+ self.assertEqual(FormatUnit(1536, 'h'), '1.5G')
+ self.assertEqual(FormatUnit(17133, 'h'), '16.7G')
+ self.assertEqual(FormatUnit(1024 * 1024 - 1, 'h'), '1024.0G')
+
+ self.assertEqual(FormatUnit(1024, 'g'), '1.0')
+ self.assertEqual(FormatUnit(1536, 'g'), '1.5')
+ self.assertEqual(FormatUnit(17133, 'g'), '16.7')
+ self.assertEqual(FormatUnit(1024 * 1024 - 1, 'g'), '1024.0')
+
+ self.assertEqual(FormatUnit(1024 * 1024, 'g'), '1024.0')
+ self.assertEqual(FormatUnit(5120 * 1024, 'g'), '5120.0')
+ self.assertEqual(FormatUnit(29829 * 1024, 'g'), '29829.0')
def testTiB(self):
- self.assertEqual(FormatUnit(1024 * 1024), '1.0T')
- self.assertEqual(FormatUnit(5120 * 1024), '5.0T')
- self.assertEqual(FormatUnit(29829 * 1024), '29.1T')
+ self.assertEqual(FormatUnit(1024 * 1024, 'h'), '1.0T')
+ self.assertEqual(FormatUnit(5120 * 1024, 'h'), '5.0T')
+ self.assertEqual(FormatUnit(29829 * 1024, 'h'), '29.1T')
+ self.assertEqual(FormatUnit(1024 * 1024, 't'), '1.0')
+ self.assertEqual(FormatUnit(5120 * 1024, 't'), '5.0')
+ self.assertEqual(FormatUnit(29829 * 1024, 't'), '29.1')
class TestParseUnit(unittest.TestCase):
"""Test case for the ParseUnit function"""
'ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b')
def setUp(self):
- (fd, self.tmpname) = tempfile.mkstemp(prefix='ganeti-test')
+ testutils.GanetiTestCase.setUp(self)
+ self.tmpname = self._CreateTempFile()
+ handle = open(self.tmpname, 'w')
try:
- handle = os.fdopen(fd, 'w')
- try:
- handle.write("%s\n" % TestSshKeys.KEY_A)
- handle.write("%s\n" % TestSshKeys.KEY_B)
- finally:
- handle.close()
- except:
- utils.RemoveFile(self.tmpname)
- raise
-
- def tearDown(self):
- utils.RemoveFile(self.tmpname)
- del self.tmpname
+ handle.write("%s\n" % TestSshKeys.KEY_A)
+ handle.write("%s\n" % TestSshKeys.KEY_B)
+ finally:
+ handle.close()
def testAddingNewKey(self):
AddAuthorizedKey(self.tmpname, 'ssh-dss AAAAB3NzaC1kc3MAAACB root@test')
"""Test functions modifying /etc/hosts"""
def setUp(self):
- (fd, self.tmpname) = tempfile.mkstemp(prefix='ganeti-test')
+ testutils.GanetiTestCase.setUp(self)
+ self.tmpname = self._CreateTempFile()
+ handle = open(self.tmpname, 'w')
try:
- handle = os.fdopen(fd, 'w')
- try:
- handle.write('# This is a test file for /etc/hosts\n')
- handle.write('127.0.0.1\tlocalhost\n')
- handle.write('192.168.1.1 router gw\n')
- finally:
- handle.close()
- except:
- utils.RemoveFile(self.tmpname)
- raise
-
- def tearDown(self):
- utils.RemoveFile(self.tmpname)
- del self.tmpname
+ handle.write('# This is a test file for /etc/hosts\n')
+ handle.write('127.0.0.1\tlocalhost\n')
+ handle.write('192.168.1.1 router gw\n')
+ finally:
+ handle.close()
def testSettingNewIp(self):
SetEtcHostsEntry(self.tmpname, '1.2.3.4', 'myhost.domain.tld', ['myhost'])
"127.0.0.1\tlocalhost\n"
"192.168.1.1 router gw\n"
"1.2.3.4\tmyhost.domain.tld myhost\n")
+ self.assertFileMode(self.tmpname, 0644)
def testSettingExistingIp(self):
SetEtcHostsEntry(self.tmpname, '192.168.1.1', 'myhost.domain.tld',
"# This is a test file for /etc/hosts\n"
"127.0.0.1\tlocalhost\n"
"192.168.1.1\tmyhost.domain.tld myhost\n")
+ self.assertFileMode(self.tmpname, 0644)
def testSettingDuplicateName(self):
SetEtcHostsEntry(self.tmpname, '1.2.3.4', 'myhost', ['myhost'])
"127.0.0.1\tlocalhost\n"
"192.168.1.1 router gw\n"
"1.2.3.4\tmyhost\n")
+ self.assertFileMode(self.tmpname, 0644)
def testRemovingExistingHost(self):
RemoveEtcHostsEntry(self.tmpname, 'router')
"# This is a test file for /etc/hosts\n"
"127.0.0.1\tlocalhost\n"
"192.168.1.1 gw\n")
+ self.assertFileMode(self.tmpname, 0644)
def testRemovingSingleExistingHost(self):
RemoveEtcHostsEntry(self.tmpname, 'localhost')
self.assertFileContent(self.tmpname,
"# This is a test file for /etc/hosts\n"
"192.168.1.1 router gw\n")
+ self.assertFileMode(self.tmpname, 0644)
def testRemovingNonExistingHost(self):
RemoveEtcHostsEntry(self.tmpname, 'myhost')
"# This is a test file for /etc/hosts\n"
"127.0.0.1\tlocalhost\n"
"192.168.1.1 router gw\n")
+ self.assertFileMode(self.tmpname, 0644)
def testRemovingAlias(self):
RemoveEtcHostsEntry(self.tmpname, 'gw')
"# This is a test file for /etc/hosts\n"
"127.0.0.1\tlocalhost\n"
"192.168.1.1 router\n")
+ self.assertFileMode(self.tmpname, 0644)
class TestShellQuoting(unittest.TestCase):
self.failUnlessRaises(AssertionError, FirstFree, [0, 3, 4, 6], base=3)
+class TestTailFile(testutils.GanetiTestCase):
+ """Test case for the TailFile function"""
+
+ def testEmpty(self):
+ fname = self._CreateTempFile()
+ self.failUnlessEqual(TailFile(fname), [])
+ self.failUnlessEqual(TailFile(fname, lines=25), [])
+
+ def testAllLines(self):
+ data = ["test %d" % i for i in range(30)]
+ for i in range(30):
+ fname = self._CreateTempFile()
+ fd = open(fname, "w")
+ fd.write("\n".join(data[:i]))
+ if i > 0:
+ fd.write("\n")
+ fd.close()
+ self.failUnlessEqual(TailFile(fname, lines=i), data[:i])
+
+ def testPartialLines(self):
+ data = ["test %d" % i for i in range(30)]
+ fname = self._CreateTempFile()
+ fd = open(fname, "w")
+ fd.write("\n".join(data))
+ fd.write("\n")
+ fd.close()
+ for i in range(1, 30):
+ self.failUnlessEqual(TailFile(fname, lines=i), data[-i:])
+
+ def testBigFile(self):
+ data = ["test %d" % i for i in range(30)]
+ fname = self._CreateTempFile()
+ fd = open(fname, "w")
+ fd.write("X" * 1048576)
+ fd.write("\n")
+ fd.write("\n".join(data))
+ fd.write("\n")
+ fd.close()
+ for i in range(1, 30):
+ self.failUnlessEqual(TailFile(fname, lines=i), data[-i:])
+
+
class TestFileLock(unittest.TestCase):
"""Test case for the FileLock class"""
self.assertRaises(AssertionError, utils.MergeTime, (-9999, 0))
+class FieldSetTestCase(unittest.TestCase):
+ """Test case for FieldSets"""
+
+ def testSimpleMatch(self):
+ f = utils.FieldSet("a", "b", "c", "def")
+ self.failUnless(f.Matches("a"))
+ self.failIf(f.Matches("d"), "Substring matched")
+ self.failIf(f.Matches("defghi"), "Prefix string matched")
+ self.failIf(f.NonMatching(["b", "c"]))
+ self.failIf(f.NonMatching(["a", "b", "c", "def"]))
+ self.failUnless(f.NonMatching(["a", "d"]))
+
+ def testRegexMatch(self):
+ f = utils.FieldSet("a", "b([0-9]+)", "c")
+ self.failUnless(f.Matches("b1"))
+ self.failUnless(f.Matches("b99"))
+ self.failIf(f.Matches("b/1"))
+ self.failIf(f.NonMatching(["b12", "c"]))
+ self.failUnless(f.NonMatching(["a", "1"]))
+
+class TestForceDictType(unittest.TestCase):
+ """Test case for ForceDictType"""
+
+ def setUp(self):
+ self.key_types = {
+ 'a': constants.VTYPE_INT,
+ 'b': constants.VTYPE_BOOL,
+ 'c': constants.VTYPE_STRING,
+ 'd': constants.VTYPE_SIZE,
+ }
+
+ def _fdt(self, dict, allowed_values=None):
+ if allowed_values is None:
+ ForceDictType(dict, self.key_types)
+ else:
+ ForceDictType(dict, self.key_types, allowed_values=allowed_values)
+
+ return dict
+
+ def testSimpleDict(self):
+ self.assertEqual(self._fdt({}), {})
+ self.assertEqual(self._fdt({'a': 1}), {'a': 1})
+ self.assertEqual(self._fdt({'a': '1'}), {'a': 1})
+ self.assertEqual(self._fdt({'a': 1, 'b': 1}), {'a':1, 'b': True})
+ self.assertEqual(self._fdt({'b': 1, 'c': 'foo'}), {'b': True, 'c': 'foo'})
+ self.assertEqual(self._fdt({'b': 1, 'c': False}), {'b': True, 'c': ''})
+ self.assertEqual(self._fdt({'b': 'false'}), {'b': False})
+ self.assertEqual(self._fdt({'b': 'False'}), {'b': False})
+ self.assertEqual(self._fdt({'b': 'true'}), {'b': True})
+ self.assertEqual(self._fdt({'b': 'True'}), {'b': True})
+ self.assertEqual(self._fdt({'d': '4'}), {'d': 4})
+ self.assertEqual(self._fdt({'d': '4M'}), {'d': 4})
+
+ def testErrors(self):
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, {'a': 'astring'})
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, {'c': True})
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, {'d': 'astring'})
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, {'d': '4 L'})
+
+
+class TestIsAbsNormPath(unittest.TestCase):
+ """Testing case for IsProcessAlive"""
+
+ def _pathTestHelper(self, path, result):
+ if result:
+ self.assert_(IsNormAbsPath(path),
+ "Path %s should result absolute and normalized" % path)
+ else:
+ self.assert_(not IsNormAbsPath(path),
+ "Path %s should not result absolute and normalized" % path)
+
+ def testBase(self):
+ self._pathTestHelper('/etc', True)
+ self._pathTestHelper('/srv', True)
+ self._pathTestHelper('etc', False)
+ self._pathTestHelper('/etc/../root', False)
+ self._pathTestHelper('/etc/', False)
+
if __name__ == '__main__':
unittest.main()
projects / ganeti-local / blobdiff