1
0
Fork 0
mirror of https://git.code.sf.net/p/archivemail/code synced 2024-12-22 15:56:27 +00:00

Fixed unittest TestMboxExclusiveLock: on some systems flock(2) is emulated with

fcntl(2) calls.  fcntl locks don't support interlocking within a process, so we
need to fork() to correctly test them.
This commit is contained in:
Nikolaus Schulz 2006-10-20 22:50:38 +00:00
parent 157db85b7f
commit 06d158c576

View file

@ -158,14 +158,47 @@ class TestMboxExclusiveLock(unittest.TestCase):
def testExclusiveLock(self): def testExclusiveLock(self):
"""exclusive_lock/unlock should create/delete an advisory lock""" """exclusive_lock/unlock should create/delete an advisory lock"""
# We're using flock(2) locks; these aren't completely portable, and on
# some systems (e.g. Solaris) they may be emulated with fcntl(2) locks,
# which have pretty different semantics. We could test real flock
# locks within this process, but that doesn't work for fcntl locks.
#
# The following code snippet heavily lends from the Python 2.5 mailbox
# unittest.
# BEGIN robbery:
# Fork off a subprocess that will lock the file for 2 seconds,
# unlock it, and then exit.
if not hasattr(os, 'fork'):
return
pid = os.fork()
if pid == 0:
# In the child, lock the mailbox.
self.mbox.exclusive_lock() self.mbox.exclusive_lock()
time.sleep(2)
self.mbox.exclusive_unlock()
os._exit(0)
# In the parent, sleep a bit to give the child time to acquire
# the lock.
time.sleep(0.5)
# The parent's file self.mbox.mbox_file shares flock locks with the
# duplicated FD in the child; reopen it so we get a different file
# table entry.
file = open(self.mbox_name, "r+") file = open(self.mbox_name, "r+")
lock_nb = fcntl.LOCK_EX | fcntl.LOCK_NB lock_nb = fcntl.LOCK_EX | fcntl.LOCK_NB
self.assertRaises(IOError, fcntl.flock, file.fileno(), lock_nb) fd = file.fileno()
try:
self.assertRaises(IOError, fcntl.flock, fd, lock_nb)
self.mbox.exclusive_unlock() finally:
fcntl.flock(file.fileno(), lock_nb) # Wait for child to exit. Locking should now succeed.
fcntl.flock(file.fileno(), fcntl.LOCK_UN) exited_pid, status = os.waitpid(pid, 0)
fcntl.flock(fd, lock_nb)
fcntl.flock(fd, fcntl.LOCK_UN)
# END robbery
def tearDown(self): def tearDown(self):
if os.path.exists(self.mbox_name): if os.path.exists(self.mbox_name):