#!/usr/bin/python
#
-# Copyright (C) 2006, 2007 Google Inc.
+# Copyright (C) 2006, 2007, 2010, 2011 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
"""Script for unittesting the utils module"""
-import unittest
+import errno
+import fcntl
+import glob
import os
-import time
-import tempfile
import os.path
-import md5
+import re
+import shutil
+import signal
import socket
-from errno import EADDRNOTAVAIL
-
-import ganeti
-from ganeti.utils import IsProcessAlive, Lock, Unlock, RunCmd, \
- RemoveFile, CheckDict, MatchNameComponent, FormatUnit, \
- ParseUnit, AddAuthorizedKey, RemoveAuthorizedKey, \
- ShellQuote, ShellQuoteArgs, _ParseIpOutput, TcpPing
-from ganeti.errors import LockError, UnitParseError
-
-
-class TestIsProcessAlive(unittest.TestCase):
- """Testing case for IsProcessAlive"""
- def setUp(self):
- # create a zombie and a (hopefully) non-existing process id
- self.pid_zombie = os.fork()
- if self.pid_zombie == 0:
- os._exit(0)
- elif self.pid_zombie < 0:
- raise SystemError("can't fork")
- self.pid_non_existing = os.fork()
- if self.pid_non_existing == 0:
- os._exit(0)
- elif self.pid_non_existing > 0:
- os.waitpid(self.pid_non_existing, 0)
- else:
- raise SystemError("can't fork")
-
-
- def testExists(self):
- mypid = os.getpid()
- self.assert_(IsProcessAlive(mypid),
- "can't find myself running")
-
- def testZombie(self):
- self.assert_(not IsProcessAlive(self.pid_zombie),
- "zombie not detected as zombie")
+import stat
+import tempfile
+import time
+import unittest
+import warnings
+import random
+import operator
+import testutils
+from ganeti import constants
+from ganeti import compat
+from ganeti import utils
+from ganeti import errors
+from ganeti.utils import RunCmd, \
+ FirstFree, \
+ RunParts
- def testNotExisting(self):
- self.assert_(not IsProcessAlive(self.pid_non_existing),
- "noexisting process detected")
+class TestParseCpuMask(unittest.TestCase):
+ """Test case for the ParseCpuMask function."""
-class TestLocking(unittest.TestCase):
- """Testing case for the Lock/Unlock functions"""
- def clean_lock(self, name):
- try:
- ganeti.utils.Unlock("unittest")
- except LockError:
- pass
+ def testWellFormed(self):
+ self.assertEqual(utils.ParseCpuMask(""), [])
+ self.assertEqual(utils.ParseCpuMask("1"), [1])
+ self.assertEqual(utils.ParseCpuMask("0-2,4,5-5"), [0,1,2,4,5])
+ def testInvalidInput(self):
+ for data in ["garbage", "0,", "0-1-2", "2-1", "1-a"]:
+ self.assertRaises(errors.ParseError, utils.ParseCpuMask, data)
- def testLock(self):
- self.clean_lock("unittest")
- self.assertEqual(None, Lock("unittest"))
+class TestParseMultiCpuMask(unittest.TestCase):
+ """Test case for the ParseMultiCpuMask function."""
- def testUnlock(self):
- self.clean_lock("unittest")
- ganeti.utils.Lock("unittest")
- self.assertEqual(None, Unlock("unittest"))
+ def testWellFormed(self):
+ self.assertEqual(utils.ParseMultiCpuMask(""), [])
+ self.assertEqual(utils.ParseMultiCpuMask("1"), [[1]])
+ self.assertEqual(utils.ParseMultiCpuMask("0-2,4,5-5"), [[0, 1, 2, 4, 5]])
+ self.assertEqual(utils.ParseMultiCpuMask("all"), [[-1]])
+ self.assertEqual(utils.ParseMultiCpuMask("0-2:all:4,6-8"),
+ [[0, 1, 2], [-1], [4, 6, 7, 8]])
+ def testInvalidInput(self):
+ for data in ["garbage", "0,", "0-1-2", "2-1", "1-a", "all-all"]:
+ self.assertRaises(errors.ParseError, utils.ParseCpuMask, data)
- def testDoubleLock(self):
- self.clean_lock("unittest")
- ganeti.utils.Lock("unittest")
- self.assertRaises(LockError, Lock, "unittest")
+class TestGetMounts(unittest.TestCase):
+ """Test case for GetMounts()."""
-class TestRunCmd(unittest.TestCase):
- """Testing case for the RunCmd function"""
+ TESTDATA = (
+ "rootfs / rootfs rw 0 0\n"
+ "none /sys sysfs rw,nosuid,nodev,noexec,relatime 0 0\n"
+ "none /proc proc rw,nosuid,nodev,noexec,relatime 0 0\n")
def setUp(self):
- self.magic = time.ctime() + " ganeti test"
-
- def testOk(self):
- """Test successful exit code"""
- result = RunCmd("/bin/sh -c 'exit 0'")
- self.assertEqual(result.exit_code, 0)
-
- def testFail(self):
- """Test fail exit code"""
- result = RunCmd("/bin/sh -c 'exit 1'")
- self.assertEqual(result.exit_code, 1)
-
-
- 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)
-
-
- 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)
-
-
- def testCombined(self):
- """Test combined output"""
- cmd = 'echo -n "A%s"; echo -n "B%s" 1>&2' % (self.magic, self.magic)
- result = RunCmd("/bin/sh -c '%s'" % cmd)
- self.assertEqual(result.output, "A" + self.magic + "B" + self.magic)
-
- def testSignal(self):
- """Test standard error"""
- result = RunCmd("/bin/sh -c 'kill -15 $$'")
- self.assertEqual(result.signal, 15)
-
- def testListRun(self):
- """Test list runs"""
- result = RunCmd(["true"])
- self.assertEqual(result.signal, None)
- self.assertEqual(result.exit_code, 0)
- result = RunCmd(["/bin/sh", "-c", "exit 1"])
- self.assertEqual(result.signal, None)
- self.assertEqual(result.exit_code, 1)
- result = RunCmd(["echo", "-n", self.magic])
- self.assertEqual(result.signal, None)
- self.assertEqual(result.exit_code, 0)
- self.assertEqual(result.stdout, self.magic)
-
- def testLang(self):
- """Test locale environment"""
- old_env = os.environ.copy()
- try:
- os.environ["LANG"] = "en_US.UTF-8"
- os.environ["LC_ALL"] = "en_US.UTF-8"
- result = RunCmd(["locale"])
- for line in result.output.splitlines():
- key, value = line.split("=", 1)
- # Ignore these variables, they're overridden by LC_ALL
- if key == "LANG" or key == "LANGUAGE":
- continue
- self.failIf(value and value != "C" and value != '"C"',
- "Variable %s is set to the invalid value '%s'" % (key, value))
- finally:
- os.environ = old_env
-
-
-class TestRemoveFile(unittest.TestCase):
- """Test case for the RemoveFile function"""
-
+ self.tmpfile = tempfile.NamedTemporaryFile()
+ utils.WriteFile(self.tmpfile.name, data=self.TESTDATA)
+
+ def testGetMounts(self):
+ self.assertEqual(utils.GetMounts(filename=self.tmpfile.name),
+ [
+ ("rootfs", "/", "rootfs", "rw"),
+ ("none", "/sys", "sysfs", "rw,nosuid,nodev,noexec,relatime"),
+ ("none", "/proc", "proc", "rw,nosuid,nodev,noexec,relatime"),
+ ])
+
+
+class TestFirstFree(unittest.TestCase):
+ """Test case for the FirstFree function"""
+
+ def test(self):
+ """Test FirstFree"""
+ self.failUnlessEqual(FirstFree([0, 1, 3]), 2)
+ self.failUnlessEqual(FirstFree([]), None)
+ self.failUnlessEqual(FirstFree([3, 4, 6]), 0)
+ self.failUnlessEqual(FirstFree([3, 4, 6], base=3), 5)
+ self.failUnlessRaises(AssertionError, FirstFree, [0, 3, 4, 6], base=3)
+
+
+class TestTimeFunctions(unittest.TestCase):
+ """Test case for time functions"""
+
+ def runTest(self):
+ self.assertEqual(utils.SplitTime(1), (1, 0))
+ self.assertEqual(utils.SplitTime(1.5), (1, 500000))
+ self.assertEqual(utils.SplitTime(1218448917.4809151), (1218448917, 480915))
+ self.assertEqual(utils.SplitTime(123.48012), (123, 480120))
+ self.assertEqual(utils.SplitTime(123.9996), (123, 999600))
+ self.assertEqual(utils.SplitTime(123.9995), (123, 999500))
+ self.assertEqual(utils.SplitTime(123.9994), (123, 999400))
+ self.assertEqual(utils.SplitTime(123.999999999), (123, 999999))
+
+ self.assertRaises(AssertionError, utils.SplitTime, -1)
+
+ self.assertEqual(utils.MergeTime((1, 0)), 1.0)
+ self.assertEqual(utils.MergeTime((1, 500000)), 1.5)
+ self.assertEqual(utils.MergeTime((1218448917, 500000)), 1218448917.5)
+
+ self.assertEqual(round(utils.MergeTime((1218448917, 481000)), 3),
+ 1218448917.481)
+ self.assertEqual(round(utils.MergeTime((1, 801000)), 3), 1.801)
+
+ self.assertRaises(AssertionError, utils.MergeTime, (0, -1))
+ self.assertRaises(AssertionError, utils.MergeTime, (0, 1000000))
+ self.assertRaises(AssertionError, utils.MergeTime, (0, 9999999))
+ self.assertRaises(AssertionError, utils.MergeTime, (-1, 0))
+ 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"""
+ KEY_TYPES = {
+ "a": constants.VTYPE_INT,
+ "b": constants.VTYPE_BOOL,
+ "c": constants.VTYPE_STRING,
+ "d": constants.VTYPE_SIZE,
+ "e": constants.VTYPE_MAYBE_STRING,
+ }
+
+ def _fdt(self, dict, allowed_values=None):
+ if allowed_values is None:
+ utils.ForceDictType(dict, self.KEY_TYPES)
+ else:
+ utils.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": 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})
+ self.assertEqual(self._fdt({"e": None, }), {"e": None, })
+ self.assertEqual(self._fdt({"e": "Hello World", }), {"e": "Hello World", })
+ self.assertEqual(self._fdt({"e": False, }), {"e": "", })
+ self.assertEqual(self._fdt({"b": "hello", }, ["hello"]), {"b": "hello"})
+
+ def testErrors(self):
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, {"a": "astring"})
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, {"b": "hello"})
+ 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"})
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, {"e": object(), })
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, {"e": [], })
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, {"x": None, })
+ self.assertRaises(errors.TypeEnforcementError, self._fdt, [])
+ self.assertRaises(errors.ProgrammerError, utils.ForceDictType,
+ {"b": "hello"}, {"b": "no-such-type"})
+
+
+class TestValidateServiceName(unittest.TestCase):
+ def testValid(self):
+ testnames = [
+ 0, 1, 2, 3, 1024, 65000, 65534, 65535,
+ "ganeti",
+ "gnt-masterd",
+ "HELLO_WORLD_SVC",
+ "hello.world.1",
+ "0", "80", "1111", "65535",
+ ]
+
+ for name in testnames:
+ self.assertEqual(utils.ValidateServiceName(name), name)
+
+ def testInvalid(self):
+ testnames = [
+ -15756, -1, 65536, 133428083,
+ "", "Hello World!", "!", "'", "\"", "\t", "\n", "`",
+ "-8546", "-1", "65536",
+ (129 * "A"),
+ ]
+
+ for name in testnames:
+ self.assertRaises(errors.OpPrereqError, utils.ValidateServiceName, name)
+
+
+class TestReadLockedPidFile(unittest.TestCase):
def setUp(self):
- """Create a temp dir and file for each case"""
- self.tmpdir = tempfile.mkdtemp('', 'ganeti-unittest-')
- fd, self.tmpfile = tempfile.mkstemp('', '', self.tmpdir)
- os.close(fd)
+ self.tmpdir = tempfile.mkdtemp()
def tearDown(self):
- if os.path.exists(self.tmpfile):
- os.unlink(self.tmpfile)
- os.rmdir(self.tmpdir)
-
-
- def testIgnoreDirs(self):
- """Test that RemoveFile() ignores directories"""
- self.assertEqual(None, RemoveFile(self.tmpdir))
-
-
- def testIgnoreNotExisting(self):
- """Test that RemoveFile() ignores non-existing files"""
- RemoveFile(self.tmpfile)
- RemoveFile(self.tmpfile)
-
-
- def testRemoveFile(self):
- """Test that RemoveFile does remove a file"""
- RemoveFile(self.tmpfile)
- if os.path.exists(self.tmpfile):
- self.fail("File '%s' not removed" % self.tmpfile)
-
+ shutil.rmtree(self.tmpdir)
- def testRemoveSymlink(self):
- """Test that RemoveFile does remove symlinks"""
- symlink = self.tmpdir + "/symlink"
- os.symlink("no-such-file", symlink)
- RemoveFile(symlink)
- if os.path.exists(symlink):
- self.fail("File '%s' not removed" % symlink)
- os.symlink(self.tmpfile, symlink)
- RemoveFile(symlink)
- if os.path.exists(symlink):
- self.fail("File '%s' not removed" % symlink)
-
-
-class TestCheckdict(unittest.TestCase):
- """Test case for the CheckDict function"""
-
- def testAdd(self):
- """Test that CheckDict adds a missing key with the correct value"""
-
- tgt = {'a':1}
- tmpl = {'b': 2}
- CheckDict(tgt, tmpl)
- if 'b' not in tgt or tgt['b'] != 2:
- self.fail("Failed to update dict")
-
-
- def testNoUpdate(self):
- """Test that CheckDict does not overwrite an existing key"""
- tgt = {'a':1, 'b': 3}
- tmpl = {'b': 2}
- CheckDict(tgt, tmpl)
- self.failUnlessEqual(tgt['b'], 3)
-
-
-class TestMatchNameComponent(unittest.TestCase):
- """Test case for the MatchNameComponent function"""
-
- def testEmptyList(self):
- """Test that there is no match against an empty list"""
-
- self.failUnlessEqual(MatchNameComponent("", []), None)
- self.failUnlessEqual(MatchNameComponent("test", []), None)
-
- def testSingleMatch(self):
- """Test that a single match is performed correctly"""
- mlist = ["test1.example.com", "test2.example.com", "test3.example.com"]
- for key in "test2", "test2.example", "test2.example.com":
- self.failUnlessEqual(MatchNameComponent(key, mlist), mlist[1])
-
- def testMultipleMatches(self):
- """Test that a multiple match is returned as None"""
- mlist = ["test1.example.com", "test1.example.org", "test1.example.net"]
- for key in "test1", "test1.example":
- self.failUnlessEqual(MatchNameComponent(key, mlist), None)
-
-
-class TestFormatUnit(unittest.TestCase):
- """Test case for the FormatUnit function"""
-
- def testMiB(self):
- self.assertEqual(FormatUnit(1), '1M')
- self.assertEqual(FormatUnit(100), '100M')
- self.assertEqual(FormatUnit(1023), '1023M')
-
- 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')
-
- def testTiB(self):
- self.assertEqual(FormatUnit(1024 * 1024), '1.0T')
- self.assertEqual(FormatUnit(5120 * 1024), '5.0T')
- self.assertEqual(FormatUnit(29829 * 1024), '29.1T')
-
-
-class TestParseUnit(unittest.TestCase):
- """Test case for the ParseUnit function"""
-
- SCALES = (('', 1),
- ('M', 1), ('G', 1024), ('T', 1024 * 1024),
- ('MB', 1), ('GB', 1024), ('TB', 1024 * 1024),
- ('MiB', 1), ('GiB', 1024), ('TiB', 1024 * 1024))
-
- def testRounding(self):
- self.assertEqual(ParseUnit('0'), 0)
- self.assertEqual(ParseUnit('1'), 4)
- self.assertEqual(ParseUnit('2'), 4)
- self.assertEqual(ParseUnit('3'), 4)
-
- self.assertEqual(ParseUnit('124'), 124)
- self.assertEqual(ParseUnit('125'), 128)
- self.assertEqual(ParseUnit('126'), 128)
- self.assertEqual(ParseUnit('127'), 128)
- self.assertEqual(ParseUnit('128'), 128)
- self.assertEqual(ParseUnit('129'), 132)
- self.assertEqual(ParseUnit('130'), 132)
-
- def testFloating(self):
- self.assertEqual(ParseUnit('0'), 0)
- self.assertEqual(ParseUnit('0.5'), 4)
- self.assertEqual(ParseUnit('1.75'), 4)
- self.assertEqual(ParseUnit('1.99'), 4)
- self.assertEqual(ParseUnit('2.00'), 4)
- self.assertEqual(ParseUnit('2.01'), 4)
- self.assertEqual(ParseUnit('3.99'), 4)
- self.assertEqual(ParseUnit('4.00'), 4)
- self.assertEqual(ParseUnit('4.01'), 8)
- self.assertEqual(ParseUnit('1.5G'), 1536)
- self.assertEqual(ParseUnit('1.8G'), 1844)
- self.assertEqual(ParseUnit('8.28T'), 8682212)
-
- def testSuffixes(self):
- for sep in ('', ' ', ' ', "\t", "\t "):
- for suffix, scale in TestParseUnit.SCALES:
- for func in (lambda x: x, str.lower, str.upper):
- self.assertEqual(ParseUnit('1024' + sep + func(suffix)), 1024 * scale)
+ def testNonExistent(self):
+ path = utils.PathJoin(self.tmpdir, "nonexist")
+ self.assert_(utils.ReadLockedPidFile(path) is None)
- def testInvalidInput(self):
- for sep in ('-', '_', ',', 'a'):
- for suffix, _ in TestParseUnit.SCALES:
- self.assertRaises(UnitParseError, ParseUnit, '1' + sep + suffix)
-
- for suffix, _ in TestParseUnit.SCALES:
- self.assertRaises(UnitParseError, ParseUnit, '1,3' + suffix)
-
-
-class TestSshKeys(unittest.TestCase):
- """Test case for the AddAuthorizedKey function"""
-
- KEY_A = 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a'
- KEY_B = ('command="/usr/bin/fooserver -t --verbose",from="1.2.3.4" '
- 'ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b')
-
- # NOTE: The MD5 sums below were calculated after manually
- # checking the output files.
-
- def writeTestFile(self):
- (fd, tmpname) = tempfile.mkstemp(prefix = 'ganeti-test')
- f = os.fdopen(fd, 'w')
- try:
- f.write(TestSshKeys.KEY_A)
- f.write("\n")
- f.write(TestSshKeys.KEY_B)
- f.write("\n")
- finally:
- f.close()
-
- return tmpname
-
- def testAddingNewKey(self):
- tmpname = self.writeTestFile()
- try:
- AddAuthorizedKey(tmpname, 'ssh-dss AAAAB3NzaC1kc3MAAACB root@test')
-
- f = open(tmpname, 'r')
- try:
- self.assertEqual(md5.new(f.read(8192)).hexdigest(),
- 'ccc71523108ca6e9d0343797dc3e9f16')
- finally:
- f.close()
- finally:
- os.unlink(tmpname)
-
- def testAddingAlmostButNotCompletlyTheSameKey(self):
- tmpname = self.writeTestFile()
- try:
- AddAuthorizedKey(tmpname,
- 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@test')
-
- f = open(tmpname, 'r')
- try:
- self.assertEqual(md5.new(f.read(8192)).hexdigest(),
- 'f2c939d57addb5b3a6846884be896b46')
- finally:
- f.close()
- finally:
- os.unlink(tmpname)
+ def testUnlocked(self):
+ path = utils.PathJoin(self.tmpdir, "pid")
+ utils.WriteFile(path, data="123")
+ self.assert_(utils.ReadLockedPidFile(path) is None)
- def testAddingExistingKeyWithSomeMoreSpaces(self):
- tmpname = self.writeTestFile()
- try:
- AddAuthorizedKey(tmpname,
- 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a')
-
- f = open(tmpname, 'r')
- try:
- self.assertEqual(md5.new(f.read(8192)).hexdigest(),
- '4e612764808bd46337eb0f575415fc30')
- finally:
- f.close()
- finally:
- os.unlink(tmpname)
+ def testLocked(self):
+ path = utils.PathJoin(self.tmpdir, "pid")
+ utils.WriteFile(path, data="123")
- def testRemovingExistingKeyWithSomeMoreSpaces(self):
- tmpname = self.writeTestFile()
+ fl = utils.FileLock.Open(path)
try:
- RemoveAuthorizedKey(tmpname,
- 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a')
-
- f = open(tmpname, 'r')
- try:
- self.assertEqual(md5.new(f.read(8192)).hexdigest(),
- '77516d987fca07f70e30b830b3e4f2ed')
- finally:
- f.close()
- finally:
- os.unlink(tmpname)
+ fl.Exclusive(blocking=True)
- def testRemovingNonExistingKey(self):
- tmpname = self.writeTestFile()
- try:
- RemoveAuthorizedKey(tmpname,
- 'ssh-dss AAAAB3Nsdfj230xxjxJjsjwjsjdjU root@test')
-
- f = open(tmpname, 'r')
- try:
- self.assertEqual(md5.new(f.read(8192)).hexdigest(),
- '4e612764808bd46337eb0f575415fc30')
- finally:
- f.close()
+ self.assertEqual(utils.ReadLockedPidFile(path), 123)
finally:
- os.unlink(tmpname)
-
+ fl.Close()
-class TestShellQuoting(unittest.TestCase):
- """Test case for shell quoting functions"""
+ self.assert_(utils.ReadLockedPidFile(path) is None)
- def testShellQuote(self):
- self.assertEqual(ShellQuote('abc'), "abc")
- self.assertEqual(ShellQuote('ab"c'), "'ab\"c'")
- self.assertEqual(ShellQuote("a'bc"), "'a'\\''bc'")
- self.assertEqual(ShellQuote("a b c"), "'a b c'")
- self.assertEqual(ShellQuote("a b\\ c"), "'a b\\ c'")
+ def testError(self):
+ path = utils.PathJoin(self.tmpdir, "foobar", "pid")
+ utils.WriteFile(utils.PathJoin(self.tmpdir, "foobar"), data="")
+ # open(2) should return ENOTDIR
+ self.assertRaises(EnvironmentError, utils.ReadLockedPidFile, path)
- def testShellQuoteArgs(self):
- self.assertEqual(ShellQuoteArgs(['a', 'b', 'c']), "a b c")
- self.assertEqual(ShellQuoteArgs(['a', 'b"', 'c']), "a 'b\"' c")
- self.assertEqual(ShellQuoteArgs(['a', 'b\'', 'c']), "a 'b'\\\''' c")
+class TestFindMatch(unittest.TestCase):
+ def test(self):
+ data = {
+ "aaaa": "Four A",
+ "bb": {"Two B": True},
+ re.compile(r"^x(foo|bar|bazX)([0-9]+)$"): (1, 2, 3),
+ }
-class TestIpAdressList(unittest.TestCase):
- """Test case for local IP addresses"""
+ self.assertEqual(utils.FindMatch(data, "aaaa"), ("Four A", []))
+ self.assertEqual(utils.FindMatch(data, "bb"), ({"Two B": True}, []))
- def _test(self, output, required):
- ips = _ParseIpOutput(output)
+ for i in ["foo", "bar", "bazX"]:
+ for j in range(1, 100, 7):
+ self.assertEqual(utils.FindMatch(data, "x%s%s" % (i, j)),
+ ((1, 2, 3), [i, str(j)]))
- # Sort the output, so our check below works in all cases
- ips.sort()
- required.sort()
+ def testNoMatch(self):
+ self.assert_(utils.FindMatch({}, "") is None)
+ self.assert_(utils.FindMatch({}, "foo") is None)
+ self.assert_(utils.FindMatch({}, 1234) is None)
- self.assertEqual(required, ips)
+ data = {
+ "X": "Hello World",
+ re.compile("^(something)$"): "Hello World",
+ }
- def testSingleIpAddress(self):
- output = \
- ("3: lo inet 127.0.0.1/8 brd 127.255.255.255 scope host lo\n"
- "5: eth0 inet 10.0.0.1/24 brd 172.30.15.127 scope global eth0\n")
- self._test(output, ['127.0.0.1', '10.0.0.1'])
+ self.assert_(utils.FindMatch(data, "") is None)
+ self.assert_(utils.FindMatch(data, "Hello World") is None)
- def testMultipleIpAddresses(self):
- output = \
- ("3: lo inet 127.0.0.1/8 brd 127.255.255.255 scope host lo\n"
- "5: eth0 inet 10.0.0.1/24 brd 172.30.15.127 scope global eth0\n"
- "5: eth0 inet 1.2.3.4/8 brd 1.255.255.255 scope global eth0:test\n")
- self._test(output, ['127.0.0.1', '10.0.0.1', '1.2.3.4'])
+class TestTryConvert(unittest.TestCase):
+ def test(self):
+ for src, fn, result in [
+ ("1", int, 1),
+ ("a", int, "a"),
+ ("", bool, False),
+ ("a", bool, True),
+ ]:
+ self.assertEqual(utils.TryConvert(fn, src), result)
-class TestTcpPing(unittest.TestCase):
- """Testcase for TCP version of ping - against listen(2)ing port"""
+class TestVerifyDictOptions(unittest.TestCase):
def setUp(self):
- self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.listener.bind(("127.0.0.1", 0))
- self.listenerport = self.listener.getsockname()[1]
- self.listener.listen(1)
-
- def tearDown(self):
- self.listener.shutdown(socket.SHUT_RDWR)
- del self.listener
- del self.listenerport
-
- def testTcpPingToLocalHostAccept(self):
- self.assert_(TcpPing("127.0.0.1",
- "127.0.0.1",
- self.listenerport,
- timeout=10,
- live_port_needed=True),
- "failed to connect to test listener")
-
-
-class TestTcpPingDeaf(unittest.TestCase):
- """Testcase for TCP version of ping - against non listen(2)ing port"""
-
- def setUp(self):
- self.deaflistener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.deaflistener.bind(("127.0.0.1", 0))
- self.deaflistenerport = self.deaflistener.getsockname()[1]
-
- def tearDown(self):
- del self.deaflistener
- del self.deaflistenerport
-
- def testTcpPingToLocalHostAcceptDeaf(self):
- self.failIf(TcpPing("127.0.0.1",
- "127.0.0.1",
- self.deaflistenerport,
- timeout=10, # timeout for blocking operations
- live_port_needed=True), # need successful connect(2)
- "successfully connected to deaf listener")
-
- def testTcpPingToLocalHostNoAccept(self):
- self.assert_(TcpPing("127.0.0.1",
- "127.0.0.1",
- self.deaflistenerport,
- timeout=10, # timeout for blocking operations
- live_port_needed=False), # ECONNREFUSED is OK
- "failed to ping alive host on deaf port")
-
-
-if __name__ == '__main__':
- unittest.main()
+ self.defaults = {
+ "first_key": "foobar",
+ "foobar": {
+ "key1": "value2",
+ "key2": "value1",
+ },
+ "another_key": "another_value",
+ }
+
+ def test(self):
+ some_keys = {
+ "first_key": "blubb",
+ "foobar": {
+ "key2": "foo",
+ },
+ }
+ utils.VerifyDictOptions(some_keys, self.defaults)
+
+ def testInvalid(self):
+ some_keys = {
+ "invalid_key": "blubb",
+ "foobar": {
+ "key2": "foo",
+ },
+ }
+ self.assertRaises(errors.OpPrereqError, utils.VerifyDictOptions,
+ some_keys, self.defaults)
+
+ def testNestedInvalid(self):
+ some_keys = {
+ "foobar": {
+ "key2": "foo",
+ "key3": "blibb"
+ },
+ }
+ self.assertRaises(errors.OpPrereqError, utils.VerifyDictOptions,
+ some_keys, self.defaults)
+
+ def testMultiInvalid(self):
+ some_keys = {
+ "foobar": {
+ "key1": "value3",
+ "key6": "Right here",
+ },
+ "invalid_with_sub": {
+ "sub1": "value3",
+ },
+ }
+ self.assertRaises(errors.OpPrereqError, utils.VerifyDictOptions,
+ some_keys, self.defaults)
+
+
+if __name__ == "__main__":
+ testutils.GanetiTestProgram()
projects / ganeti-local / blobdiff