Method pipe

Lib/test/test_threading.py:1532–1538

(self)

class SubinterpThreadingTests(BaseTestCase):
    def pipe(self):
        r, w = os.pipe()
        self.addCleanup(os.close, r)
        self.addCleanup(os.close, w)
        if hasattr(os, 'set_blocking'):
            os.set_blocking(r, False)
        return (r, w)

    @unittest.expectedFailure # TODO: RUSTPYTHON
    def test_threads_join(self):