Back to index

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

List of all members.

Public Member Functions

def test_current
def test_process
def test_terminate
def test_cpu_count
def test_active_children
def test_recursion
def assertTimingAlmostEqual
def assertReturnsIfImplemented
def __reduce__

Public Attributes

 TYPE
_PyObject_HEAD_EXTRA Py_ssize_t ob_refcnt
struct _typeobjectob_type

Static Public Attributes

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

Private Member Functions

def _test
def _test_terminate
def _test_recursion

Detailed Description

Definition at line 149 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._TestProcess._test (   cls,
  q,
  args,
  kwds 
) [private]

Definition at line 168 of file test_multiprocessing.py.

00168 
00169     def _test(cls, q, *args, **kwds):
00170         current = cls.current_process()
00171         q.put(args)
00172         q.put(kwds)
00173         q.put(current.name)
00174         if cls.TYPE != 'threads':
00175             q.put(bytes(current.authkey))
00176             q.put(current.pid)

Here is the caller graph for this function:

def test.test_multiprocessing._TestProcess._test_recursion (   cls,
  wconn,
  id 
) [private]

Definition at line 267 of file test_multiprocessing.py.

00267 
00268     def _test_recursion(cls, wconn, id):
00269         from multiprocessing import forking
00270         wconn.send(id)
00271         if len(id) < 2:
00272             for i in range(2):
00273                 p = cls.Process(
00274                     target=cls._test_recursion, args=(wconn, id+[i])
00275                     )
00276                 p.start()
00277                 p.join()

Here is the caller graph for this function:

Definition at line 217 of file test_multiprocessing.py.

00217 
00218     def _test_terminate(cls):
00219         time.sleep(1000)

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 254 of file test_multiprocessing.py.

00254 
00255     def test_active_children(self):
00256         self.assertEqual(type(self.active_children()), list)
00257 
00258         p = self.Process(target=time.sleep, args=(DELTA,))
00259         self.assertNotIn(p, self.active_children())
00260 
00261         p.start()
00262         self.assertIn(p, self.active_children())
00263 
00264         p.join()
00265         self.assertNotIn(p, self.active_children())

Here is the call graph for this function:

Definition at line 246 of file test_multiprocessing.py.

00246 
00247     def test_cpu_count(self):
00248         try:
00249             cpus = multiprocessing.cpu_count()
00250         except NotImplementedError:
00251             cpus = 1
00252         self.assertTrue(type(cpus) is int)
00253         self.assertTrue(cpus >= 1)

Here is the call graph for this function:

Definition at line 153 of file test_multiprocessing.py.

00153 
00154     def test_current(self):
00155         if self.TYPE == 'threads':
00156             return
00157 
00158         current = self.current_process()
00159         authkey = current.authkey
00160 
00161         self.assertTrue(current.is_alive())
00162         self.assertTrue(not current.daemon)
00163         self.assertIsInstance(authkey, bytes)
00164         self.assertTrue(len(authkey) > 0)
00165         self.assertEqual(current.ident, os.getpid())
00166         self.assertEqual(current.exitcode, None)

Definition at line 177 of file test_multiprocessing.py.

00177 
00178     def test_process(self):
00179         q = self.Queue(1)
00180         e = self.Event()
00181         args = (q, 1, 2)
00182         kwargs = {'hello':23, 'bye':2.54}
00183         name = 'SomeProcess'
00184         p = self.Process(
00185             target=self._test, args=args, kwargs=kwargs, name=name
00186             )
00187         p.daemon = True
00188         current = self.current_process()
00189 
00190         if self.TYPE != 'threads':
00191             self.assertEqual(p.authkey, current.authkey)
00192         self.assertEqual(p.is_alive(), False)
00193         self.assertEqual(p.daemon, True)
00194         self.assertNotIn(p, self.active_children())
00195         self.assertTrue(type(self.active_children()) is list)
00196         self.assertEqual(p.exitcode, None)
00197 
00198         p.start()
00199 
00200         self.assertEqual(p.exitcode, None)
00201         self.assertEqual(p.is_alive(), True)
00202         self.assertIn(p, self.active_children())
00203 
00204         self.assertEqual(q.get(), args[1:])
00205         self.assertEqual(q.get(), kwargs)
00206         self.assertEqual(q.get(), p.name)
00207         if self.TYPE != 'threads':
00208             self.assertEqual(q.get(), current.authkey)
00209             self.assertEqual(q.get(), p.pid)
00210 
00211         p.join()
00212 
00213         self.assertEqual(p.exitcode, 0)
00214         self.assertEqual(p.is_alive(), False)
00215         self.assertNotIn(p, self.active_children())

Here is the call graph for this function:

Definition at line 278 of file test_multiprocessing.py.

00278 
00279     def test_recursion(self):
00280         rconn, wconn = self.Pipe(duplex=False)
00281         self._test_recursion(wconn, [])
00282 
00283         time.sleep(DELTA)
00284         result = []
00285         while rconn.poll():
00286             result.append(rconn.recv())
00287 
00288         expected = [
00289             [],
00290               [0],
00291                 [0, 0],
00292                 [0, 1],
00293               [1],
00294                 [1, 0],
00295                 [1, 1]
00296             ]
00297         self.assertEqual(result, expected)
00298 
00299 #
00300 #
00301 #

Here is the call graph for this function:

Definition at line 220 of file test_multiprocessing.py.

00220 
00221     def test_terminate(self):
00222         if self.TYPE == 'threads':
00223             return
00224 
00225         p = self.Process(target=self._test_terminate)
00226         p.daemon = True
00227         p.start()
00228 
00229         self.assertEqual(p.is_alive(), True)
00230         self.assertIn(p, self.active_children())
00231         self.assertEqual(p.exitcode, None)
00232 
00233         p.terminate()
00234 
00235         join = TimingWrapper(p.join)
00236         self.assertEqual(join(), None)
00237         self.assertTimingAlmostEqual(join.elapsed, 0.0)
00238 
00239         self.assertEqual(p.is_alive(), False)
00240         self.assertNotIn(p, self.active_children())
00241 
00242         p.join()
00243 
00244         # XXX sometimes get p.exitcode == 0 on Windows ...
00245         #self.assertEqual(p.exitcode, -signal.SIGTERM)

Here is the call graph for this function:


Member Data Documentation

tuple test.test_multiprocessing._TestProcess.ALLOWED_TYPES = ('processes', 'threads') [static]

Reimplemented from test.test_multiprocessing.BaseTestCase.

Definition at line 151 of file test_multiprocessing.py.

Definition at line 107 of file object.h.

struct _typeobject* _object::ob_type [inherited]

Definition at line 108 of file object.h.

Definition at line 154 of file test_multiprocessing.py.


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