(lock_file: str, success_flag: Synchronized[int])
| 408 | |
| 409 | |
| 410 | def recursive_write_lock(lock_file: str, success_flag: Synchronized[int]) -> None: |
| 411 | lock = ReadWriteLock(lock_file) |
| 412 | with lock.write_lock(): |
| 413 | assert lock._lock_level == 1 |
| 414 | assert lock._current_mode == "write" |
| 415 | |
| 416 | with lock.write_lock(): |
| 417 | assert lock._lock_level == 2 |
| 418 | assert lock._current_mode == "write" |
| 419 | |
| 420 | with lock.write_lock(): |
| 421 | assert lock._lock_level == 3 |
| 422 | assert lock._current_mode == "write" |
| 423 | |
| 424 | assert lock._lock_level == 2 |
| 425 | assert lock._current_mode == "write" |
| 426 | |
| 427 | assert lock._lock_level == 1 |
| 428 | assert lock._current_mode == "write" |
| 429 | |
| 430 | assert lock._lock_level == 0 |
| 431 | assert lock._current_mode is None |
| 432 | |
| 433 | success_flag.value = 1 |
| 434 | |
| 435 | |
| 436 | @pytest.mark.timeout(10) |
nothing calls this directly
no test coverage detected
searching dependent graphs…