import time
import tempfile
import os.path
+import os
import md5
import socket
import shutil
class TestLocking(unittest.TestCase):
"""Testing case for the Lock/Unlock functions"""
+
+ def setUp(self):
+ lock_dir = tempfile.mkdtemp(prefix="ganeti.unittest.",
+ suffix=".locking")
+ self.old_lock_dir = constants.LOCK_DIR
+ constants.LOCK_DIR = lock_dir
+
+ def tearDown(self):
+ try:
+ ganeti.utils.Unlock("unittest")
+ except LockError:
+ pass
+ shutil.rmtree(constants.LOCK_DIR, ignore_errors=True)
+ constants.LOCK_DIR = self.old_lock_dir
+
def clean_lock(self, name):
try:
ganeti.utils.Unlock("unittest")
ganeti.utils.Lock("unittest")
self.assertEqual(None, Unlock("unittest"))
-
def testDoubleLock(self):
self.clean_lock("unittest")
ganeti.utils.Lock("unittest")
def testTcpPingToLocalHostAccept(self):
self.assert_(TcpPing(constants.LOCALHOST_IP_ADDRESS,
- 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 testTcpPingToLocalHostAcceptDeaf(self):
self.failIf(TcpPing(constants.LOCALHOST_IP_ADDRESS,
- constants.LOCALHOST_IP_ADDRESS,
self.deaflistenerport,
timeout=constants.TCP_PING_TIMEOUT,
- live_port_needed=True), # need successful connect(2)
+ 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(constants.LOCALHOST_IP_ADDRESS,
- constants.LOCALHOST_IP_ADDRESS,
self.deaflistenerport,
timeout=constants.TCP_PING_TIMEOUT,
- live_port_needed=False), # ECONNREFUSED is OK
+ 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 TestListVisibleFiles(unittest.TestCase):
"""Test case for ListVisibleFiles"""