Update of /cvsroot/mailman/mailman3/src/mailman/tests
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv31133/src/mailman/tests
Added Files:
Tag: 3.0
__init__.py test_lock.py
Log Message:
Initial checkin of Mailman3 source
--- NEW FILE: __init__.py ---
--- NEW FILE: test_lock.py ---
"""Tests for the lock object.
"""
import os
import time
import tempfile
import unittest
import threading
from mailman.lock import Lock
from mailman.errors import TimeOutError
class LockThread(threading.Thread):
def __init__(self, lock, event):
threading.Thread.__init__(self)
self._lock = lock
self._event = event
self._gotlock = False
def run(self):
self._lock.lock()
self._gotlock = True
# Block forever on shutdown event
self._event.wait()
self._lock.unlock()
def gotLock(self):
return self._gotlock
class TestOneLock(unittest.TestCase):
def setUp(self):
self._lockfile = tempfile.mktemp()
self._lock = Lock(self._lockfile)
def test_lock_is_locked(self):
self._lock.lock()
self.failUnless(self._lock.isLocked())
def test_unlocked_lock_is_unlocked(self):
unless = self.failUnless
unless(not self._lock.isLocked())
self._lock.lock()
self._lock.unlock()
unless(not self._lock.isLocked())
class TestTwoLocks(unittest.TestCase):
def setUp(self):
self._lockfile = tempfile.mktemp()
self._lock1 = Lock(self._lockfile)
self._lock2 = Lock(self._lockfile)
def test_lock_timeout(self):
event = threading.Event()
lockthread = LockThread(self._lock1, event)
lockthread.setDaemon(True)
lockthread.start()
while not lockthread.gotLock():
time.sleep(0.1)
try:
# Now attempt to acquire the lock with a timeout
self.assertRaises(TimeOutError, self._lock2.lock, 0.1)
finally:
# Shut the thread down
event.set()
lockthread.join()
def test_lock_gets_lock_eventually(self):
unless = self.failUnless
event1 = threading.Event()
event2 = threading.Event()
thread1 = LockThread(self._lock1, event1)
thread2 = LockThread(self._lock2, event2)
thread1.setDaemon(True)
thread2.setDaemon(True)
# Acquire the lock in the first thread
thread1.start()
while not thread1.gotLock():
time.sleep(0.1)
try:
# Start the second thread, which should block waiting on the lock
thread2.start()
unless(not thread2.gotLock())
# Shutdown the first thread, releasing the lock, which should
# eventually be acquired by the second thread
event1.set()
i = 0
while i < 10 and not thread2.gotLock():
i += 1
time.sleep(0.1)
unless(thread2.gotLock())
event2.set()
finally:
thread1.join()
thread2.join()
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(TestOneLock))
suite.addTest(unittest.makeSuite(TestTwoLocks))
return suite
_______________________________________________
Mailman-checkins mailing list
[EMAIL PROTECTED]
Unsubscribe: http://mail.python.org/mailman/options/mailman-checkins/archive%40jab.org
