import time
import tempfile
import os.path
+import os
import md5
+import signal
import socket
-
+import shutil
+import re
+import select
import ganeti
-from ganeti.utils import IsProcessAlive, Lock, Unlock, RunCmd, \
+import testutils
+from ganeti import constants
+from ganeti import utils
+from ganeti.utils import IsProcessAlive, RunCmd, \
RemoveFile, CheckDict, MatchNameComponent, FormatUnit, \
ParseUnit, AddAuthorizedKey, RemoveAuthorizedKey, \
- ShellQuote, ShellQuoteArgs, _ParseIpOutput, TcpPing
-from ganeti.errors import LockError, UnitParseError
+ ShellQuote, ShellQuoteArgs, TcpPing, ListVisibleFiles, \
+ SetEtcHostsEntry, RemoveEtcHostsEntry, FirstFree, OwnIpAddress
+from ganeti.errors import LockError, UnitParseError, GenericError, \
+ ProgrammerError
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:
+
+ 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 self.pid_non_existing > 0:
- os.waitpid(self.pid_non_existing, 0)
- else:
+ 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()
"can't find myself running")
def testZombie(self):
- self.assert_(not IsProcessAlive(self.pid_zombie),
- "zombie not detected as zombie")
-
+ 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):
- self.assert_(not IsProcessAlive(self.pid_non_existing),
- "noexisting process detected")
-
-
-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 testLock(self):
- self.clean_lock("unittest")
- self.assertEqual(None, Lock("unittest"))
+ pid_non_existing = os.fork()
+ if pid_non_existing == 0:
+ os._exit(0)
+ elif pid_non_existing < 0:
+ raise SystemError("can't fork")
+ os.waitpid(pid_non_existing, 0)
+ self.assert_(not IsProcessAlive(pid_non_existing),
+ "nonexisting process detected")
- def testUnlock(self):
- self.clean_lock("unittest")
- ganeti.utils.Lock("unittest")
- self.assertEqual(None, Unlock("unittest"))
+class TestPidFileFunctions(unittest.TestCase):
+ """Tests for WritePidFile, RemovePidFile and ReadPidFile"""
+ def setUp(self):
+ self.dir = tempfile.mkdtemp()
+ self.f_dpn = lambda name: os.path.join(self.dir, "%s.pid" % name)
+ utils.DaemonPidFileName = self.f_dpn
+
+ def testPidFileFunctions(self):
+ pid_file = self.f_dpn('test')
+ utils.WritePidFile('test')
+ self.failUnless(os.path.exists(pid_file),
+ "PID file should have been created")
+ read_pid = utils.ReadPidFile(pid_file)
+ self.failUnlessEqual(read_pid, os.getpid())
+ self.failUnless(utils.IsProcessAlive(read_pid))
+ self.failUnlessRaises(GenericError, utils.WritePidFile, 'test')
+ utils.RemovePidFile('test')
+ self.failIf(os.path.exists(pid_file),
+ "PID file should not exist anymore")
+ self.failUnlessEqual(utils.ReadPidFile(pid_file), 0,
+ "ReadPidFile should return 0 for missing pid file")
+ fh = open(pid_file, "w")
+ fh.write("blah\n")
+ fh.close()
+ self.failUnlessEqual(utils.ReadPidFile(pid_file), 0,
+ "ReadPidFile should return 0 for invalid pid file")
+ utils.RemovePidFile('test')
+ self.failIf(os.path.exists(pid_file),
+ "PID file should not exist anymore")
+
+ def testKill(self):
+ pid_file = self.f_dpn('child')
+ r_fd, w_fd = os.pipe()
+ new_pid = os.fork()
+ if new_pid == 0: #child
+ utils.WritePidFile('child')
+ os.write(w_fd, 'a')
+ signal.pause()
+ os._exit(0)
+ return
+ # else we are in the parent
+ # wait until the child has written the pid file
+ os.read(r_fd, 1)
+ read_pid = utils.ReadPidFile(pid_file)
+ self.failUnlessEqual(read_pid, new_pid)
+ self.failUnless(utils.IsProcessAlive(new_pid))
+ utils.KillProcess(new_pid)
+ self.failIf(utils.IsProcessAlive(new_pid))
+ utils.RemovePidFile('child')
+ self.failUnlessRaises(ProgrammerError, utils.KillProcess, 0)
- def testDoubleLock(self):
- self.clean_lock("unittest")
- ganeti.utils.Lock("unittest")
- self.assertRaises(LockError, Lock, "unittest")
+ def tearDown(self):
+ for name in os.listdir(self.dir):
+ os.unlink(os.path.join(self.dir, name))
+ os.rmdir(self.dir)
class TestRunCmd(unittest.TestCase):
self.assertEqual(result.output, "A" + self.magic + "B" + self.magic)
def testSignal(self):
- """Test standard error"""
- result = RunCmd("/bin/sh -c 'kill -15 $$'")
+ """Test signal"""
+ result = RunCmd(["python", "-c", "import os; os.kill(os.getpid(), 15)"])
self.assertEqual(result.signal, 15)
def testListRun(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)
+ self.assertEqual(ParseUnit('1024' + sep + func(suffix)),
+ 1024 * scale)
def testInvalidInput(self):
for sep in ('-', '_', ',', 'a'):
self.assertRaises(UnitParseError, ParseUnit, '1,3' + suffix)
-class TestSshKeys(unittest.TestCase):
+class TestSshKeys(testutils.GanetiTestCase):
"""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')
+ def setUp(self):
+ (fd, self.tmpname) = tempfile.mkstemp(prefix='ganeti-test')
try:
- f.write(TestSshKeys.KEY_A)
- f.write("\n")
- f.write(TestSshKeys.KEY_B)
- f.write("\n")
- finally:
- f.close()
+ 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
- return tmpname
+ def tearDown(self):
+ utils.RemoveFile(self.tmpname)
+ del self.tmpname
def testAddingNewKey(self):
- tmpname = self.writeTestFile()
- try:
- AddAuthorizedKey(tmpname, 'ssh-dss AAAAB3NzaC1kc3MAAACB root@test')
+ AddAuthorizedKey(self.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)
+ self.assertFileContent(self.tmpname,
+ "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a\n"
+ 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
+ " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n"
+ "ssh-dss AAAAB3NzaC1kc3MAAACB root@test\n")
- def testAddingAlmostButNotCompletlyTheSameKey(self):
- tmpname = self.writeTestFile()
- try:
- AddAuthorizedKey(tmpname,
- 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@test')
+ def testAddingAlmostButNotCompletelyTheSameKey(self):
+ AddAuthorizedKey(self.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)
+ self.assertFileContent(self.tmpname,
+ "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a\n"
+ 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
+ " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n"
+ "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@test\n")
def testAddingExistingKeyWithSomeMoreSpaces(self):
- tmpname = self.writeTestFile()
- try:
- AddAuthorizedKey(tmpname,
- 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a')
+ AddAuthorizedKey(self.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)
+ self.assertFileContent(self.tmpname,
+ "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a\n"
+ 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
+ " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n")
def testRemovingExistingKeyWithSomeMoreSpaces(self):
- tmpname = self.writeTestFile()
- try:
- RemoveAuthorizedKey(tmpname,
- 'ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a')
+ RemoveAuthorizedKey(self.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)
+ self.assertFileContent(self.tmpname,
+ 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
+ " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n")
def testRemovingNonExistingKey(self):
- tmpname = self.writeTestFile()
- try:
- RemoveAuthorizedKey(tmpname,
- 'ssh-dss AAAAB3Nsdfj230xxjxJjsjwjsjdjU root@test')
+ RemoveAuthorizedKey(self.tmpname,
+ 'ssh-dss AAAAB3Nsdfj230xxjxJjsjwjsjdjU root@test')
- f = open(tmpname, 'r')
+ self.assertFileContent(self.tmpname,
+ "ssh-dss AAAAB3NzaC1w5256closdj32mZaQU root@key-a\n"
+ 'command="/usr/bin/fooserver -t --verbose",from="1.2.3.4"'
+ " ssh-dss AAAAB3NzaC1w520smc01ms0jfJs22 root@key-b\n")
+
+
+class TestEtcHosts(testutils.GanetiTestCase):
+ """Test functions modifying /etc/hosts"""
+
+ def setUp(self):
+ (fd, self.tmpname) = tempfile.mkstemp(prefix='ganeti-test')
+ try:
+ handle = os.fdopen(fd, 'w')
try:
- self.assertEqual(md5.new(f.read(8192)).hexdigest(),
- '4e612764808bd46337eb0f575415fc30')
+ 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:
- f.close()
- finally:
- os.unlink(tmpname)
+ handle.close()
+ except:
+ utils.RemoveFile(self.tmpname)
+ raise
+
+ def tearDown(self):
+ utils.RemoveFile(self.tmpname)
+ del self.tmpname
+
+ def testSettingNewIp(self):
+ SetEtcHostsEntry(self.tmpname, '1.2.3.4', 'myhost.domain.tld', ['myhost'])
+
+ self.assertFileContent(self.tmpname,
+ "# This is a test file for /etc/hosts\n"
+ "127.0.0.1\tlocalhost\n"
+ "192.168.1.1 router gw\n"
+ "1.2.3.4\tmyhost.domain.tld myhost\n")
+
+ def testSettingExistingIp(self):
+ SetEtcHostsEntry(self.tmpname, '192.168.1.1', 'myhost.domain.tld',
+ ['myhost'])
+
+ self.assertFileContent(self.tmpname,
+ "# This is a test file for /etc/hosts\n"
+ "127.0.0.1\tlocalhost\n"
+ "192.168.1.1\tmyhost.domain.tld myhost\n")
+
+ def testSettingDuplicateName(self):
+ SetEtcHostsEntry(self.tmpname, '1.2.3.4', 'myhost', ['myhost'])
+
+ self.assertFileContent(self.tmpname,
+ "# This is a test file for /etc/hosts\n"
+ "127.0.0.1\tlocalhost\n"
+ "192.168.1.1 router gw\n"
+ "1.2.3.4\tmyhost\n")
+
+ def testRemovingExistingHost(self):
+ RemoveEtcHostsEntry(self.tmpname, 'router')
+
+ self.assertFileContent(self.tmpname,
+ "# This is a test file for /etc/hosts\n"
+ "127.0.0.1\tlocalhost\n"
+ "192.168.1.1 gw\n")
+
+ 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")
+
+ def testRemovingNonExistingHost(self):
+ RemoveEtcHostsEntry(self.tmpname, 'myhost')
+
+ self.assertFileContent(self.tmpname,
+ "# This is a test file for /etc/hosts\n"
+ "127.0.0.1\tlocalhost\n"
+ "192.168.1.1 router gw\n")
+
+ def testRemovingAlias(self):
+ RemoveEtcHostsEntry(self.tmpname, 'gw')
+
+ self.assertFileContent(self.tmpname,
+ "# This is a test file for /etc/hosts\n"
+ "127.0.0.1\tlocalhost\n"
+ "192.168.1.1 router\n")
class TestShellQuoting(unittest.TestCase):
self.assertEqual(ShellQuoteArgs(['a', 'b\'', 'c']), "a 'b'\\\''' c")
-class TestIpAdressList(unittest.TestCase):
- """Test case for local IP addresses"""
-
- def _test(self, output, required):
- ips = _ParseIpOutput(output)
-
- # Sort the output, so our check below works in all cases
- ips.sort()
- required.sort()
-
- self.assertEqual(required, ips)
-
- 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'])
-
- 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 TestTcpPing(unittest.TestCase):
"""Testcase for TCP version of ping - against listen(2)ing port"""
def setUp(self):
self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.listener.bind(("127.0.0.1", 0))
+ self.listener.bind((constants.LOCALHOST_IP_ADDRESS, 0))
self.listenerport = self.listener.getsockname()[1]
self.listener.listen(1)
del self.listenerport
def testTcpPingToLocalHostAccept(self):
- self.assert_(TcpPing("127.0.0.1",
- "127.0.0.1",
+ self.assert_(TcpPing(constants.LOCALHOST_IP_ADDRESS,
self.listenerport,
timeout=10,
- live_port_needed=True),
+ live_port_needed=True,
+ source=constants.LOCALHOST_IP_ADDRESS,
+ ),
"failed to connect to test listener")
+ self.assert_(TcpPing(constants.LOCALHOST_IP_ADDRESS,
+ self.listenerport,
+ timeout=10,
+ live_port_needed=True,
+ ),
+ "failed to connect to test listener (no source)")
+
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.deaflistener.bind((constants.LOCALHOST_IP_ADDRESS, 0))
self.deaflistenerport = self.deaflistener.getsockname()[1]
def tearDown(self):
del self.deaflistenerport
def testTcpPingToLocalHostAcceptDeaf(self):
- self.failIf(TcpPing("127.0.0.1",
- "127.0.0.1",
+ self.failIf(TcpPing(constants.LOCALHOST_IP_ADDRESS,
self.deaflistenerport,
- timeout=10, # timeout for blocking operations
- live_port_needed=True), # need successful connect(2)
+ timeout=constants.TCP_PING_TIMEOUT,
+ live_port_needed=True,
+ source=constants.LOCALHOST_IP_ADDRESS,
+ ), # need successful connect(2)
"successfully connected to deaf listener")
+ self.failIf(TcpPing(constants.LOCALHOST_IP_ADDRESS,
+ self.deaflistenerport,
+ timeout=constants.TCP_PING_TIMEOUT,
+ live_port_needed=True,
+ ), # need successful connect(2)
+ "successfully connected to deaf listener (no source addr)")
+
def testTcpPingToLocalHostNoAccept(self):
- self.assert_(TcpPing("127.0.0.1",
- "127.0.0.1",
+ self.assert_(TcpPing(constants.LOCALHOST_IP_ADDRESS,
self.deaflistenerport,
- timeout=10, # timeout for blocking operations
- live_port_needed=False), # ECONNREFUSED is OK
+ timeout=constants.TCP_PING_TIMEOUT,
+ live_port_needed=False,
+ source=constants.LOCALHOST_IP_ADDRESS,
+ ), # ECONNREFUSED is OK
"failed to ping alive host on deaf port")
+ self.assert_(TcpPing(constants.LOCALHOST_IP_ADDRESS,
+ self.deaflistenerport,
+ timeout=constants.TCP_PING_TIMEOUT,
+ live_port_needed=False,
+ ), # ECONNREFUSED is OK
+ "failed to ping alive host on deaf port (no source addr)")
+
+
+class TestOwnIpAddress(unittest.TestCase):
+ """Testcase for OwnIpAddress"""
+
+ def testOwnLoopback(self):
+ """check having the loopback ip"""
+ self.failUnless(OwnIpAddress(constants.LOCALHOST_IP_ADDRESS),
+ "Should own the loopback address")
+
+ def testNowOwnAddress(self):
+ """check that I don't own an address"""
+
+ # network 192.0.2.0/24 is reserved for test/documentation as per
+ # rfc 3330, so we *should* not have an address of this range... if
+ # this fails, we should extend the test to multiple addresses
+ DST_IP = "192.0.2.1"
+ self.failIf(OwnIpAddress(DST_IP), "Should not own IP address %s" % DST_IP)
+
+
+class TestListVisibleFiles(unittest.TestCase):
+ """Test case for ListVisibleFiles"""
+
+ def setUp(self):
+ self.path = tempfile.mkdtemp()
+
+ def tearDown(self):
+ shutil.rmtree(self.path)
+
+ def _test(self, files, expected):
+ # Sort a copy
+ expected = expected[:]
+ expected.sort()
+
+ for name in files:
+ f = open(os.path.join(self.path, name), 'w')
+ try:
+ f.write("Test\n")
+ finally:
+ f.close()
+
+ found = ListVisibleFiles(self.path)
+ found.sort()
+
+ self.assertEqual(found, expected)
+
+ def testAllVisible(self):
+ files = ["a", "b", "c"]
+ expected = files
+ self._test(files, expected)
+
+ def testNoneVisible(self):
+ files = [".a", ".b", ".c"]
+ expected = []
+ self._test(files, expected)
+
+ def testSomeVisible(self):
+ files = ["a", "b", ".c"]
+ expected = ["a", "b"]
+ self._test(files, expected)
+
+
+class TestNewUUID(unittest.TestCase):
+ """Test case for NewUUID"""
+
+ _re_uuid = re.compile('^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-'
+ '[a-f0-9]{4}-[a-f0-9]{12}$')
+
+ def runTest(self):
+ self.failUnless(self._re_uuid.match(utils.NewUUID()))
+
+
+class TestUniqueSequence(unittest.TestCase):
+ """Test case for UniqueSequence"""
+
+ def _test(self, input, expected):
+ self.assertEqual(utils.UniqueSequence(input), expected)
+
+ def runTest(self):
+ # Ordered input
+ self._test([1, 2, 3], [1, 2, 3])
+ self._test([1, 1, 2, 2, 3, 3], [1, 2, 3])
+ self._test([1, 2, 2, 3], [1, 2, 3])
+ self._test([1, 2, 3, 3], [1, 2, 3])
+
+ # Unordered input
+ self._test([1, 2, 3, 1, 2, 3], [1, 2, 3])
+ self._test([1, 1, 2, 3, 3, 1, 2], [1, 2, 3])
+
+ # Strings
+ self._test(["a", "a"], ["a"])
+ self._test(["a", "b"], ["a", "b"])
+ self._test(["a", "b", "a"], ["a", "b"])
+
+
+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 TestFileLock(unittest.TestCase):
+ """Test case for the FileLock class"""
+
+ def setUp(self):
+ self.tmpfile = tempfile.NamedTemporaryFile()
+ self.lock = utils.FileLock(self.tmpfile.name)
+
+ def testSharedNonblocking(self):
+ self.lock.Shared(blocking=False)
+ self.lock.Close()
+
+ def testExclusiveNonblocking(self):
+ self.lock.Exclusive(blocking=False)
+ self.lock.Close()
+
+ def testUnlockNonblocking(self):
+ self.lock.Unlock(blocking=False)
+ self.lock.Close()
+
+ def testSharedBlocking(self):
+ self.lock.Shared(blocking=True)
+ self.lock.Close()
+
+ def testExclusiveBlocking(self):
+ self.lock.Exclusive(blocking=True)
+ self.lock.Close()
+
+ def testUnlockBlocking(self):
+ self.lock.Unlock(blocking=True)
+ self.lock.Close()
+
+ def testSharedExclusiveUnlock(self):
+ self.lock.Shared(blocking=False)
+ self.lock.Exclusive(blocking=False)
+ self.lock.Unlock(blocking=False)
+ self.lock.Close()
+
+ def testExclusiveSharedUnlock(self):
+ self.lock.Exclusive(blocking=False)
+ self.lock.Shared(blocking=False)
+ self.lock.Unlock(blocking=False)
+ self.lock.Close()
+
+ def testCloseShared(self):
+ self.lock.Close()
+ self.assertRaises(AssertionError, self.lock.Shared, blocking=False)
+
+ def testCloseExclusive(self):
+ self.lock.Close()
+ self.assertRaises(AssertionError, self.lock.Exclusive, blocking=False)
+
+ def testCloseUnlock(self):
+ self.lock.Close()
+ self.assertRaises(AssertionError, self.lock.Unlock, blocking=False)
+
+
+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))
+
if __name__ == '__main__':
unittest.main()
projects / ganeti-local / blobdiff