Function test_non_blocking_behavior

tests/test_read_write.py:340–368

Test that non-blocking parameter works correctly. This test directly attempts to acquire a read lock in non-blocking mode when a write lock is already held by another process.

(lock_file: str)

@pytest.mark.timeout(10)
def test_non_blocking_behavior(lock_file: str) -> None:
    """Test that non-blocking parameter works correctly.

    This test directly attempts to acquire a read lock in non-blocking mode when a write lock is already held by another
    process.

    """
    write_acquired = Event()
    release_write = Event()

    writer = Process(target=acquire_lock, args=(lock_file, "write", write_acquired, release_write))

    with cleanup_processes([writer]):
        writer.start()
        assert write_acquired.wait(timeout=2), "Write lock not acquired"

        lock = ReadWriteLock(lock_file)

        start_time = time.time()

        with pytest.raises(Timeout):
            lock.acquire_read(blocking=False)

        elapsed = time.time() - start_time

        assert elapsed < 0.1, f"Non-blocking took too long: {elapsed}s"

        release_write.set()
        writer.join(timeout=2)


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