(lock_file: str, mode: Literal["read", "write"], success_flag: Synchronized[int])
| 603 | |
| 604 | |
| 605 | def recursive_lock(lock_file: str, mode: Literal["read", "write"], success_flag: Synchronized[int]) -> None: |
| 606 | lock = ReadWriteLock(lock_file) |
| 607 | acquire = lock.read_lock if mode == "read" else lock.write_lock |
| 608 | with acquire(): |
| 609 | assert lock._lock_level == 1 |
| 610 | assert lock._current_mode == mode |
| 611 | |
| 612 | with acquire(): |
| 613 | assert lock._lock_level == 2 |
| 614 | assert lock._current_mode == mode |
| 615 | |
| 616 | with acquire(): |
| 617 | assert lock._lock_level == 3 |
| 618 | assert lock._current_mode == mode |
| 619 | |
| 620 | assert lock._lock_level == 2 |
| 621 | assert lock._current_mode == mode |
| 622 | |
| 623 | assert lock._lock_level == 1 |
| 624 | assert lock._current_mode == mode |
| 625 | |
| 626 | assert lock._lock_level == 0 |
| 627 | assert lock._current_mode is None |
| 628 | |
| 629 | success_flag.value = 1 |
| 630 | |
| 631 | |
| 632 | def acquire_lock_and_crash(lock_file: str, mode: Literal["read", "write"], acquired_event: EventType) -> None: |
nothing calls this directly
no test coverage detected