Function test_recursive_read_lock_acquisition

tests/test_read_write.py:398–407

Test that the same process can acquire the same read lock multiple times.

(lock_file: str)

@pytest.mark.timeout(10)
def test_recursive_read_lock_acquisition(lock_file: str) -> None:
    """Test that the same process can acquire the same read lock multiple times."""
    success = Value("i", 0)
    worker = Process(target=recursive_read_lock, args=(lock_file, success))

    with cleanup_processes([worker]):
        worker.start()
        worker.join(timeout=5)

    assert success.value == 1, "Recursive read lock acquisition failed"


def recursive_write_lock(lock_file: str, success_flag: Synchronized[int]) -> None: