| 466 | |
| 467 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 468 | def test_parent_process(self): |
| 469 | if self.TYPE == "threads": |
| 470 | self.skipTest('test not appropriate for {}'.format(self.TYPE)) |
| 471 | |
| 472 | # Launch a child process. Make it launch a grandchild process. Kill the |
| 473 | # child process and make sure that the grandchild notices the death of |
| 474 | # its parent (a.k.a the child process). |
| 475 | rconn, wconn = self.Pipe(duplex=False) |
| 476 | p = self.Process( |
| 477 | target=self._test_create_grandchild_process, args=(wconn, )) |
| 478 | p.start() |
| 479 | |
| 480 | if not rconn.poll(timeout=support.LONG_TIMEOUT): |
| 481 | raise AssertionError("Could not communicate with child process") |
| 482 | parent_process_status = rconn.recv() |
| 483 | self.assertEqual(parent_process_status, "alive") |
| 484 | |
| 485 | p.terminate() |
| 486 | p.join() |
| 487 | |
| 488 | if not rconn.poll(timeout=support.LONG_TIMEOUT): |
| 489 | raise AssertionError("Could not communicate with child process") |
| 490 | parent_process_status = rconn.recv() |
| 491 | self.assertEqual(parent_process_status, "not alive") |
| 492 | |
| 493 | @classmethod |
| 494 | def _test_create_grandchild_process(cls, wconn): |