(
lock_file: str,
mode: Literal["read", "write"],
acquired_event: EventType,
release_event: EventType | None = None,
timeout: float = -1,
blocking: bool = True,
ready_event: EventType | None = None,
contending_event: EventType | None = None,
)
| 514 | |
| 515 | |
| 516 | def acquire_lock( |
| 517 | lock_file: str, |
| 518 | mode: Literal["read", "write"], |
| 519 | acquired_event: EventType, |
| 520 | release_event: EventType | None = None, |
| 521 | timeout: float = -1, |
| 522 | blocking: bool = True, |
| 523 | ready_event: EventType | None = None, |
| 524 | contending_event: EventType | None = None, |
| 525 | ) -> None: |
| 526 | if ready_event: |
| 527 | ready_event.wait(timeout=_PROCESS_DEADLINE) |
| 528 | |
| 529 | lock = ReadWriteLock(lock_file, timeout=timeout, blocking=blocking) |
| 530 | if contending_event: |
| 531 | contending_event.set() # interpreter start-up is over, so a caller can time the contention itself |
| 532 | with lock.read_lock() if mode == "read" else lock.write_lock(): |
| 533 | acquired_event.set() |
| 534 | if release_event: |
| 535 | release_event.wait(timeout=_PROCESS_DEADLINE) |
| 536 | else: |
| 537 | time.sleep(0.5) # hold briefly to simulate work |
| 538 | |
| 539 | |
| 540 | def test_cleanup_reports_the_failure_that_escaped_before_a_start() -> None: |
nothing calls this directly
no test coverage detected