Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

Merge

This commit is contained in:
Antoine Pitrou 2011-12-20 14:00:12 +01:00
parent 0831676962 29b15d11bb
commit 1743db5c10
1 changed files with 8 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
Lib/test/test_mailbox.py
View File

@ -17,10 +17,6 @@ try:
import fcntl import fcntl
except ImportError: except ImportError:
pass pass
try:
import multiprocessing
except ImportError:
multiprocessing = None
class TestBase(unittest.TestCase): class TestBase(unittest.TestCase):
@ -998,12 +994,13 @@ class _TestMboxMMDF(TestMailbox):
self._box = self._factory(self._path) self._box = self._factory(self._path)
@unittest.skipUnless(hasattr(os, 'fork'), "Test needs fork().") @unittest.skipUnless(hasattr(os, 'fork'), "Test needs fork().")
@unittest.skipUnless(multiprocessing, "Test needs multiprocessing.") @unittest.skipUnless(hasattr(socket, 'socketpair'), "Test needs socketpair().")
def test_lock_conflict(self): def test_lock_conflict(self):
# Fork off a child process that will lock the mailbox temporarily, # Fork off a child process that will lock the mailbox temporarily,
# unlock it and exit. # unlock it and exit.
ready = multiprocessing.Event() c, p = socket.socketpair()
done = multiprocessing.Event() self.addCleanup(c.close)
self.addCleanup(p.close)
pid = os.fork() pid = os.fork()
if pid == 0: if pid == 0:
@ -1011,22 +1008,22 @@ class _TestMboxMMDF(TestMailbox):
try: try:
# lock the mailbox, and signal the parent it can proceed # lock the mailbox, and signal the parent it can proceed
self._box.lock() self._box.lock()
ready.set() c.send(b'c')
# wait until the parent is done, and unlock the mailbox # wait until the parent is done, and unlock the mailbox
done.wait(5) c.recv(1)
self._box.unlock() self._box.unlock()
finally: finally:
os._exit(0) os._exit(0)
# In the parent, wait until the child signals it locked the mailbox. # In the parent, wait until the child signals it locked the mailbox.
ready.wait(5) p.recv(1)
try: try:
self.assertRaises(mailbox.ExternalClashError, self.assertRaises(mailbox.ExternalClashError,
self._box.lock) self._box.lock)
finally: finally:
# Signal the child it can now release the lock and exit. # Signal the child it can now release the lock and exit.
done.set() p.send(b'p')
# Wait for child to exit. Locking should now succeed. # Wait for child to exit. Locking should now succeed.
exited_pid, status = os.waitpid(pid, 0) exited_pid, status = os.waitpid(pid, 0)