Back to index

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

List of all members.

Public Member Functions

def test_acquire
def test_unacquired_wait
def test_unacquired_notify
def test_notify
def test_timeout
def test_waitfor
def test_waitfor_timeout
def setUp
def tearDown
def assertTimeout

Private Member Functions

def _check_notify

Detailed Description

Tests for condition variables.

Definition at line 355 of file lock_tests.py.


Member Function Documentation

def test.lock_tests.ConditionTests._check_notify (   self,
  cond 
) [private]

Definition at line 387 of file lock_tests.py.

00387 
00388     def _check_notify(self, cond):
00389         N = 5
00390         results1 = []
00391         results2 = []
00392         phase_num = 0
00393         def f():
00394             cond.acquire()
00395             result = cond.wait()
00396             cond.release()
00397             results1.append((result, phase_num))
00398             cond.acquire()
00399             result = cond.wait()
00400             cond.release()
00401             results2.append((result, phase_num))
00402         b = Bunch(f, N)
00403         b.wait_for_started()
00404         _wait()
00405         self.assertEqual(results1, [])
00406         # Notify 3 threads at first
00407         cond.acquire()
00408         cond.notify(3)
00409         _wait()
00410         phase_num = 1
00411         cond.release()
00412         while len(results1) < 3:
00413             _wait()
00414         self.assertEqual(results1, [(True, 1)] * 3)
00415         self.assertEqual(results2, [])
00416         # Notify 5 threads: they might be in their first or second wait
00417         cond.acquire()
00418         cond.notify(5)
00419         _wait()
00420         phase_num = 2
00421         cond.release()
00422         while len(results1) + len(results2) < 8:
00423             _wait()
00424         self.assertEqual(results1, [(True, 1)] * 3 + [(True, 2)] * 2)
00425         self.assertEqual(results2, [(True, 2)] * 3)
00426         # Notify all threads: they are all in their second wait
00427         cond.acquire()
00428         cond.notify_all()
00429         _wait()
00430         phase_num = 3
00431         cond.release()
00432         while len(results2) < 5:
00433             _wait()
00434         self.assertEqual(results1, [(True, 1)] * 3 + [(True,2)] * 2)
00435         self.assertEqual(results2, [(True, 2)] * 3 + [(True, 3)] * 2)
00436         b.wait_for_finished()

Here is the call graph for this function:

Here is the caller graph for this function:

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 360 of file lock_tests.py.

00360 
00361     def test_acquire(self):
00362         cond = self.condtype()
00363         # Be default we have an RLock: the condition can be acquired multiple
00364         # times.
00365         cond.acquire()
00366         cond.acquire()
00367         cond.release()
00368         cond.release()
00369         lock = threading.Lock()
00370         cond = self.condtype(lock)
00371         cond.acquire()
00372         self.assertFalse(lock.acquire(False))
00373         cond.release()
00374         self.assertTrue(lock.acquire(False))
00375         self.assertFalse(cond.acquire(False))
00376         lock.release()
00377         with cond:
00378             self.assertFalse(lock.acquire(False))

Here is the call graph for this function:

Definition at line 437 of file lock_tests.py.

00437 
00438     def test_notify(self):
00439         cond = self.condtype()
00440         self._check_notify(cond)
00441         # A second time, to check internal state is still ok.
00442         self._check_notify(cond)

Here is the call graph for this function:

Definition at line 443 of file lock_tests.py.

00443 
00444     def test_timeout(self):
00445         cond = self.condtype()
00446         results = []
00447         N = 5
00448         def f():
00449             cond.acquire()
00450             t1 = time.time()
00451             result = cond.wait(0.5)
00452             t2 = time.time()
00453             cond.release()
00454             results.append((t2 - t1, result))
00455         Bunch(f, N).wait_for_finished()
00456         self.assertEqual(len(results), N)
00457         for dt, result in results:
00458             self.assertTimeout(dt, 0.5)
00459             # Note that conceptually (that"s the condition variable protocol)
00460             # a wait() may succeed even if no one notifies us and before any
00461             # timeout occurs.  Spurious wakeups can occur.
00462             # This makes it hard to verify the result value.
00463             # In practice, this implementation has no spurious wakeups.
00464             self.assertFalse(result)

Here is the call graph for this function:

Definition at line 383 of file lock_tests.py.

00383 
00384     def test_unacquired_notify(self):
00385         cond = self.condtype()
00386         self.assertRaises(RuntimeError, cond.notify)

Here is the call graph for this function:

Definition at line 379 of file lock_tests.py.

00379 
00380     def test_unacquired_wait(self):
00381         cond = self.condtype()
00382         self.assertRaises(RuntimeError, cond.wait)

Here is the call graph for this function:

Definition at line 465 of file lock_tests.py.

00465 
00466     def test_waitfor(self):
00467         cond = self.condtype()
00468         state = 0
00469         def f():
00470             with cond:
00471                 result = cond.wait_for(lambda : state==4)
00472                 self.assertTrue(result)
00473                 self.assertEqual(state, 4)
00474         b = Bunch(f, 1)
00475         b.wait_for_started()
00476         for i in range(4):
00477             time.sleep(0.01)
00478             with cond:
00479                 state += 1
00480                 cond.notify()
00481         b.wait_for_finished()

Here is the call graph for this function:

Definition at line 482 of file lock_tests.py.

00482 
00483     def test_waitfor_timeout(self):
00484         cond = self.condtype()
00485         state = 0
00486         success = []
00487         def f():
00488             with cond:
00489                 dt = time.time()
00490                 result = cond.wait_for(lambda : state==4, timeout=0.1)
00491                 dt = time.time() - dt
00492                 self.assertFalse(result)
00493                 self.assertTimeout(dt, 0.1)
00494                 success.append(None)
00495         b = Bunch(f, 1)
00496         b.wait_for_started()
00497         # Only increment 3 times, so state == 4 is never reached.
00498         for i in range(3):
00499             time.sleep(0.01)
00500             with cond:
00501                 state += 1
00502                 cond.notify()
00503         b.wait_for_finished()
00504         self.assertEqual(len(success), 1)
00505 

Here is the call graph for this function:


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