Back to index

python3.2  3.2.2
Public Member Functions
test.lock_tests.RLockTests Class Reference
Inheritance diagram for test.lock_tests.RLockTests:
Inheritance graph
[legend]
Collaboration diagram for test.lock_tests.RLockTests:
Collaboration graph
[legend]

List of all members.

Public Member Functions

def test_reacquire
def test_release_unacquired
def test_different_thread
def test__is_owned
def test_constructor
def test_acquire_destroy
def test_acquire_release
def test_try_acquire
def test_try_acquire_contended
def test_acquire_contended
def test_with
def test_thread_leak
def test_timeout
def setUp
def tearDown
def assertTimeout

Detailed Description

Tests for recursive locks.

Definition at line 233 of file lock_tests.py.


Member Function Documentation

def test.lock_tests.BaseTestCase.assertTimeout (   self,
  actual,
  expected 
) [inherited]

Definition at line 65 of file lock_tests.py.

00065 
00066     def assertTimeout(self, actual, expected):
00067         # The waiting and/or time.time() can be imprecise, which
00068         # is why comparing to the expected value would sometimes fail
00069         # (especially under Windows).
00070         self.assertGreaterEqual(actual, expected * 0.6)
00071         # Test nothing insane happened
00072         self.assertLess(actual, expected * 10.0)
00073 

Here is the call graph for this function:

Here is the caller graph for this function:

def test.lock_tests.BaseTestCase.setUp (   self) [inherited]

Reimplemented in test.lock_tests.BarrierTests.

Definition at line 58 of file lock_tests.py.

00058 
00059     def setUp(self):
00060         self._threads = support.threading_setup()

Here is the caller graph for this function:

def test.lock_tests.BaseTestCase.tearDown (   self) [inherited]

Reimplemented in test.lock_tests.BarrierTests.

Definition at line 61 of file lock_tests.py.

00061 
00062     def tearDown(self):
00063         support.threading_cleanup(*self._threads)
00064         support.reap_children()

Here is the caller graph for this function:

Definition at line 269 of file lock_tests.py.

00269 
00270     def test__is_owned(self):
00271         lock = self.locktype()
00272         self.assertFalse(lock._is_owned())
00273         lock.acquire()
00274         self.assertTrue(lock._is_owned())
00275         lock.acquire()
00276         self.assertTrue(lock._is_owned())
00277         result = []
00278         def f():
00279             result.append(lock._is_owned())
00280         Bunch(f, 1).wait_for_finished()
00281         self.assertFalse(result[0])
00282         lock.release()
00283         self.assertTrue(lock._is_owned())
00284         lock.release()
00285         self.assertFalse(lock._is_owned())
00286 

Here is the call graph for this function:

Definition at line 109 of file lock_tests.py.

00109 
00110     def test_acquire_contended(self):
00111         lock = self.locktype()
00112         lock.acquire()
00113         N = 5
00114         def f():
00115             lock.acquire()
00116             lock.release()
00117 
00118         b = Bunch(f, N)
00119         b.wait_for_started()
00120         _wait()
00121         self.assertEqual(len(b.finished), 0)
00122         lock.release()
00123         b.wait_for_finished()
00124         self.assertEqual(len(b.finished), N)

Here is the call graph for this function:

Definition at line 83 of file lock_tests.py.

00083 
00084     def test_acquire_destroy(self):
00085         lock = self.locktype()
00086         lock.acquire()
00087         del lock

Definition at line 88 of file lock_tests.py.

00088 
00089     def test_acquire_release(self):
00090         lock = self.locktype()
00091         lock.acquire()
00092         lock.release()
00093         del lock

Definition at line 79 of file lock_tests.py.

00079 
00080     def test_constructor(self):
00081         lock = self.locktype()
00082         del lock

Definition at line 258 of file lock_tests.py.

00258 
00259     def test_different_thread(self):
00260         # Cannot release from a different thread
00261         lock = self.locktype()
00262         def f():
00263             lock.acquire()
00264         b = Bunch(f, 1, True)
00265         try:
00266             self.assertRaises(RuntimeError, lock.release)
00267         finally:
00268             b.do_finish()

Here is the call graph for this function:

Definition at line 237 of file lock_tests.py.

00237 
00238     def test_reacquire(self):
00239         lock = self.locktype()
00240         lock.acquire()
00241         lock.acquire()
00242         lock.release()
00243         lock.acquire()
00244         lock.release()
00245         lock.release()

Definition at line 246 of file lock_tests.py.

00246 
00247     def test_release_unacquired(self):
00248         # Cannot release an unacquired lock
00249         lock = self.locktype()
00250         self.assertRaises(RuntimeError, lock.release)
00251         lock.acquire()
00252         lock.acquire()
00253         lock.release()
00254         lock.acquire()
00255         lock.release()
00256         lock.release()
00257         self.assertRaises(RuntimeError, lock.release)

Here is the call graph for this function:

Definition at line 141 of file lock_tests.py.

00141 
00142     def test_thread_leak(self):
00143         # The lock shouldn't leak a Thread instance when used from a foreign
00144         # (non-threading) thread.
00145         lock = self.locktype()
00146         def f():
00147             lock.acquire()
00148             lock.release()
00149         n = len(threading.enumerate())
00150         # We run many threads in the hope that existing threads ids won't
00151         # be recycled.
00152         Bunch(f, 15).wait_for_finished()
00153         if len(threading.enumerate()) != n:
00154             # There is a small window during which a Thread instance's
00155             # target function has finished running, but the Thread is still
00156             # alive and registered.  Avoid spurious failures by waiting a
00157             # bit more (seen on a buildbot).
00158             time.sleep(0.4)
00159             self.assertEqual(n, len(threading.enumerate()))

Here is the call graph for this function:

def test.lock_tests.BaseLockTests.test_timeout (   self) [inherited]

Definition at line 160 of file lock_tests.py.

00160 
00161     def test_timeout(self):
00162         lock = self.locktype()
00163         # Can't set timeout if not blocking
00164         self.assertRaises(ValueError, lock.acquire, 0, 1)
00165         # Invalid timeout values
00166         self.assertRaises(ValueError, lock.acquire, timeout=-100)
00167         self.assertRaises(OverflowError, lock.acquire, timeout=1e100)
00168         self.assertRaises(OverflowError, lock.acquire, timeout=TIMEOUT_MAX + 1)
00169         # TIMEOUT_MAX is ok
00170         lock.acquire(timeout=TIMEOUT_MAX)
00171         lock.release()
00172         t1 = time.time()
00173         self.assertTrue(lock.acquire(timeout=5))
00174         t2 = time.time()
00175         # Just a sanity test that it didn't actually wait for the timeout.
00176         self.assertLess(t2 - t1, 5)
00177         results = []
00178         def f():
00179             t1 = time.time()
00180             results.append(lock.acquire(timeout=0.5))
00181             t2 = time.time()
00182             results.append(t2 - t1)
00183         Bunch(f, 1).wait_for_finished()
00184         self.assertFalse(results[0])
00185         self.assertTimeout(results[1], 0.5)
00186 

Here is the call graph for this function:

Definition at line 94 of file lock_tests.py.

00094 
00095     def test_try_acquire(self):
00096         lock = self.locktype()
00097         self.assertTrue(lock.acquire(False))
00098         lock.release()

Here is the call graph for this function:

Definition at line 99 of file lock_tests.py.

00099 
00100     def test_try_acquire_contended(self):
00101         lock = self.locktype()
00102         lock.acquire()
00103         result = []
00104         def f():
00105             result.append(lock.acquire(False))
00106         Bunch(f, 1).wait_for_finished()
00107         self.assertFalse(result[0])
00108         lock.release()

Here is the call graph for this function:

def test.lock_tests.BaseLockTests.test_with (   self) [inherited]

Definition at line 125 of file lock_tests.py.

00125 
00126     def test_with(self):
00127         lock = self.locktype()
00128         def f():
00129             lock.acquire()
00130             lock.release()
00131         def _with(err=None):
00132             with lock:
00133                 if err is not None:
00134                     raise err
00135         _with()
00136         # Check the lock is unacquired
00137         Bunch(f, 1).wait_for_finished()
00138         self.assertRaises(TypeError, _with, TypeError)
00139         # Check the lock is unacquired
00140         Bunch(f, 1).wait_for_finished()

Here is the call graph for this function:


The documentation for this class was generated from the following file: