Function test_custom_executor

tests/test_async_read_write.py:214–220

(lock_file: str)

@pytest.mark.asyncio
async def test_custom_executor(lock_file: str) -> None:
    executor = ThreadPoolExecutor(max_workers=1)
    lock = AsyncReadWriteLock(lock_file, is_singleton=False, executor=executor)
    async with lock.read_lock():
        assert lock._lock._current_mode == "read"
    assert lock._lock._lock_level == 0
    executor.shutdown(wait=False)


@pytest.mark.asyncio