(self, lock_func, lock_name)
| 507 | |
| 508 | class FCNTLEINTRTest(EINTRBaseTest): |
| 509 | def _lock(self, lock_func, lock_name): |
| 510 | self.addCleanup(os_helper.unlink, os_helper.TESTFN) |
| 511 | rd1, wr1 = os.pipe() |
| 512 | rd2, wr2 = os.pipe() |
| 513 | for fd in (rd1, wr1, rd2, wr2): |
| 514 | self.addCleanup(os.close, fd) |
| 515 | code = textwrap.dedent(f""" |
| 516 | import fcntl, os, time |
| 517 | with open('{os_helper.TESTFN}', 'wb') as f: |
| 518 | fcntl.{lock_name}(f, fcntl.LOCK_EX) |
| 519 | os.write({wr1}, b"ok") |
| 520 | _ = os.read({rd2}, 2) # wait for parent process |
| 521 | time.sleep({self.sleep_time}) |
| 522 | """) |
| 523 | proc = self.subprocess(code, pass_fds=[wr1, rd2]) |
| 524 | with kill_on_error(proc): |
| 525 | with open(os_helper.TESTFN, 'wb') as f: |
| 526 | # synchronize the subprocess |
| 527 | ok = os.read(rd1, 2) |
| 528 | self.assertEqual(ok, b"ok") |
| 529 | |
| 530 | # notify the child that the parent is ready |
| 531 | start_time = time.monotonic() |
| 532 | os.write(wr2, b"go") |
| 533 | |
| 534 | # the child locked the file just a moment ago for 'sleep_time' seconds |
| 535 | # that means that the lock below will block for 'sleep_time' minus some |
| 536 | # potential context switch delay |
| 537 | lock_func(f, fcntl.LOCK_EX) |
| 538 | dt = time.monotonic() - start_time |
| 539 | self.stop_alarm() |
| 540 | self.check_elapsed_time(dt) |
| 541 | proc.wait() |
| 542 | |
| 543 | # Issue 35633: See https://bugs.python.org/issue35633#msg333662 |
| 544 | # skip test rather than accept PermissionError from all platforms |
no test coverage detected