(self)
| 3478 | p.join() |
| 3479 | |
| 3480 | def test_duplex_false(self): |
| 3481 | reader, writer = self.Pipe(duplex=False) |
| 3482 | self.assertEqual(writer.send(1), None) |
| 3483 | self.assertEqual(reader.recv(), 1) |
| 3484 | if self.TYPE == 'processes': |
| 3485 | self.assertEqual(reader.readable, True) |
| 3486 | self.assertEqual(reader.writable, False) |
| 3487 | self.assertEqual(writer.readable, False) |
| 3488 | self.assertEqual(writer.writable, True) |
| 3489 | self.assertRaises(OSError, reader.send, 2) |
| 3490 | self.assertRaises(OSError, writer.recv) |
| 3491 | self.assertRaises(OSError, writer.poll) |
| 3492 | |
| 3493 | def test_spawn_close(self): |
| 3494 | # We test that a pipe connection can be closed by parent |
nothing calls this directly
no test coverage detected