(self)
| 3927 | p.join() |
| 3928 | |
| 3929 | def test_context(self): |
| 3930 | a, b = self.Pipe() |
| 3931 | |
| 3932 | with a, b: |
| 3933 | a.send(1729) |
| 3934 | self.assertEqual(b.recv(), 1729) |
| 3935 | if self.TYPE == 'processes': |
| 3936 | self.assertFalse(a.closed) |
| 3937 | self.assertFalse(b.closed) |
| 3938 | |
| 3939 | if self.TYPE == 'processes': |
| 3940 | self.assertTrue(a.closed) |
| 3941 | self.assertTrue(b.closed) |
| 3942 | self.assertRaises(OSError, a.recv) |
| 3943 | self.assertRaises(OSError, b.recv) |
| 3944 | |
| 3945 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 3946 | def test_wait_empty(self): |
nothing calls this directly
no test coverage detected