Function test_recursive_write_lock_acquisition

tests/test_read_write.py:437–446

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

(lock_file: str)

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

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

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


def acquire_write_lock_and_crash(lock_file: str, acquired_event: EventType) -> None: