Back to index

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

List of all members.

Public Member Functions

def test_release_unacquired
def test_constructor
def test_acquire
def test_acquire_destroy
def test_acquire_contended
def test_try_acquire
def test_try_acquire_contended
def test_acquire_timeout
def test_default_value
def test_with
def setUp
def tearDown
def assertTimeout

Detailed Description

Tests for unbounded semaphores.

Definition at line 630 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 515 of file lock_tests.py.

00515 
00516     def test_acquire(self):
00517         sem = self.semtype(1)
00518         sem.acquire()
00519         sem.release()
00520         sem = self.semtype(2)
00521         sem.acquire()
00522         sem.acquire()
00523         sem.release()
00524         sem.release()

Definition at line 530 of file lock_tests.py.

00530 
00531     def test_acquire_contended(self):
00532         sem = self.semtype(7)
00533         sem.acquire()
00534         N = 10
00535         results1 = []
00536         results2 = []
00537         phase_num = 0
00538         def f():
00539             sem.acquire()
00540             results1.append(phase_num)
00541             sem.acquire()
00542             results2.append(phase_num)
00543         b = Bunch(f, 10)
00544         b.wait_for_started()
00545         while len(results1) + len(results2) < 6:
00546             _wait()
00547         self.assertEqual(results1 + results2, [0] * 6)
00548         phase_num = 1
00549         for i in range(7):
00550             sem.release()
00551         while len(results1) + len(results2) < 13:
00552             _wait()
00553         self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7)
00554         phase_num = 2
00555         for i in range(6):
00556             sem.release()
00557         while len(results1) + len(results2) < 19:
00558             _wait()
00559         self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7 + [2] * 6)
00560         # The semaphore is still locked
00561         self.assertFalse(sem.acquire(False))
00562         # Final release, to let the last thread finish
00563         sem.release()
00564         b.wait_for_finished()

Here is the call graph for this function:

Definition at line 525 of file lock_tests.py.

00525 
00526     def test_acquire_destroy(self):
00527         sem = self.semtype()
00528         sem.acquire()
00529         del sem

Definition at line 586 of file lock_tests.py.

00586 
00587     def test_acquire_timeout(self):
00588         sem = self.semtype(2)
00589         self.assertRaises(ValueError, sem.acquire, False, timeout=1.0)
00590         self.assertTrue(sem.acquire(timeout=0.005))
00591         self.assertTrue(sem.acquire(timeout=0.005))
00592         self.assertFalse(sem.acquire(timeout=0.005))
00593         sem.release()
00594         self.assertTrue(sem.acquire(timeout=0.005))
00595         t = time.time()
00596         self.assertFalse(sem.acquire(timeout=0.5))
00597         dt = time.time() - t
00598         self.assertTimeout(dt, 0.5)

Here is the call graph for this function:

Definition at line 511 of file lock_tests.py.

00511 
00512     def test_constructor(self):
00513         self.assertRaises(ValueError, self.semtype, value = -1)
00514         self.assertRaises(ValueError, self.semtype, value = -sys.maxsize)

Here is the call graph for this function:

Definition at line 599 of file lock_tests.py.

00599 
00600     def test_default_value(self):
00601         # The default initial value is 1.
00602         sem = self.semtype()
00603         sem.acquire()
00604         def f():
00605             sem.acquire()
00606             sem.release()
00607         b = Bunch(f, 1)
00608         b.wait_for_started()
00609         _wait()
00610         self.assertFalse(b.finished)
00611         sem.release()
00612         b.wait_for_finished()

Here is the call graph for this function:

Definition at line 635 of file lock_tests.py.

00635 
00636     def test_release_unacquired(self):
00637         # Unbounded releases are allowed and increment the semaphore's value
00638         sem = self.semtype(1)
00639         sem.release()
00640         sem.acquire()
00641         sem.acquire()
00642         sem.release()
00643 

Definition at line 565 of file lock_tests.py.

00565 
00566     def test_try_acquire(self):
00567         sem = self.semtype(2)
00568         self.assertTrue(sem.acquire(False))
00569         self.assertTrue(sem.acquire(False))
00570         self.assertFalse(sem.acquire(False))
00571         sem.release()
00572         self.assertTrue(sem.acquire(False))

Here is the call graph for this function:

Definition at line 573 of file lock_tests.py.

00573 
00574     def test_try_acquire_contended(self):
00575         sem = self.semtype(4)
00576         sem.acquire()
00577         results = []
00578         def f():
00579             results.append(sem.acquire(False))
00580             results.append(sem.acquire(False))
00581         Bunch(f, 5).wait_for_finished()
00582         # There can be a thread switch between acquiring the semaphore and
00583         # appending the result, therefore results will not necessarily be
00584         # ordered.
00585         self.assertEqual(sorted(results), [False] * 7 + [True] *  3 )

Here is the call graph for this function:

Definition at line 613 of file lock_tests.py.

00613 
00614     def test_with(self):
00615         sem = self.semtype(2)
00616         def _with(err=None):
00617             with sem:
00618                 self.assertTrue(sem.acquire(False))
00619                 sem.release()
00620                 with sem:
00621                     self.assertFalse(sem.acquire(False))
00622                     if err:
00623                         raise err
00624         _with()
00625         self.assertTrue(sem.acquire(False))
00626         sem.release()
00627         self.assertRaises(TypeError, _with, TypeError)
00628         self.assertTrue(sem.acquire(False))
00629         sem.release()

Here is the call graph for this function:


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