Revision b4478d34 test/ganeti.utils_unittest.py
| b/test/ganeti.utils_unittest.py | ||
|---|---|---|
| 992 | 992 |
self.failUnlessEqual(TailFile(fname, lines=i), data[-i:]) |
| 993 | 993 |
|
| 994 | 994 |
|
| 995 |
class TestFileLock(unittest.TestCase):
|
|
| 995 |
class _BaseFileLockTest:
|
|
| 996 | 996 |
"""Test case for the FileLock class""" |
| 997 | 997 |
|
| 998 |
def setUp(self): |
|
| 999 |
self.tmpfile = tempfile.NamedTemporaryFile() |
|
| 1000 |
self.lock = utils.FileLock(self.tmpfile.name) |
|
| 1001 |
|
|
| 1002 | 998 |
def testSharedNonblocking(self): |
| 1003 | 999 |
self.lock.Shared(blocking=False) |
| 1004 | 1000 |
self.lock.Close() |
| ... | ... | |
| 1035 | 1031 |
self.lock.Unlock(blocking=False) |
| 1036 | 1032 |
self.lock.Close() |
| 1037 | 1033 |
|
| 1034 |
def testSimpleTimeout(self): |
|
| 1035 |
# These will succeed on the first attempt, hence a short timeout |
|
| 1036 |
self.lock.Shared(blocking=True, timeout=10.0) |
|
| 1037 |
self.lock.Exclusive(blocking=False, timeout=10.0) |
|
| 1038 |
self.lock.Unlock(blocking=True, timeout=10.0) |
|
| 1039 |
self.lock.Close() |
|
| 1040 |
|
|
| 1041 |
@staticmethod |
|
| 1042 |
def _TryLockInner(filename, shared, blocking): |
|
| 1043 |
lock = utils.FileLock.Open(filename) |
|
| 1044 |
|
|
| 1045 |
if shared: |
|
| 1046 |
fn = lock.Shared |
|
| 1047 |
else: |
|
| 1048 |
fn = lock.Exclusive |
|
| 1049 |
|
|
| 1050 |
try: |
|
| 1051 |
# The timeout doesn't really matter as the parent process waits for us to |
|
| 1052 |
# finish anyway. |
|
| 1053 |
fn(blocking=blocking, timeout=0.01) |
|
| 1054 |
except errors.LockError, err: |
|
| 1055 |
return False |
|
| 1056 |
|
|
| 1057 |
return True |
|
| 1058 |
|
|
| 1059 |
def _TryLock(self, *args): |
|
| 1060 |
return utils.RunInSeparateProcess(self._TryLockInner, self.tmpfile.name, |
|
| 1061 |
*args) |
|
| 1062 |
|
|
| 1063 |
def testTimeout(self): |
|
| 1064 |
for blocking in [True, False]: |
|
| 1065 |
self.lock.Exclusive(blocking=True) |
|
| 1066 |
self.failIf(self._TryLock(False, blocking)) |
|
| 1067 |
self.failIf(self._TryLock(True, blocking)) |
|
| 1068 |
|
|
| 1069 |
self.lock.Shared(blocking=True) |
|
| 1070 |
self.assert_(self._TryLock(True, blocking)) |
|
| 1071 |
self.failIf(self._TryLock(False, blocking)) |
|
| 1072 |
|
|
| 1038 | 1073 |
def testCloseShared(self): |
| 1039 | 1074 |
self.lock.Close() |
| 1040 | 1075 |
self.assertRaises(AssertionError, self.lock.Shared, blocking=False) |
| ... | ... | |
| 1048 | 1083 |
self.assertRaises(AssertionError, self.lock.Unlock, blocking=False) |
| 1049 | 1084 |
|
| 1050 | 1085 |
|
| 1086 |
class TestFileLockWithFilename(testutils.GanetiTestCase, _BaseFileLockTest): |
|
| 1087 |
TESTDATA = "Hello World\n" * 10 |
|
| 1088 |
|
|
| 1089 |
def setUp(self): |
|
| 1090 |
testutils.GanetiTestCase.setUp(self) |
|
| 1091 |
|
|
| 1092 |
self.tmpfile = tempfile.NamedTemporaryFile() |
|
| 1093 |
utils.WriteFile(self.tmpfile.name, data=self.TESTDATA) |
|
| 1094 |
self.lock = utils.FileLock.Open(self.tmpfile.name) |
|
| 1095 |
|
|
| 1096 |
# Ensure "Open" didn't truncate file |
|
| 1097 |
self.assertFileContent(self.tmpfile.name, self.TESTDATA) |
|
| 1098 |
|
|
| 1099 |
def tearDown(self): |
|
| 1100 |
self.assertFileContent(self.tmpfile.name, self.TESTDATA) |
|
| 1101 |
|
|
| 1102 |
testutils.GanetiTestCase.tearDown(self) |
|
| 1103 |
|
|
| 1104 |
|
|
| 1105 |
class TestFileLockWithFileObject(unittest.TestCase, _BaseFileLockTest): |
|
| 1106 |
def setUp(self): |
|
| 1107 |
self.tmpfile = tempfile.NamedTemporaryFile() |
|
| 1108 |
self.lock = utils.FileLock(open(self.tmpfile.name, "w"), self.tmpfile.name) |
|
| 1109 |
|
|
| 1110 |
|
|
| 1051 | 1111 |
class TestTimeFunctions(unittest.TestCase): |
| 1052 | 1112 |
"""Test case for time functions""" |
| 1053 | 1113 |
|
Also available in: Unified diff