Back to index

python3.2  3.2.2
Public Member Functions | Public Attributes | Static Public Attributes | Private Member Functions
test.test_multiprocessing._TestSemaphore Class Reference
Inheritance diagram for test.test_multiprocessing._TestSemaphore:
Inheritance graph
[legend]
Collaboration diagram for test.test_multiprocessing._TestSemaphore:
Collaboration graph
[legend]

List of all members.

Public Member Functions

def test_semaphore
def test_bounded_semaphore
def test_timeout
def assertTimingAlmostEqual
def assertReturnsIfImplemented
def __reduce__

Public Attributes

_PyObject_HEAD_EXTRA Py_ssize_t ob_refcnt
struct _typeobjectob_type

Static Public Attributes

tuple ALLOWED_TYPES = ('processes', 'manager', 'threads')

Private Member Functions

def _test_semaphore

Detailed Description

Definition at line 589 of file test_multiprocessing.py.


Member Function Documentation

def test.test_multiprocessing.BaseTestCase.__reduce__ (   self,
  args 
) [inherited]

Definition at line 124 of file test_multiprocessing.py.

00124 
00125     def __reduce__(self, *args):
00126         raise NotImplementedError("shouldn't try to pickle a test case")

def test.test_multiprocessing._TestSemaphore._test_semaphore (   self,
  sem 
) [private]

Definition at line 591 of file test_multiprocessing.py.

00591 
00592     def _test_semaphore(self, sem):
00593         self.assertReturnsIfImplemented(2, get_value, sem)
00594         self.assertEqual(sem.acquire(), True)
00595         self.assertReturnsIfImplemented(1, get_value, sem)
00596         self.assertEqual(sem.acquire(), True)
00597         self.assertReturnsIfImplemented(0, get_value, sem)
00598         self.assertEqual(sem.acquire(False), False)
00599         self.assertReturnsIfImplemented(0, get_value, sem)
00600         self.assertEqual(sem.release(), None)
00601         self.assertReturnsIfImplemented(1, get_value, sem)
00602         self.assertEqual(sem.release(), None)
00603         self.assertReturnsIfImplemented(2, get_value, sem)

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_multiprocessing.BaseTestCase.assertReturnsIfImplemented (   self,
  value,
  func,
  args 
) [inherited]

Definition at line 114 of file test_multiprocessing.py.

00114 
00115     def assertReturnsIfImplemented(self, value, func, *args):
00116         try:
00117             res = func(*args)
00118         except NotImplementedError:
00119             pass
00120         else:
00121             return self.assertEqual(value, res)

Here is the call graph for this function:

Here is the caller graph for this function:

def test.test_multiprocessing.BaseTestCase.assertTimingAlmostEqual (   self,
  a,
  b 
) [inherited]

Definition at line 110 of file test_multiprocessing.py.

00110 
00111     def assertTimingAlmostEqual(self, a, b):
00112         if CHECK_TIMINGS:
00113             self.assertAlmostEqual(a, b, 1)

Here is the call graph for this function:

Here is the caller graph for this function:

Definition at line 612 of file test_multiprocessing.py.

00612 
00613     def test_bounded_semaphore(self):
00614         sem = self.BoundedSemaphore(2)
00615         self._test_semaphore(sem)
00616         # Currently fails on OS/X
00617         #if HAVE_GETVALUE:
00618         #    self.assertRaises(ValueError, sem.release)
00619         #    self.assertReturnsIfImplemented(2, get_value, sem)

Here is the call graph for this function:

Definition at line 604 of file test_multiprocessing.py.

00604 
00605     def test_semaphore(self):
00606         sem = self.Semaphore(2)
00607         self._test_semaphore(sem)
00608         self.assertEqual(sem.release(), None)
00609         self.assertReturnsIfImplemented(3, get_value, sem)
00610         self.assertEqual(sem.release(), None)
00611         self.assertReturnsIfImplemented(4, get_value, sem)

Here is the call graph for this function:

Definition at line 620 of file test_multiprocessing.py.

00620 
00621     def test_timeout(self):
00622         if self.TYPE != 'processes':
00623             return
00624 
00625         sem = self.Semaphore(0)
00626         acquire = TimingWrapper(sem.acquire)
00627 
00628         self.assertEqual(acquire(False), False)
00629         self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
00630 
00631         self.assertEqual(acquire(False, None), False)
00632         self.assertTimingAlmostEqual(acquire.elapsed, 0.0)
00633 
00634         self.assertEqual(acquire(False, TIMEOUT1), False)
00635         self.assertTimingAlmostEqual(acquire.elapsed, 0)
00636 
00637         self.assertEqual(acquire(True, TIMEOUT2), False)
00638         self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT2)
00639 
00640         self.assertEqual(acquire(timeout=TIMEOUT3), False)
00641         self.assertTimingAlmostEqual(acquire.elapsed, TIMEOUT3)
00642 

Here is the call graph for this function:


Member Data Documentation

tuple test.test_multiprocessing.BaseTestCase.ALLOWED_TYPES = ('processes', 'manager', 'threads') [static, inherited]

Definition at line 107 of file object.h.

struct _typeobject* _object::ob_type [inherited]

Definition at line 108 of file object.h.


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