(lock_file: str, success_flag: Synchronized[int])
| 369 | |
| 370 | |
| 371 | def recursive_read_lock(lock_file: str, success_flag: Synchronized[int]) -> None: |
| 372 | lock = ReadWriteLock(lock_file) |
| 373 | with lock.read_lock(): |
| 374 | assert lock._lock_level == 1 |
| 375 | assert lock._current_mode == "read" |
| 376 | |
| 377 | with lock.read_lock(): |
| 378 | assert lock._lock_level == 2 |
| 379 | assert lock._current_mode == "read" |
| 380 | |
| 381 | with lock.read_lock(): |
| 382 | assert lock._lock_level == 3 |
| 383 | assert lock._current_mode == "read" |
| 384 | |
| 385 | assert lock._lock_level == 2 |
| 386 | assert lock._current_mode == "read" |
| 387 | |
| 388 | assert lock._lock_level == 1 |
| 389 | assert lock._current_mode == "read" |
| 390 | |
| 391 | assert lock._lock_level == 0 |
| 392 | assert lock._current_mode is None |
| 393 | |
| 394 | success_flag.value = 1 |
| 395 | |
| 396 | |
| 397 | @pytest.mark.timeout(10) |
nothing calls this directly
no test coverage detected
searching dependent graphs…