(self)
| 1688 | @unittest.skipUnless(sys.platform != 'win32', |
| 1689 | "Don't support pipes for Windows") |
| 1690 | def test_write_pipe(self): |
| 1691 | rpipe, wpipe = os.pipe() |
| 1692 | pipeobj = io.open(wpipe, 'wb', 1024) |
| 1693 | |
| 1694 | proto = MyWritePipeProto(loop=self.loop) |
| 1695 | connect = self.loop.connect_write_pipe(lambda: proto, pipeobj) |
| 1696 | transport, p = self.loop.run_until_complete(connect) |
| 1697 | self.assertIs(p, proto) |
| 1698 | self.assertIs(transport, proto.transport) |
| 1699 | self.assertEqual('CONNECTED', proto.state) |
| 1700 | |
| 1701 | transport.write(b'1') |
| 1702 | |
| 1703 | data = bytearray() |
| 1704 | def reader(data): |
| 1705 | chunk = os.read(rpipe, 1024) |
| 1706 | data += chunk |
| 1707 | return len(data) |
| 1708 | |
| 1709 | test_utils.run_until(self.loop, lambda: reader(data) >= 1) |
| 1710 | self.assertEqual(b'1', data) |
| 1711 | |
| 1712 | transport.write(b'2345') |
| 1713 | test_utils.run_until(self.loop, lambda: reader(data) >= 5) |
| 1714 | self.assertEqual(b'12345', data) |
| 1715 | self.assertEqual('CONNECTED', proto.state) |
| 1716 | |
| 1717 | os.close(rpipe) |
| 1718 | |
| 1719 | # extra info is available |
| 1720 | self.assertIsNotNone(proto.transport.get_extra_info('pipe')) |
| 1721 | |
| 1722 | # close connection |
| 1723 | proto.transport.close() |
| 1724 | self.loop.run_until_complete(proto.done) |
| 1725 | self.assertEqual('CLOSED', proto.state) |
| 1726 | |
| 1727 | @unittest.skipUnless(sys.platform != 'win32', |
| 1728 | "Don't support pipes for Windows") |
nothing calls this directly
no test coverage detected