Function test_non_blocking_behavior

tests/test_read_write.py:268–288

(lock_file: str)

@pytest.mark.timeout(_PROCESS_DEADLINE * 3)
def test_non_blocking_behavior(lock_file: str) -> None:
    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=_PROCESS_DEADLINE), "Write lock not acquired"

        start_time = time.time()

        with pytest.raises(Timeout):
            ReadWriteLock(lock_file).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=_PROCESS_DEADLINE)


@pytest.mark.parametrize(