(self, do_open_close_reader, do_open_close_writer)
| 336 | @support.requires_freebsd_version(10, 3) |
| 337 | @unittest.skipUnless(hasattr(os, 'mkfifo'), 'needs mkfifo()') |
| 338 | def _test_open(self, do_open_close_reader, do_open_close_writer): |
| 339 | filename = os_helper.TESTFN |
| 340 | |
| 341 | # Use a fifo: until the child opens it for reading, the parent will |
| 342 | # block when trying to open it for writing. |
| 343 | os_helper.unlink(filename) |
| 344 | try: |
| 345 | os.mkfifo(filename) |
| 346 | except PermissionError as exc: |
| 347 | self.skipTest(f'os.mkfifo(): {exc!r}') |
| 348 | self.addCleanup(os_helper.unlink, filename) |
| 349 | |
| 350 | code = '\n'.join(( |
| 351 | 'import os, time', |
| 352 | '', |
| 353 | f'path = {filename!a}', |
| 354 | f'sleep_time = {self.sleep_time!r}', |
| 355 | '', |
| 356 | '# let the parent block', |
| 357 | 'time.sleep(sleep_time)', |
| 358 | '', |
| 359 | do_open_close_reader, |
| 360 | )) |
| 361 | |
| 362 | proc = self.subprocess(code) |
| 363 | with kill_on_error(proc): |
| 364 | do_open_close_writer(filename) |
| 365 | self.assertEqual(proc.wait(), 0) |
| 366 | |
| 367 | def python_open(self, path): |
| 368 | fp = open(path, 'w') |
no test coverage detected