Back to index

python3.2  3.2.2
Public Member Functions | Static Public Attributes
test.test_threading.ConditionTests Class Reference
Inheritance diagram for test.test_threading.ConditionTests:
Inheritance graph
[legend]
Collaboration diagram for test.test_threading.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

Static Public Attributes

tuple condtype = staticmethod(threading.Condition)

Detailed Description

Definition at line 739 of file test_threading.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:

def test.lock_tests.ConditionTests.test_acquire (   self) [inherited]

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:

def test.lock_tests.ConditionTests.test_notify (   self) [inherited]

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:

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

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:

def test.lock_tests.ConditionTests.test_waitfor (   self) [inherited]

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:


Member Data Documentation

Definition at line 740 of file test_threading.py.


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