| 408 | wconn.send([parent_process().pid, parent_process().name]) |
| 409 | |
| 410 | def test_parent_process(self): |
| 411 | if self.TYPE == "threads": |
| 412 | self.skipTest('test not appropriate for {}'.format(self.TYPE)) |
| 413 | |
| 414 | # Launch a child process. Make it launch a grandchild process. Kill the |
| 415 | # child process and make sure that the grandchild notices the death of |
| 416 | # its parent (a.k.a the child process). |
| 417 | rconn, wconn = self.Pipe(duplex=False) |
| 418 | p = self.Process( |
| 419 | target=self._test_create_grandchild_process, args=(wconn, )) |
| 420 | p.start() |
| 421 | |
| 422 | if not rconn.poll(timeout=support.LONG_TIMEOUT): |
| 423 | raise AssertionError("Could not communicate with child process") |
| 424 | parent_process_status = rconn.recv() |
| 425 | self.assertEqual(parent_process_status, "alive") |
| 426 | |
| 427 | p.terminate() |
| 428 | p.join() |
| 429 | |
| 430 | if not rconn.poll(timeout=support.LONG_TIMEOUT): |
| 431 | raise AssertionError("Could not communicate with child process") |
| 432 | parent_process_status = rconn.recv() |
| 433 | self.assertEqual(parent_process_status, "not alive") |
| 434 | |
| 435 | @classmethod |
| 436 | def _test_create_grandchild_process(cls, wconn): |