(lock_file: str)
| 41 | |
| 42 | @pytest.mark.timeout(_PROCESS_DEADLINE * 5) |
| 43 | def test_read_locks_are_shared(lock_file: str) -> None: |
| 44 | read1_acquired = Event() |
| 45 | read2_acquired = Event() |
| 46 | |
| 47 | reader1 = Process(target=acquire_lock, args=(lock_file, "read", read1_acquired)) |
| 48 | reader2 = Process(target=acquire_lock, args=(lock_file, "read", read2_acquired)) |
| 49 | |
| 50 | with cleanup_processes([reader1, reader2]): |
| 51 | reader1.start() |
| 52 | time.sleep(0.5) # let reader1 acquire before reader2 starts |
| 53 | reader2.start() |
| 54 | |
| 55 | assert read1_acquired.wait(timeout=_PROCESS_DEADLINE), f"First read lock not acquired on {lock_file}" |
| 56 | assert read2_acquired.wait(timeout=_PROCESS_DEADLINE), f"Second read lock not acquired on {lock_file}" |
| 57 | |
| 58 | reader1.join(timeout=_PROCESS_DEADLINE) |
| 59 | reader2.join(timeout=_PROCESS_DEADLINE) |
| 60 | assert not reader1.is_alive(), "Reader 1 did not exit cleanly" |
| 61 | assert not reader2.is_alive(), "Reader 2 did not exit cleanly" |
| 62 | |
| 63 | |
| 64 | @pytest.mark.timeout(_PROCESS_DEADLINE * 5) |
nothing calls this directly
no test coverage detected