lightrag-cn-git-d5efd3/tests/setup/test_validate.py

# Regression tests for interactive setup wizard.
# Classification: keep tests here when they verify validate_* and related checks that accept or reject final env/security/runtime configuration values.

from __future__ import annotations

import subprocess
from pathlib import Path

import pytest

from tests.setup._helpers import (
    REPO_ROOT,
    parse_lines,
    run_bash,
    run_bash_lines,
    write_text_lines,
)

pytestmark = pytest.mark.offline


def test_validate_env_file_rejects_missing_ssl_files(tmp_path: Path) -> None:
    """Validation should fail when SSL is enabled with missing cert/key paths."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "SSL=true",
                "SSL_CERTFILE=/missing/cert.pem",
                "SSL_KEYFILE=/missing/key.pem",
                "LIGHTRAG_KV_STORAGE=JsonKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    result = subprocess.run(
        [
            "bash",
            "-lc",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
validate_env_file
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 1
    assert "Invalid SSL_CERTFILE" in result.stderr
    assert "Invalid SSL_KEYFILE" in result.stderr


def test_validate_env_file_rejects_container_ssl_paths_for_host_target(
    tmp_path: Path,
) -> None:
    """host-target .env must not accept /app/data/certs/* even when the staged file exists."""
    (tmp_path / "data" / "certs").mkdir(parents=True)
    (tmp_path / "data" / "certs" / "cert.pem").write_text("cert", encoding="utf-8")
    (tmp_path / "data" / "certs" / "key.pem").write_text("key", encoding="utf-8")
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "SSL=true",
                "SSL_CERTFILE=/app/data/certs/cert.pem",
                "SSL_KEYFILE=/app/data/certs/key.pem",
                "LIGHTRAG_RUNTIME_TARGET=host",
                "LIGHTRAG_KV_STORAGE=JsonKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    result = subprocess.run(
        [
            "bash",
            "-lc",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
validate_env_file
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 1
    assert "Invalid SSL_CERTFILE" in result.stderr
    assert "Invalid SSL_KEYFILE" in result.stderr


def test_validate_env_file_rejects_container_ssl_paths_for_default_host_target(
    tmp_path: Path,
) -> None:
    """Omitting LIGHTRAG_RUNTIME_TARGET defaults to host; container paths must still be rejected."""
    (tmp_path / "data" / "certs").mkdir(parents=True)
    (tmp_path / "data" / "certs" / "cert.pem").write_text("cert", encoding="utf-8")
    (tmp_path / "data" / "certs" / "key.pem").write_text("key", encoding="utf-8")
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "SSL=true",
                "SSL_CERTFILE=/app/data/certs/cert.pem",
                "SSL_KEYFILE=/app/data/certs/key.pem",
                "LIGHTRAG_KV_STORAGE=JsonKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    result = subprocess.run(
        [
            "bash",
            "-lc",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
validate_env_file
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 1
    assert "Invalid SSL_CERTFILE" in result.stderr
    assert "Invalid SSL_KEYFILE" in result.stderr


def test_validate_env_file_accepts_container_ssl_paths_for_compose_target(
    tmp_path: Path,
) -> None:
    """compose-target .env may use /app/data/certs/* when the staged files exist."""
    (tmp_path / "data" / "certs").mkdir(parents=True)
    (tmp_path / "data" / "certs" / "cert.pem").write_text("cert", encoding="utf-8")
    (tmp_path / "data" / "certs" / "key.pem").write_text("key", encoding="utf-8")
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "SSL=true",
                "SSL_CERTFILE=/app/data/certs/cert.pem",
                "SSL_KEYFILE=/app/data/certs/key.pem",
                "LIGHTRAG_RUNTIME_TARGET=compose",
                "LIGHTRAG_KV_STORAGE=JsonKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    result = subprocess.run(
        [
            "bash",
            "-lc",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
validate_env_file
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    assert result.returncode == 0


def test_validate_sensitive_env_literals_rejects_interpolation_syntax() -> None:
    """Sensitive values should reject `${...}` so default dotenv interpolation stays safe."""
    output = run_bash(f"""
set -euo pipefail
source "{REPO_ROOT}/scripts/setup/setup.sh"
reset_state

ENV_VALUES[TOKEN_SECRET]='${{JWT_SECRET}}'
ENV_VALUES[LIGHTRAG_API_KEY]='plain$token'
ENV_VALUES[WEBUI_DESCRIPTION]='${{ALLOWED_MACRO}}'

if validate_sensitive_env_literals; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""")
    values = parse_lines(output)
    assert values["VALID"] == "no"


def test_validate_env_file_allows_predictable_auth_passwords_and_leaves_them_to_audit(
    tmp_path: Path,
) -> None:
    """validate_env_file should allow risky-but-runnable auth settings."""
    write_text_lines(
        tmp_path / ".env",
        [
            "AUTH_ACCOUNTS=admin:Passw0rd!",
            "TOKEN_SECRET=jwt-secret",
            "LIGHTRAG_KV_STORAGE=JsonKVStorage",
            "LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage",
            "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
            "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
        ],
    )
    write_text_lines(tmp_path / "env.example", ["LLM_BINDING=openai"])
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "yes"
    audit_result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
security_check_env_file
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    assert audit_result.returncode == 1
    assert "AUTH_ACCOUNTS uses a predictable password prefix." in audit_result.stdout


def test_validate_uri_accepts_neo4j_self_signed_tls_scheme() -> None:
    """Neo4j self-signed TLS URIs should pass validation."""
    output = run_bash(f"""
set -euo pipefail
source "{REPO_ROOT}/scripts/setup/setup.sh"

if validate_uri "neo4j+ssc://db.example.com:7687" neo4j; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""")
    values = parse_lines(output)
    assert values["VALID"] == "yes"


def test_validate_security_config_rejects_malformed_auth_accounts() -> None:
    """Security validation should reject auth entries the API cannot parse."""
    output = run_bash(f"""
set -euo pipefail
source "{REPO_ROOT}/scripts/setup/setup.sh"
reset_state

if validate_security_config "admin" "token-secret" "" no "/health"; then
  printf 'MISSING_COLON=yes\\n'
else
  printf 'MISSING_COLON=no\\n'
fi

if validate_security_config "admin:secret," "token-secret" "" no "/health"; then
  printf 'TRAILING_COMMA=yes\\n'
else
  printf 'TRAILING_COMMA=no\\n'
fi

if validate_security_config "admin:secret,reader:hunter2" "token-secret" "" no "/health"; then
  printf 'VALID_FORMAT=yes\\n'
else
  printf 'VALID_FORMAT=no\\n'
fi

if validate_security_config 'admin:{{bcrypt}}$2b$12$abcdefghijklmnopqrstuuuuuuuuuuuuuuuuuuuuuuuuuuuu' "token-secret" "" no "/health"; then
  printf 'BCRYPT_FORMAT=yes\\n'
else
  printf 'BCRYPT_FORMAT=no\\n'
fi

if validate_security_config "admin:admin123!" "token-secret" "" no "/health"; then
  printf 'ADMIN_PREFIX=yes\\n'
else
  printf 'ADMIN_PREFIX=no\\n'
fi

if validate_security_config "admin:Passw0rd!" "token-secret" "" no "/health"; then
  printf 'PASS_PREFIX=yes\\n'
else
  printf 'PASS_PREFIX=no\\n'
fi
""")
    values = parse_lines(output)
    assert values["MISSING_COLON"] == "no"
    assert values["TRAILING_COMMA"] == "no"
    assert values["VALID_FORMAT"] == "yes"
    assert values["BCRYPT_FORMAT"] == "yes"
    assert values["ADMIN_PREFIX"] == "no"
    assert values["PASS_PREFIX"] == "no"


def test_validate_env_file_handles_supported_and_unsupported_uri_schemes(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject malformed schemes and allow supported TLS variants."""
    cases = {
        "invalid-neo4j-scheme": (
            [
                "LIGHTRAG_GRAPH_STORAGE=Neo4JStorage",
                "NEO4J_URI=http://localhost:7687",
                "NEO4J_USERNAME=neo4j",
                "NEO4J_PASSWORD=secret",
            ],
            "no",
            "Invalid NEO4J_URI",
        ),
        "invalid-redis-scheme": (
            ["LIGHTRAG_KV_STORAGE=RedisKVStorage", "REDIS_URI=tcp://localhost:6379"],
            "no",
            "Invalid REDIS_URI",
        ),
        "valid-rediss-scheme": (
            ["LIGHTRAG_KV_STORAGE=RedisKVStorage", "REDIS_URI=rediss://localhost:6380"],
            "yes",
            "",
        ),
    }
    for case_name, (extra_lines, expected_valid, expected_stderr) in cases.items():
        case_dir = tmp_path / case_name
        case_dir.mkdir()
        write_text_lines(
            case_dir / ".env",
            [
                "LIGHTRAG_KV_STORAGE=JsonKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
                *extra_lines,
            ],
        )
        write_text_lines(case_dir / "env.example", ["LLM_BINDING=openai"])
        result = subprocess.run(
            [
                "bash",
                "--norc",
                "--noprofile",
                "-c",
                f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{case_dir}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
            ],
            cwd=REPO_ROOT,
            capture_output=True,
            text=True,
            check=False,
        )
        values = parse_lines(result.stdout)
        assert values["VALID"] == expected_valid
        if expected_stderr:
            assert expected_stderr in result.stderr


def test_validate_env_file_rejects_invalid_runtime_target(tmp_path: Path) -> None:
    """validate_env_file should reject unsupported LIGHTRAG_RUNTIME_TARGET values."""
    write_text_lines(
        tmp_path / ".env",
        [
            "LIGHTRAG_RUNTIME_TARGET=laptop",
            "LIGHTRAG_KV_STORAGE=JsonKVStorage",
            "LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage",
            "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
            "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
        ],
    )
    write_text_lines(tmp_path / "env.example", ["LLM_BINDING=openai"])
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "Invalid LIGHTRAG_RUNTIME_TARGET" in result.stderr


def test_validate_required_variables_requires_opensearch_basic_auth() -> None:
    """OpenSearch storages should require both OPENSEARCH_USER and OPENSEARCH_PASSWORD."""
    values = run_bash_lines(f"""
set -euo pipefail
source "{REPO_ROOT}/scripts/setup/setup.sh"
reset_state

ENV_VALUES[LIGHTRAG_KV_STORAGE]="OpenSearchKVStorage"
ENV_VALUES[LIGHTRAG_VECTOR_STORAGE]="OpenSearchVectorDBStorage"
ENV_VALUES[LIGHTRAG_GRAPH_STORAGE]="OpenSearchGraphStorage"
ENV_VALUES[LIGHTRAG_DOC_STATUS_STORAGE]="OpenSearchDocStatusStorage"
ENV_VALUES[OPENSEARCH_HOSTS]="localhost:9200"

if validate_required_variables   "${{ENV_VALUES[LIGHTRAG_KV_STORAGE]}}"   "${{ENV_VALUES[LIGHTRAG_VECTOR_STORAGE]}}"   "${{ENV_VALUES[LIGHTRAG_GRAPH_STORAGE]}}"   "${{ENV_VALUES[LIGHTRAG_DOC_STATUS_STORAGE]}}"; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""")
    assert values["VALID"] == "no"


def test_validate_env_file_rejects_invalid_opensearch_index_settings(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject invalid OpenSearch shard and replica counts."""
    write_text_lines(
        tmp_path / ".env",
        [
            "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
            "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
            "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
            "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
            "OPENSEARCH_HOSTS=localhost:9200",
            "OPENSEARCH_USER=admin",
            "OPENSEARCH_PASSWORD=StrongPass1!",
            "OPENSEARCH_NUMBER_OF_SHARDS=abc",
            "OPENSEARCH_NUMBER_OF_REPLICAS=-1",
        ],
    )
    write_text_lines(tmp_path / "env.example", ["LLM_BINDING=openai"])
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "OPENSEARCH_NUMBER_OF_SHARDS must be a positive integer." in result.stderr


def test_validate_env_file_rejects_blank_opensearch_index_settings(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject blank OpenSearch shard and replica counts."""
    write_text_lines(
        tmp_path / ".env",
        [
            "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
            "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
            "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
            "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
            "OPENSEARCH_HOSTS=localhost:9200",
            "OPENSEARCH_USER=admin",
            "OPENSEARCH_PASSWORD=StrongPass1!",
            "OPENSEARCH_NUMBER_OF_SHARDS=",
            "OPENSEARCH_NUMBER_OF_REPLICAS=",
        ],
    )
    write_text_lines(tmp_path / "env.example", ["LLM_BINDING=openai"])
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "OPENSEARCH_NUMBER_OF_SHARDS must be a positive integer." in result.stderr


def test_validate_env_file_rejects_mongo_vector_storage_without_atlas_capable_uri(
    tmp_path: Path,
) -> None:
    """validate_env_file must reject MongoVectorDBStorage when the URI is not an Atlas cluster and no Atlas Local marker is set."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=JsonKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=MongoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
                "MONGO_URI=mongodb://localhost:27017",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "MongoVectorDBStorage requires an Atlas-capable MongoDB URI" in result.stderr


def test_validate_env_file_allows_mongo_vector_storage_with_wizard_managed_atlas_local(
    tmp_path: Path,
) -> None:
    """validate_env_file should allow MongoVectorDBStorage with the bundled Atlas Local deployment."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_SETUP_MONGODB_DEPLOYMENT=docker",
                "LIGHTRAG_KV_STORAGE=MongoKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=MongoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=MongoGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=MongoDocStatusStorage",
                "MONGO_URI=mongodb://localhost:27017/?directConnection=true",
                "MONGO_DATABASE=LightRAG",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "yes"


def test_validate_env_file_allows_external_atlas_local_for_mongo_vector_storage(
    tmp_path: Path,
) -> None:
    """validate_env_file should allow Atlas Local URIs outside the wizard-managed docker path."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=MongoKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=MongoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=MongoGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=MongoDocStatusStorage",
                "MONGO_URI=mongodb://atlas-local.example.com:27017/LightRAG?replicaSet=rs0&directConnection=true",
                "MONGO_DATABASE=LightRAG",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "yes"


def test_validate_env_file_rejects_remote_mongo_uri_with_docker_marker(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject remote mongodb:// URIs when the docker marker is set."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_SETUP_MONGODB_DEPLOYMENT=docker",
                "LIGHTRAG_KV_STORAGE=MongoKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=MongoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=MongoGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=MongoDocStatusStorage",
                "MONGO_URI=mongodb://mongo.example.com:27017/?directConnection=true",
                "MONGO_DATABASE=LightRAG",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert (
        "MongoVectorDBStorage requires the bundled Atlas Local endpoint"
        in result.stderr
    )


def test_validate_env_file_rejects_stale_local_mongo_uri_without_direct_connection(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject the old local MongoDB URI format when the docker marker is set."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_SETUP_MONGODB_DEPLOYMENT=docker",
                "LIGHTRAG_KV_STORAGE=MongoKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=MongoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=MongoGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=MongoDocStatusStorage",
                "MONGO_URI=mongodb://localhost:27017/",
                "MONGO_DATABASE=LightRAG",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert (
        "MongoVectorDBStorage requires the bundled Atlas Local endpoint"
        in result.stderr
    )


def test_validate_env_file_rejects_wrong_local_mongo_port_with_docker_marker(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject local MongoDB URIs that do not use the managed Atlas Local port."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_SETUP_MONGODB_DEPLOYMENT=docker",
                "LIGHTRAG_KV_STORAGE=MongoKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=MongoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=MongoGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=MongoDocStatusStorage",
                "MONGO_URI=mongodb://localhost:9999/?directConnection=true",
                "MONGO_DATABASE=LightRAG",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert (
        "MongoVectorDBStorage requires the bundled Atlas Local endpoint"
        in result.stderr
    )


def test_validate_env_file_rejects_empty_opensearch_hosts(tmp_path: Path) -> None:
    """validate_env_file should reject an explicitly empty OPENSEARCH_HOSTS setting."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
                "OPENSEARCH_HOSTS=",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "Empty OPENSEARCH_HOSTS" in result.stderr


def test_validate_env_file_rejects_whitespace_only_opensearch_hosts(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject OpenSearch host lists with only blank entries."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
                "OPENSEARCH_HOSTS=   ,   ",
                "OPENSEARCH_USER=admin",
                "OPENSEARCH_PASSWORD=StrongPass1!",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "OPENSEARCH_HOSTS must not contain empty host entries." in result.stderr


def test_validate_env_file_rejects_docker_opensearch_without_password(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject bundled OpenSearch when auth is incomplete."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
                "LIGHTRAG_SETUP_OPENSEARCH_DEPLOYMENT=docker",
                "OPENSEARCH_HOSTS=localhost:9200",
                "OPENSEARCH_USER=admin",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert (
        "Bundled OpenSearch requires OPENSEARCH_USER and OPENSEARCH_PASSWORD"
        in result.stderr
    )


def test_validate_env_file_rejects_weak_docker_opensearch_password(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject bundled OpenSearch passwords the image will refuse."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
                "LIGHTRAG_SETUP_OPENSEARCH_DEPLOYMENT=docker",
                "OPENSEARCH_HOSTS=localhost:9200",
                "OPENSEARCH_USER=admin",
                "OPENSEARCH_PASSWORD=weakpass",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "OpenSearch requires a strong OPENSEARCH_PASSWORD" in result.stderr


def test_validate_env_file_rejects_weak_host_opensearch_password(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject weak OpenSearch passwords even for host deployments."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
                "OPENSEARCH_HOSTS=localhost:9200",
                "OPENSEARCH_USER=admin",
                "OPENSEARCH_PASSWORD=weakpass",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "OpenSearch requires a strong OPENSEARCH_PASSWORD" in result.stderr


def test_validate_env_file_rejects_unauthenticated_host_opensearch(
    tmp_path: Path,
) -> None:
    """validate_env_file should reject host-mode OpenSearch with no auth fields."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
                "OPENSEARCH_HOSTS=localhost:9200",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "OPENSEARCH_USER" in result.stderr
    assert "OPENSEARCH_PASSWORD" in result.stderr


def test_validate_env_file_rejects_partial_host_opensearch_auth(tmp_path: Path) -> None:
    """validate_env_file should reject host-mode OpenSearch when only one auth field is set."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
                "OPENSEARCH_HOSTS=localhost:9200",
                "OPENSEARCH_USER=admin",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert "OPENSEARCH_PASSWORD" in result.stderr


def test_validate_env_file_rejects_opensearch_hosts_with_uri_scheme(
    tmp_path: Path,
) -> None:
    """validate_env_file should require OPENSEARCH_HOSTS to stay as host:port entries."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=OpenSearchKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=OpenSearchVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=OpenSearchGraphStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=OpenSearchDocStatusStorage",
                "OPENSEARCH_HOSTS=https://localhost:9200",
                "OPENSEARCH_USER=admin",
                "OPENSEARCH_PASSWORD=StrongPass1!",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "no"
    assert (
        "OPENSEARCH_HOSTS must use bare host:port entries, not URLs." in result.stderr
    )


def test_validate_env_file_ignores_invalid_unused_storage_settings(
    tmp_path: Path,
) -> None:
    """validate_env_file should ignore malformed settings for backends not selected by storage."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=JsonKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
                "NEO4J_URI=http://localhost:7687",
                "MONGO_URI=not-a-mongo-uri",
                "REDIS_URI=tcp://localhost:6379",
                "MILVUS_URI=tcp://localhost:19530",
                "QDRANT_URL=tcp://localhost:6333",
                "MEMGRAPH_URI=http://localhost:7687",
                "POSTGRES_PORT=99999",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "yes"
    assert "Invalid NEO4J_URI" not in result.stderr
    assert "Invalid MONGO_URI" not in result.stderr
    assert "Invalid REDIS_URI" not in result.stderr
    assert "Invalid MILVUS_URI" not in result.stderr
    assert "Invalid QDRANT_URL" not in result.stderr
    assert "Invalid MEMGRAPH_URI" not in result.stderr
    assert "Invalid POSTGRES_PORT" not in result.stderr


def test_validate_env_file_allows_empty_opensearch_hosts_when_unused(
    tmp_path: Path,
) -> None:
    """validate_env_file should ignore blank OpenSearch hosts when no OpenSearch storage is selected."""
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "LIGHTRAG_KV_STORAGE=JsonKVStorage",
                "LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage",
                "LIGHTRAG_GRAPH_STORAGE=NetworkXStorage",
                "LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage",
                "OPENSEARCH_HOSTS=",
            ]
        )
        + "\n",
        encoding="utf-8",
    )
    (tmp_path / "env.example").write_text("LLM_BINDING=openai\n", encoding="utf-8")
    result = subprocess.run(
        [
            "bash",
            "--norc",
            "--noprofile",
            "-c",
            f"""
source "{REPO_ROOT}/scripts/setup/setup.sh"
REPO_ROOT="{tmp_path}"
reset_state
if validate_env_file; then
  printf 'VALID=yes\\n'
else
  printf 'VALID=no\\n'
fi
""",
        ],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    values = parse_lines(result.stdout)
    assert values["VALID"] == "yes"
    assert "Empty OPENSEARCH_HOSTS" not in result.stderr