(self)
| 6431 | |
| 6432 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 6433 | def test_empty(self): |
| 6434 | queue = multiprocessing.SimpleQueue() |
| 6435 | child_can_start = multiprocessing.Event() |
| 6436 | parent_can_continue = multiprocessing.Event() |
| 6437 | |
| 6438 | proc = multiprocessing.Process( |
| 6439 | target=self._test_empty, |
| 6440 | args=(queue, child_can_start, parent_can_continue) |
| 6441 | ) |
| 6442 | proc.daemon = True |
| 6443 | proc.start() |
| 6444 | |
| 6445 | self.assertTrue(queue.empty()) |
| 6446 | |
| 6447 | child_can_start.set() |
| 6448 | parent_can_continue.wait() |
| 6449 | |
| 6450 | self.assertFalse(queue.empty()) |
| 6451 | self.assertEqual(queue.get(), True) |
| 6452 | self.assertEqual(queue.get(), False) |
| 6453 | self.assertTrue(queue.empty()) |
| 6454 | |
| 6455 | proc.join() |
| 6456 | |
| 6457 | def test_close(self): |
| 6458 | queue = multiprocessing.SimpleQueue() |
nothing calls this directly
no test coverage detected