import sys import pytest from lightrag.api.config import parse_args pytestmark = pytest.mark.offline ROLE_MAX_ASYNC_ENV_KEYS = ( "MAX_ASYNC", "EXTRACT_MAX_ASYNC_LLM", "KEYWORD_MAX_ASYNC_LLM", "QUERY_MAX_ASYNC_LLM", "VLM_MAX_ASYNC_LLM", ) def _clear_max_async_env(monkeypatch): for key in ROLE_MAX_ASYNC_ENV_KEYS: monkeypatch.delenv(key, raising=False) def test_role_max_async_defaults_none_when_env_unset(monkeypatch): _clear_max_async_env(monkeypatch) monkeypatch.setattr(sys, "argv", ["lightrag-server"]) monkeypatch.setenv("MAX_ASYNC", "10") args = parse_args() assert args.max_async == 10 assert args.extract_llm_max_async is None assert args.keyword_llm_max_async is None assert args.query_llm_max_async is None assert args.vlm_llm_max_async is None def test_role_max_async_env_override_keeps_other_roles_none(monkeypatch): _clear_max_async_env(monkeypatch) monkeypatch.setattr(sys, "argv", ["lightrag-server"]) monkeypatch.setenv("MAX_ASYNC", "10") monkeypatch.setenv("EXTRACT_MAX_ASYNC_LLM", "7") args = parse_args() assert args.max_async == 10 assert args.extract_llm_max_async == 7 assert args.keyword_llm_max_async is None assert args.query_llm_max_async is None assert args.vlm_llm_max_async is None def test_role_max_async_literal_none_string_is_preserved(monkeypatch): _clear_max_async_env(monkeypatch) monkeypatch.setattr(sys, "argv", ["lightrag-server"]) monkeypatch.setenv("MAX_ASYNC", "10") monkeypatch.setenv("QUERY_MAX_ASYNC_LLM", "None") args = parse_args() assert args.max_async == 10 assert args.query_llm_max_async is None