()
| 174 | write_finished = threading.Event() |
| 175 | |
| 176 | def writer(): |
| 177 | os.write(wfd, b'abc') |
| 178 | self.assertTrue(write_suspended.wait(support.SHORT_TIMEOUT)) |
| 179 | os.write(wfd, b'def') |
| 180 | write_finished.set() |
| 181 | |
| 182 | with threading_helper.start_threads([threading.Thread(target=writer)]): |
| 183 | self.assertEqual(os.read(rfd, 3), b'abc') |
no test coverage detected