(self)
| 659 | class UnixReadPipeTransportTests(test_utils.TestCase): |
| 660 | |
| 661 | def setUp(self): |
| 662 | super().setUp() |
| 663 | self.loop = self.new_test_loop() |
| 664 | self.protocol = test_utils.make_test_protocol(asyncio.Protocol) |
| 665 | self.pipe = mock.Mock(spec_set=io.RawIOBase) |
| 666 | self.pipe.fileno.return_value = 5 |
| 667 | |
| 668 | blocking_patcher = mock.patch('os.set_blocking') |
| 669 | blocking_patcher.start() |
| 670 | self.addCleanup(blocking_patcher.stop) |
| 671 | |
| 672 | fstat_patcher = mock.patch('os.fstat') |
| 673 | m_fstat = fstat_patcher.start() |
| 674 | st = mock.Mock() |
| 675 | st.st_mode = stat.S_IFIFO |
| 676 | m_fstat.return_value = st |
| 677 | self.addCleanup(fstat_patcher.stop) |
| 678 | |
| 679 | def read_pipe_transport(self, waiter=None): |
| 680 | transport = unix_events._UnixReadPipeTransport(self.loop, self.pipe, |
nothing calls this directly
no test coverage detected