OXIESEC PANEL
- Current Dir:
/
/
usr
/
lib
/
python3
/
dist-packages
/
twisted
/
test
Server IP: 139.59.38.164
Upload:
Create Dir:
Name
Size
Modified
Perms
📁
..
-
03/31/2022 06:22:38 AM
rwxr-xr-x
📄
__init__.py
103 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📁
__pycache__
-
03/31/2022 06:22:39 AM
rwxr-xr-x
📄
cert.pem.no_trailing_newline
1.38 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
crash_test_dummy.py
543 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
iosim.py
17.3 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
key.pem.no_trailing_newline
1.67 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
mock_win32process.py
1.46 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
myrebuilder1.py
158 bytes
09/08/2017 10:38:35 AM
rw-r--r--
📄
myrebuilder2.py
158 bytes
09/08/2017 10:38:35 AM
rw-r--r--
📄
plugin_basic.py
943 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
plugin_extra1.py
407 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
plugin_extra2.py
579 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
process_cmdline.py
162 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
process_echoer.py
214 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
process_fds.py
945 bytes
09/08/2017 10:38:35 AM
rw-r--r--
📄
process_getargv.py
283 bytes
09/08/2017 10:38:35 AM
rw-r--r--
📄
process_getenv.py
268 bytes
09/08/2017 10:38:35 AM
rw-r--r--
📄
process_linger.py
286 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
process_reader.py
188 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
process_signal.py
214 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
process_stdinreader.py
857 bytes
09/08/2017 10:38:35 AM
rw-r--r--
📄
process_tester.py
1.01 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
process_tty.py
130 bytes
09/08/2017 10:38:35 AM
rw-r--r--
📄
process_twisted.py
1.18 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
proto_helpers.py
26.33 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
raiser.c
93.05 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
raiser.cpython-36m-x86_64-linux-gnu.so
19.16 KB
03/22/2022 11:03:56 AM
rw-r--r--
📄
raiser.pyx
466 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
reflect_helper_IE.py
61 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
reflect_helper_VE.py
82 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
reflect_helper_ZDE.py
47 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
server.pem
4.34 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
ssl_helpers.py
1.01 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
stdio_test_consumer.py
1.19 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
stdio_test_halfclose.py
1.89 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
stdio_test_hostpeer.py
1021 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
stdio_test_lastwrite.py
1.18 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
stdio_test_loseconn.py
1.51 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
stdio_test_producer.py
1.47 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
stdio_test_write.py
923 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
stdio_test_writeseq.py
915 bytes
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_abstract.py
3.42 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_adbapi.py
25.53 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_amp.py
107.96 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_application.py
32.05 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_compat.py
27.32 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_context.py
1.48 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_cooperator.py
20.96 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_defer.py
100.93 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_defgen.py
10.45 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_dict.py
1.41 KB
09/23/2017 05:52:22 AM
rw-r--r--
📄
test_dirdbm.py
6.76 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
test_error.py
8.39 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_factories.py
4.53 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_failure.py
29.92 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_fdesc.py
7.2 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_finger.py
1.95 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_formmethod.py
3.56 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_ftp.py
127.27 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_ftp_options.py
2.62 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_htb.py
3.12 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_ident.py
6.85 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_internet.py
45.33 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_iosim.py
8.49 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
test_iutils.py
13.13 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_lockfile.py
15.14 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_log.py
35.48 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_logfile.py
17.8 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_loopback.py
14.15 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_main.py
2.44 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
test_memcache.py
24.55 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_modules.py
17.47 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_monkey.py
5.5 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_nooldstyle.py
5.82 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
test_paths.py
72.61 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_pcp.py
12.26 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_persisted.py
14.28 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_plugin.py
25.5 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_policies.py
32.04 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_postfix.py
3.53 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_process.py
84.1 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_protocols.py
7.28 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_randbytes.py
3.28 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_rebuild.py
8.3 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_reflect.py
25.47 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_roots.py
1.77 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_shortcut.py
1.89 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
test_sip.py
24.69 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_sob.py
5.5 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_socks.py
17.32 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_ssl.py
23.29 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_sslverify.py
104.28 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_stateful.py
1.97 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_stdio.py
12.85 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_strerror.py
5.06 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_stringtransport.py
12.95 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_strports.py
1.75 KB
09/08/2017 10:38:35 AM
rw-r--r--
📄
test_task.py
38.4 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_tcp.py
64.07 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_tcp_internals.py
8.54 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_text.py
6.3 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_threadable.py
3.65 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_threadpool.py
22.47 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_threads.py
12.96 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_tpfile.py
1.56 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_twistd.py
61.05 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_twisted.py
18.42 KB
09/09/2017 05:11:54 AM
rw-r--r--
📄
test_udp.py
24.1 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_unix.py
14.8 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
test_usage.py
23.09 KB
09/08/2017 10:38:36 AM
rw-r--r--
📄
testutils.py
5.19 KB
09/08/2017 10:38:36 AM
rw-r--r--
Editing: test_lockfile.py
Close
# Copyright (c) 2005 Divmod, Inc. # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.python.lockfile}. """ from __future__ import absolute_import, division import errno import os from twisted.trial import unittest from twisted.python import lockfile from twisted.python.reflect import requireModule from twisted.python.runtime import platform skipKill = None if platform.isWindows(): if(requireModule('win32api.OpenProcess') is None and requireModule('pywintypes') is None ): skipKill = ("On windows, lockfile.kill is not implemented in the " "absence of win32api and/or pywintypes.") class UtilTests(unittest.TestCase): """ Tests for the helper functions used to implement L{FilesystemLock}. """ def test_symlinkEEXIST(self): """ L{lockfile.symlink} raises L{OSError} with C{errno} set to L{EEXIST} when an attempt is made to create a symlink which already exists. """ name = self.mktemp() lockfile.symlink('foo', name) exc = self.assertRaises(OSError, lockfile.symlink, 'foo', name) self.assertEqual(exc.errno, errno.EEXIST) def test_symlinkEIOWindows(self): """ L{lockfile.symlink} raises L{OSError} with C{errno} set to L{EIO} when the underlying L{rename} call fails with L{EIO}. Renaming a file on Windows may fail if the target of the rename is in the process of being deleted (directory deletion appears not to be atomic). """ name = self.mktemp() def fakeRename(src, dst): raise IOError(errno.EIO, None) self.patch(lockfile, 'rename', fakeRename) exc = self.assertRaises(IOError, lockfile.symlink, name, "foo") self.assertEqual(exc.errno, errno.EIO) if not platform.isWindows(): test_symlinkEIOWindows.skip = ( "special rename EIO handling only necessary and correct on " "Windows.") def test_readlinkENOENT(self): """ L{lockfile.readlink} raises L{OSError} with C{errno} set to L{ENOENT} when an attempt is made to read a symlink which does not exist. """ name = self.mktemp() exc = self.assertRaises(OSError, lockfile.readlink, name) self.assertEqual(exc.errno, errno.ENOENT) def test_readlinkEACCESWindows(self): """ L{lockfile.readlink} raises L{OSError} with C{errno} set to L{EACCES} on Windows when the underlying file open attempt fails with C{EACCES}. Opening a file on Windows may fail if the path is inside a directory which is in the process of being deleted (directory deletion appears not to be atomic). """ name = self.mktemp() def fakeOpen(path, mode): raise IOError(errno.EACCES, None) self.patch(lockfile, '_open', fakeOpen) exc = self.assertRaises(IOError, lockfile.readlink, name) self.assertEqual(exc.errno, errno.EACCES) if not platform.isWindows(): test_readlinkEACCESWindows.skip = ( "special readlink EACCES handling only necessary and correct on " "Windows.") def test_kill(self): """ L{lockfile.kill} returns without error if passed the PID of a process which exists and signal C{0}. """ lockfile.kill(os.getpid(), 0) test_kill.skip = skipKill def test_killESRCH(self): """ L{lockfile.kill} raises L{OSError} with errno of L{ESRCH} if passed a PID which does not correspond to any process. """ # Hopefully there is no process with PID 2 ** 31 - 1 exc = self.assertRaises(OSError, lockfile.kill, 2 ** 31 - 1, 0) self.assertEqual(exc.errno, errno.ESRCH) test_killESRCH.skip = skipKill def test_noKillCall(self): """ Verify that when L{lockfile.kill} does end up as None (e.g. on Windows without pywin32), it doesn't end up being called and raising a L{TypeError}. """ self.patch(lockfile, "kill", None) fl = lockfile.FilesystemLock(self.mktemp()) fl.lock() self.assertFalse(fl.lock()) class LockingTests(unittest.TestCase): def _symlinkErrorTest(self, errno): def fakeSymlink(source, dest): raise OSError(errno, None) self.patch(lockfile, 'symlink', fakeSymlink) lockf = self.mktemp() lock = lockfile.FilesystemLock(lockf) exc = self.assertRaises(OSError, lock.lock) self.assertEqual(exc.errno, errno) def test_symlinkError(self): """ An exception raised by C{symlink} other than C{EEXIST} is passed up to the caller of L{FilesystemLock.lock}. """ self._symlinkErrorTest(errno.ENOSYS) def test_symlinkErrorPOSIX(self): """ An L{OSError} raised by C{symlink} on a POSIX platform with an errno of C{EACCES} or C{EIO} is passed to the caller of L{FilesystemLock.lock}. On POSIX, unlike on Windows, these are unexpected errors which cannot be handled by L{FilesystemLock}. """ self._symlinkErrorTest(errno.EACCES) self._symlinkErrorTest(errno.EIO) if platform.isWindows(): test_symlinkErrorPOSIX.skip = ( "POSIX-specific error propagation not expected on Windows.") def test_cleanlyAcquire(self): """ If the lock has never been held, it can be acquired and the C{clean} and C{locked} attributes are set to C{True}. """ lockf = self.mktemp() lock = lockfile.FilesystemLock(lockf) self.assertTrue(lock.lock()) self.assertTrue(lock.clean) self.assertTrue(lock.locked) def test_cleanlyRelease(self): """ If a lock is released cleanly, it can be re-acquired and the C{clean} and C{locked} attributes are set to C{True}. """ lockf = self.mktemp() lock = lockfile.FilesystemLock(lockf) self.assertTrue(lock.lock()) lock.unlock() self.assertFalse(lock.locked) lock = lockfile.FilesystemLock(lockf) self.assertTrue(lock.lock()) self.assertTrue(lock.clean) self.assertTrue(lock.locked) def test_cannotLockLocked(self): """ If a lock is currently locked, it cannot be locked again. """ lockf = self.mktemp() firstLock = lockfile.FilesystemLock(lockf) self.assertTrue(firstLock.lock()) secondLock = lockfile.FilesystemLock(lockf) self.assertFalse(secondLock.lock()) self.assertFalse(secondLock.locked) def test_uncleanlyAcquire(self): """ If a lock was held by a process which no longer exists, it can be acquired, the C{clean} attribute is set to C{False}, and the C{locked} attribute is set to C{True}. """ owner = 12345 def fakeKill(pid, signal): if signal != 0: raise OSError(errno.EPERM, None) if pid == owner: raise OSError(errno.ESRCH, None) lockf = self.mktemp() self.patch(lockfile, 'kill', fakeKill) lockfile.symlink(str(owner), lockf) lock = lockfile.FilesystemLock(lockf) self.assertTrue(lock.lock()) self.assertFalse(lock.clean) self.assertTrue(lock.locked) self.assertEqual(lockfile.readlink(lockf), str(os.getpid())) def test_lockReleasedBeforeCheck(self): """ If the lock is initially held but then released before it can be examined to determine if the process which held it still exists, it is acquired and the C{clean} and C{locked} attributes are set to C{True}. """ def fakeReadlink(name): # Pretend to be another process releasing the lock. lockfile.rmlink(lockf) # Fall back to the real implementation of readlink. readlinkPatch.restore() return lockfile.readlink(name) readlinkPatch = self.patch(lockfile, 'readlink', fakeReadlink) def fakeKill(pid, signal): if signal != 0: raise OSError(errno.EPERM, None) if pid == 43125: raise OSError(errno.ESRCH, None) self.patch(lockfile, 'kill', fakeKill) lockf = self.mktemp() lock = lockfile.FilesystemLock(lockf) lockfile.symlink(str(43125), lockf) self.assertTrue(lock.lock()) self.assertTrue(lock.clean) self.assertTrue(lock.locked) def test_lockReleasedDuringAcquireSymlink(self): """ If the lock is released while an attempt is made to acquire it, the lock attempt fails and C{FilesystemLock.lock} returns C{False}. This can happen on Windows when L{lockfile.symlink} fails with L{IOError} of C{EIO} because another process is in the middle of a call to L{os.rmdir} (implemented in terms of RemoveDirectory) which is not atomic. """ def fakeSymlink(src, dst): # While another process id doing os.rmdir which the Windows # implementation of rmlink does, a rename call will fail with EIO. raise OSError(errno.EIO, None) self.patch(lockfile, 'symlink', fakeSymlink) lockf = self.mktemp() lock = lockfile.FilesystemLock(lockf) self.assertFalse(lock.lock()) self.assertFalse(lock.locked) if not platform.isWindows(): test_lockReleasedDuringAcquireSymlink.skip = ( "special rename EIO handling only necessary and correct on " "Windows.") def test_lockReleasedDuringAcquireReadlink(self): """ If the lock is initially held but is released while an attempt is made to acquire it, the lock attempt fails and L{FilesystemLock.lock} returns C{False}. """ def fakeReadlink(name): # While another process is doing os.rmdir which the # Windows implementation of rmlink does, a readlink call # will fail with EACCES. raise IOError(errno.EACCES, None) self.patch(lockfile, 'readlink', fakeReadlink) lockf = self.mktemp() lock = lockfile.FilesystemLock(lockf) lockfile.symlink(str(43125), lockf) self.assertFalse(lock.lock()) self.assertFalse(lock.locked) if not platform.isWindows(): test_lockReleasedDuringAcquireReadlink.skip = ( "special readlink EACCES handling only necessary and correct on " "Windows.") def _readlinkErrorTest(self, exceptionType, errno): def fakeReadlink(name): raise exceptionType(errno, None) self.patch(lockfile, 'readlink', fakeReadlink) lockf = self.mktemp() # Make it appear locked so it has to use readlink lockfile.symlink(str(43125), lockf) lock = lockfile.FilesystemLock(lockf) exc = self.assertRaises(exceptionType, lock.lock) self.assertEqual(exc.errno, errno) self.assertFalse(lock.locked) def test_readlinkError(self): """ An exception raised by C{readlink} other than C{ENOENT} is passed up to the caller of L{FilesystemLock.lock}. """ self._readlinkErrorTest(OSError, errno.ENOSYS) self._readlinkErrorTest(IOError, errno.ENOSYS) def test_readlinkErrorPOSIX(self): """ Any L{IOError} raised by C{readlink} on a POSIX platform passed to the caller of L{FilesystemLock.lock}. On POSIX, unlike on Windows, these are unexpected errors which cannot be handled by L{FilesystemLock}. """ self._readlinkErrorTest(IOError, errno.ENOSYS) self._readlinkErrorTest(IOError, errno.EACCES) if platform.isWindows(): test_readlinkErrorPOSIX.skip = ( "POSIX-specific error propagation not expected on Windows.") def test_lockCleanedUpConcurrently(self): """ If a second process cleans up the lock after a first one checks the lock and finds that no process is holding it, the first process does not fail when it tries to clean up the lock. """ def fakeRmlink(name): rmlinkPatch.restore() # Pretend to be another process cleaning up the lock. lockfile.rmlink(lockf) # Fall back to the real implementation of rmlink. return lockfile.rmlink(name) rmlinkPatch = self.patch(lockfile, 'rmlink', fakeRmlink) def fakeKill(pid, signal): if signal != 0: raise OSError(errno.EPERM, None) if pid == 43125: raise OSError(errno.ESRCH, None) self.patch(lockfile, 'kill', fakeKill) lockf = self.mktemp() lock = lockfile.FilesystemLock(lockf) lockfile.symlink(str(43125), lockf) self.assertTrue(lock.lock()) self.assertTrue(lock.clean) self.assertTrue(lock.locked) def test_rmlinkError(self): """ An exception raised by L{rmlink} other than C{ENOENT} is passed up to the caller of L{FilesystemLock.lock}. """ def fakeRmlink(name): raise OSError(errno.ENOSYS, None) self.patch(lockfile, 'rmlink', fakeRmlink) def fakeKill(pid, signal): if signal != 0: raise OSError(errno.EPERM, None) if pid == 43125: raise OSError(errno.ESRCH, None) self.patch(lockfile, 'kill', fakeKill) lockf = self.mktemp() # Make it appear locked so it has to use readlink lockfile.symlink(str(43125), lockf) lock = lockfile.FilesystemLock(lockf) exc = self.assertRaises(OSError, lock.lock) self.assertEqual(exc.errno, errno.ENOSYS) self.assertFalse(lock.locked) def test_killError(self): """ If L{kill} raises an exception other than L{OSError} with errno set to C{ESRCH}, the exception is passed up to the caller of L{FilesystemLock.lock}. """ def fakeKill(pid, signal): raise OSError(errno.EPERM, None) self.patch(lockfile, 'kill', fakeKill) lockf = self.mktemp() # Make it appear locked so it has to use readlink lockfile.symlink(str(43125), lockf) lock = lockfile.FilesystemLock(lockf) exc = self.assertRaises(OSError, lock.lock) self.assertEqual(exc.errno, errno.EPERM) self.assertFalse(lock.locked) def test_unlockOther(self): """ L{FilesystemLock.unlock} raises L{ValueError} if called for a lock which is held by a different process. """ lockf = self.mktemp() lockfile.symlink(str(os.getpid() + 1), lockf) lock = lockfile.FilesystemLock(lockf) self.assertRaises(ValueError, lock.unlock) def test_isLocked(self): """ L{isLocked} returns C{True} if the named lock is currently locked, C{False} otherwise. """ lockf = self.mktemp() self.assertFalse(lockfile.isLocked(lockf)) lock = lockfile.FilesystemLock(lockf) self.assertTrue(lock.lock()) self.assertTrue(lockfile.isLocked(lockf)) lock.unlock() self.assertFalse(lockfile.isLocked(lockf))