(self)
| 3855 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 3856 | @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") |
| 3857 | def test_fd_transfer(self): |
| 3858 | if self.TYPE != 'processes': |
| 3859 | self.skipTest("only makes sense with processes") |
| 3860 | conn, child_conn = self.Pipe(duplex=True) |
| 3861 | |
| 3862 | p = self.Process(target=self._writefd, args=(child_conn, b"foo")) |
| 3863 | p.daemon = True |
| 3864 | p.start() |
| 3865 | self.addCleanup(os_helper.unlink, os_helper.TESTFN) |
| 3866 | with open(os_helper.TESTFN, "wb") as f: |
| 3867 | fd = f.fileno() |
| 3868 | if msvcrt: |
| 3869 | fd = msvcrt.get_osfhandle(fd) |
| 3870 | reduction.send_handle(conn, fd, p.pid) |
| 3871 | p.join() |
| 3872 | with open(os_helper.TESTFN, "rb") as f: |
| 3873 | self.assertEqual(f.read(), b"foo") |
| 3874 | |
| 3875 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 3876 | @unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction") |
nothing calls this directly
no test coverage detected