Test that multiple processes can acquire read locks simultaneously.
(lock_file: str)
| 59 | @pytest.mark.timeout(20) |
| 60 | @pytest.mark.timeout(20) |
| 61 | def test_read_locks_are_shared(lock_file: str) -> None: |
| 62 | """Test that multiple processes can acquire read locks simultaneously.""" |
| 63 | read1_acquired = Event() |
| 64 | read2_acquired = Event() |
| 65 | |
| 66 | reader1 = Process(target=acquire_lock, args=(lock_file, "read", read1_acquired)) |
| 67 | reader2 = Process(target=acquire_lock, args=(lock_file, "read", read2_acquired)) |
| 68 | |
| 69 | with cleanup_processes([reader1, reader2]): |
| 70 | reader1.start() |
| 71 | time.sleep(0.5) # give reader1 time to acquire lock before starting reader2 |
| 72 | reader2.start() |
| 73 | |
| 74 | assert read1_acquired.wait(timeout=10), f"First read lock not acquired on {lock_file}" |
| 75 | assert read2_acquired.wait(timeout=10), f"Second read lock not acquired on {lock_file}" |
| 76 | |
| 77 | reader1.join(timeout=10) |
| 78 | reader2.join(timeout=10) |
| 79 | assert not reader1.is_alive(), "Reader 1 did not exit cleanly" |
| 80 | assert not reader2.is_alive(), "Reader 2 did not exit cleanly" |
| 81 | |
| 82 | |
| 83 | @pytest.mark.timeout(20) |
nothing calls this directly
no test coverage detected
searching dependent graphs…