(self)
| 836 | class UnixWritePipeTransportTests(test_utils.TestCase): |
| 837 | |
| 838 | def setUp(self): |
| 839 | super().setUp() |
| 840 | self.loop = self.new_test_loop() |
| 841 | self.protocol = test_utils.make_test_protocol(asyncio.BaseProtocol) |
| 842 | self.pipe = mock.Mock(spec_set=io.RawIOBase) |
| 843 | self.pipe.fileno.return_value = 5 |
| 844 | |
| 845 | blocking_patcher = mock.patch('os.set_blocking') |
| 846 | blocking_patcher.start() |
| 847 | self.addCleanup(blocking_patcher.stop) |
| 848 | |
| 849 | fstat_patcher = mock.patch('os.fstat') |
| 850 | m_fstat = fstat_patcher.start() |
| 851 | st = mock.Mock() |
| 852 | st.st_mode = stat.S_IFSOCK |
| 853 | m_fstat.return_value = st |
| 854 | self.addCleanup(fstat_patcher.stop) |
| 855 | |
| 856 | def write_pipe_transport(self, waiter=None): |
| 857 | transport = unix_events._UnixWritePipeTransport(self.loop, self.pipe, |
nothing calls this directly
no test coverage detected