AstrBot/tests/test_pip_installer.py

import asyncio
import json
import ntpath
import threading
from pathlib import Path
from unittest.mock import AsyncMock

import pytest

from astrbot.core.utils import core_constraints as core_constraints_module
from astrbot.core.utils import pip_installer as pip_installer_module
from astrbot.core.utils import requirements_utils
from astrbot.core.utils.pip_installer import PipInstaller

WINDOWS_RUNTIME_ROOT = ntpath.join(r"C:\astrbot-test", "backend", "python")
WINDOWS_RUNTIME_EXECUTABLE = ntpath.join(WINDOWS_RUNTIME_ROOT, "python.exe")
WINDOWS_PACKAGED_RUNTIME_EXECUTABLE = f"\\\\?\\{WINDOWS_RUNTIME_EXECUTABLE}"
WINDOWS_RUNTIME_INCLUDE_DIR = ntpath.join(WINDOWS_RUNTIME_ROOT, "include")
WINDOWS_RUNTIME_LIBS_DIR = ntpath.join(WINDOWS_RUNTIME_ROOT, "libs")
EXISTING_WINDOWS_INCLUDE_DIR = ntpath.join(r"C:\toolchain", "include")
EXISTING_WINDOWS_LIB_DIR = ntpath.join(r"C:\toolchain", "lib")
_ENV_MISSING = object()


def _make_run_pip_mock(
    code: int = 0,
    output_lines: list[str] | None = None,
    conflict=None,
):
    del output_lines, conflict

    async def run_pip(*args, **kwargs):
        del args, kwargs
        return code

    return AsyncMock(side_effect=run_pip)


def _configure_run_pip_in_process_capture(
    monkeypatch,
    *,
    platform: str,
    packaged_runtime: bool,
    runtime_executable: str = WINDOWS_PACKAGED_RUNTIME_EXECUTABLE,
    include_value: str | object = _ENV_MISSING,
    lib_value: str | object = _ENV_MISSING,
    existing_runtime_dirs: set[str] | None = None,
) -> dict[str, str | None]:
    observed_env: dict[str, str | None] = {}

    def fake_pip_main(args):
        del args
        observed_env["INCLUDE"] = pip_installer_module.os.environ.get("INCLUDE")
        observed_env["LIB"] = pip_installer_module.os.environ.get("LIB")
        return 0

    if packaged_runtime:
        monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    else:
        monkeypatch.delenv("ASTRBOT_DESKTOP_CLIENT", raising=False)

    if include_value is _ENV_MISSING:
        monkeypatch.delenv("INCLUDE", raising=False)
    else:
        monkeypatch.setenv("INCLUDE", include_value)

    if lib_value is _ENV_MISSING:
        monkeypatch.delenv("LIB", raising=False)
    else:
        monkeypatch.setenv("LIB", lib_value)

    monkeypatch.setattr(pip_installer_module.sys, "platform", platform)
    monkeypatch.setattr(pip_installer_module.sys, "executable", runtime_executable)

    if existing_runtime_dirs is not None:
        monkeypatch.setattr(
            pip_installer_module.os.path,
            "isdir",
            lambda path: path in existing_runtime_dirs,
        )

    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._get_pip_main",
        lambda: fake_pip_main,
    )
    return observed_env


@pytest.fixture
def configure_run_pip_in_process_capture(monkeypatch):
    def _configure(**kwargs):
        return _configure_run_pip_in_process_capture(monkeypatch, **kwargs)

    return _configure


@pytest.mark.asyncio
async def test_install_targets_site_packages_for_desktop_client(monkeypatch, tmp_path):
    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.delattr("sys.frozen", raising=False)

    site_packages_path = tmp_path / "site-packages"
    run_pip = _make_run_pip_mock()
    prepend_sys_path_calls = []
    ensure_preferred_calls = []

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.get_astrbot_site_packages_path",
        lambda: str(site_packages_path),
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._prepend_sys_path",
        lambda path: prepend_sys_path_calls.append(path),
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._ensure_plugin_dependencies_preferred",
        lambda path, requirements: ensure_preferred_calls.append((path, requirements)),
    )

    installer = PipInstaller("")
    await installer.install(package_name="demo-package")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert "--target" in recorded_args
    assert str(site_packages_path) in recorded_args
    assert prepend_sys_path_calls == [str(site_packages_path), str(site_packages_path)]
    assert ensure_preferred_calls == [(str(site_packages_path), {"demo-package"})]


@pytest.mark.asyncio
async def test_install_keeps_target_upgrade_enabled_by_default_for_desktop_client(
    monkeypatch, tmp_path
):
    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.delattr("sys.frozen", raising=False)

    site_packages_path = tmp_path / "site-packages"
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.get_astrbot_site_packages_path",
        lambda: str(site_packages_path),
    )

    installer = PipInstaller("")
    await installer.install(package_name="demo-package")

    recorded_args = run_pip.await_args_list[0].args[0]

    assert "--target" in recorded_args
    assert "--upgrade" in recorded_args
    assert "--upgrade-strategy" in recorded_args


@pytest.mark.asyncio
async def test_install_skips_target_upgrade_when_disabled_for_desktop_client(
    monkeypatch, tmp_path
):
    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.delattr("sys.frozen", raising=False)

    site_packages_path = tmp_path / "site-packages"
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.get_astrbot_site_packages_path",
        lambda: str(site_packages_path),
    )

    installer = PipInstaller("")
    await installer.install(package_name="demo-package", allow_target_upgrade=False)

    recorded_args = run_pip.await_args_list[0].args[0]

    assert "--target" in recorded_args
    assert "--upgrade" not in recorded_args
    assert "--upgrade-strategy" not in recorded_args


@pytest.mark.asyncio
async def test_run_pip_in_process_streams_output_lines(monkeypatch):
    logged_lines = []
    first_line_seen = asyncio.Event()
    unblock_pip = threading.Event()

    def fake_pip_main(args):
        del args
        print("Collecting demo-package")
        unblock_pip.wait(timeout=1)
        print("Downloading demo-package.whl")
        return 0

    loop = asyncio.get_running_loop()

    def record_log(line, *args):
        message = line % args if args else line
        logged_lines.append(message)
        if message == "Collecting demo-package":
            loop.call_soon_threadsafe(first_line_seen.set)

    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._get_pip_main",
        lambda: fake_pip_main,
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.logger.info",
        record_log,
    )

    installer = PipInstaller("")
    task = asyncio.create_task(
        installer._run_pip_in_process(["install", "demo-package"])
    )

    await asyncio.wait_for(first_line_seen.wait(), timeout=1)
    unblock_pip.set()
    result = await task

    assert result == 0
    assert logged_lines[-2:] == [
        "Collecting demo-package",
        "Downloading demo-package.whl",
    ]


@pytest.mark.asyncio
async def test_run_pip_in_process_preserves_shared_stream_order(monkeypatch):
    logged_lines = []

    def fake_pip_main(args):
        del args
        import sys

        sys.stdout.write("out")
        sys.stderr.write("err\n")
        sys.stdout.write(" line\n")
        return 0

    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._get_pip_main",
        lambda: fake_pip_main,
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.logger.info",
        lambda line, *args: logged_lines.append(line % args if args else line),
    )

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    assert logged_lines[-2:] == ["outerr", " line"]


@pytest.mark.asyncio
async def test_run_pip_in_process_preserves_blank_lines(monkeypatch):
    logged_lines = []

    def fake_pip_main(args):
        del args
        print("Collecting demo-package")
        print()
        print("Installing collected packages")
        return 0

    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._get_pip_main",
        lambda: fake_pip_main,
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.logger.info",
        lambda line, *args: logged_lines.append(line % args if args else line),
    )

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    assert logged_lines[-3:] == [
        "Collecting demo-package",
        "",
        "Installing collected packages",
    ]


@pytest.mark.asyncio
async def test_run_pip_in_process_preserves_trailing_blank_line_on_flush(monkeypatch):
    logged_lines = []

    def fake_pip_main(args):
        del args
        import sys

        sys.stdout.write("Collecting demo-package\n\n")
        return 0

    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._get_pip_main",
        lambda: fake_pip_main,
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.logger.info",
        lambda line, *args: logged_lines.append(line % args if args else line),
    )

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    assert logged_lines[-2:] == ["Collecting demo-package", ""]


@pytest.mark.asyncio
async def test_run_pip_in_process_normalizes_crlf_without_extra_blank_lines(
    monkeypatch,
):
    logged_lines = []

    def fake_pip_main(args):
        del args
        import sys

        sys.stdout.write("Collecting demo-package\r\n")
        sys.stdout.write("Installing collected packages\r\n")
        return 0

    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._get_pip_main",
        lambda: fake_pip_main,
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.logger.info",
        lambda line, *args: logged_lines.append(line % args if args else line),
    )

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    assert logged_lines[-2:] == [
        "Collecting demo-package",
        "Installing collected packages",
    ]


@pytest.mark.parametrize(
    ("path", "expected"),
    [
        (
            WINDOWS_RUNTIME_EXECUTABLE,
            WINDOWS_RUNTIME_EXECUTABLE,
        ),
        (
            WINDOWS_PACKAGED_RUNTIME_EXECUTABLE,
            WINDOWS_RUNTIME_EXECUTABLE,
        ),
        (
            f"\\??\\{WINDOWS_RUNTIME_EXECUTABLE}",
            WINDOWS_RUNTIME_EXECUTABLE,
        ),
        (
            r"\\?\UNC\server\share\include",
            r"\\server\share\include",
        ),
        (
            r"\??\UNC\server\share\libs",
            r"\\server\share\libs",
        ),
        (
            r"\\server\share\include",
            r"\\server\share\include",
        ),
        (
            "C:/astrbot-test/backend/python/libs",
            WINDOWS_RUNTIME_LIBS_DIR,
        ),
    ],
)
def test_normalize_windows_native_build_path_variants(path, expected):
    assert pip_installer_module._normalize_windows_native_build_path(path) == expected


def test_temporary_environ_restores_previous_values(monkeypatch):
    monkeypatch.setenv("INCLUDE", EXISTING_WINDOWS_INCLUDE_DIR)
    monkeypatch.delenv("LIB", raising=False)

    with pip_installer_module._temporary_environ(
        {
            "INCLUDE": WINDOWS_RUNTIME_INCLUDE_DIR,
            "LIB": WINDOWS_RUNTIME_LIBS_DIR,
        }
    ):
        assert pip_installer_module.os.environ["INCLUDE"] == WINDOWS_RUNTIME_INCLUDE_DIR
        assert pip_installer_module.os.environ["LIB"] == WINDOWS_RUNTIME_LIBS_DIR

    assert pip_installer_module.os.environ["INCLUDE"] == EXISTING_WINDOWS_INCLUDE_DIR
    assert "LIB" not in pip_installer_module.os.environ


def test_build_packaged_windows_runtime_build_env_uses_base_env_snapshot(
    monkeypatch,
):
    snapshot_include = ntpath.join(r"C:\snapshot-toolchain", "include")
    snapshot_lib = ntpath.join(r"C:\snapshot-toolchain", "lib")
    process_include = ntpath.join(r"C:\process-toolchain", "include")
    process_lib = ntpath.join(r"C:\process-toolchain", "lib")

    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.setenv("INCLUDE", process_include)
    monkeypatch.setenv("LIB", process_lib)
    monkeypatch.setattr(pip_installer_module.sys, "platform", "win32")
    monkeypatch.setattr(
        pip_installer_module.sys,
        "executable",
        WINDOWS_PACKAGED_RUNTIME_EXECUTABLE,
    )
    monkeypatch.setattr(
        pip_installer_module.os.path,
        "isdir",
        lambda path: path in {WINDOWS_RUNTIME_INCLUDE_DIR, WINDOWS_RUNTIME_LIBS_DIR},
    )

    env_updates = pip_installer_module._build_packaged_windows_runtime_build_env(
        base_env={
            "INCLUDE": snapshot_include,
            "LIB": snapshot_lib,
        }
    )

    assert env_updates == {
        "INCLUDE": f"{WINDOWS_RUNTIME_INCLUDE_DIR};{snapshot_include}",
        "LIB": f"{WINDOWS_RUNTIME_LIBS_DIR};{snapshot_lib}",
    }


def test_build_packaged_windows_runtime_build_env_matches_snapshot_keys_case_insensitively(
    monkeypatch,
):
    snapshot_include = ntpath.join(r"C:\snapshot-toolchain", "include")
    snapshot_lib = ntpath.join(r"C:\snapshot-toolchain", "lib")

    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.setattr(pip_installer_module.sys, "platform", "win32")
    monkeypatch.setattr(
        pip_installer_module.sys,
        "executable",
        WINDOWS_PACKAGED_RUNTIME_EXECUTABLE,
    )
    monkeypatch.setattr(
        pip_installer_module.os.path,
        "isdir",
        lambda path: path in {WINDOWS_RUNTIME_INCLUDE_DIR, WINDOWS_RUNTIME_LIBS_DIR},
    )

    env_updates = pip_installer_module._build_packaged_windows_runtime_build_env(
        base_env={
            "include": snapshot_include,
            "lib": snapshot_lib,
        }
    )

    assert env_updates == {
        "INCLUDE": f"{WINDOWS_RUNTIME_INCLUDE_DIR};{snapshot_include}",
        "LIB": f"{WINDOWS_RUNTIME_LIBS_DIR};{snapshot_lib}",
    }


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("include_exists", "libs_exists"),
    [
        (True, True),
        (True, False),
        (False, True),
    ],
)
async def test_run_pip_in_process_injects_windows_runtime_build_env(
    configure_run_pip_in_process_capture, include_exists, libs_exists
):
    existing_runtime_dirs = set()
    if include_exists:
        existing_runtime_dirs.add(WINDOWS_RUNTIME_INCLUDE_DIR)
    if libs_exists:
        existing_runtime_dirs.add(WINDOWS_RUNTIME_LIBS_DIR)

    observed_env = configure_run_pip_in_process_capture(
        platform="win32",
        packaged_runtime=True,
        include_value=EXISTING_WINDOWS_INCLUDE_DIR,
        lib_value=EXISTING_WINDOWS_LIB_DIR,
        existing_runtime_dirs=existing_runtime_dirs,
    )

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    expected_include = EXISTING_WINDOWS_INCLUDE_DIR
    expected_lib = EXISTING_WINDOWS_LIB_DIR
    if include_exists:
        expected_include = (
            f"{WINDOWS_RUNTIME_INCLUDE_DIR};{EXISTING_WINDOWS_INCLUDE_DIR}"
        )
    if libs_exists:
        expected_lib = f"{WINDOWS_RUNTIME_LIBS_DIR};{EXISTING_WINDOWS_LIB_DIR}"
    assert observed_env == {"INCLUDE": expected_include, "LIB": expected_lib}
    assert pip_installer_module.os.environ["INCLUDE"] == EXISTING_WINDOWS_INCLUDE_DIR
    assert pip_installer_module.os.environ["LIB"] == EXISTING_WINDOWS_LIB_DIR


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("include_exists", "libs_exists"),
    [
        (True, True),
        (True, False),
        (False, True),
    ],
)
async def test_run_pip_in_process_injects_windows_runtime_build_env_without_existing_paths(
    configure_run_pip_in_process_capture, include_exists, libs_exists
):
    existing_runtime_dirs = set()
    if include_exists:
        existing_runtime_dirs.add(WINDOWS_RUNTIME_INCLUDE_DIR)
    if libs_exists:
        existing_runtime_dirs.add(WINDOWS_RUNTIME_LIBS_DIR)

    observed_env = configure_run_pip_in_process_capture(
        platform="win32",
        packaged_runtime=True,
        existing_runtime_dirs=existing_runtime_dirs,
    )

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    assert observed_env == {
        "INCLUDE": WINDOWS_RUNTIME_INCLUDE_DIR if include_exists else None,
        "LIB": WINDOWS_RUNTIME_LIBS_DIR if libs_exists else None,
    }
    if include_exists:
        assert ";" not in observed_env["INCLUDE"]
    if libs_exists:
        assert ";" not in observed_env["LIB"]
    assert "INCLUDE" not in pip_installer_module.os.environ
    assert "LIB" not in pip_installer_module.os.environ


@pytest.mark.asyncio
async def test_run_pip_in_process_does_not_inject_when_runtime_dirs_missing(
    configure_run_pip_in_process_capture,
):
    observed_env = configure_run_pip_in_process_capture(
        platform="win32",
        packaged_runtime=True,
        include_value=EXISTING_WINDOWS_INCLUDE_DIR,
        lib_value=EXISTING_WINDOWS_LIB_DIR,
        existing_runtime_dirs=set(),
    )

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    assert observed_env == {
        "INCLUDE": EXISTING_WINDOWS_INCLUDE_DIR,
        "LIB": EXISTING_WINDOWS_LIB_DIR,
    }
    assert pip_installer_module.os.environ["INCLUDE"] == EXISTING_WINDOWS_INCLUDE_DIR
    assert pip_installer_module.os.environ["LIB"] == EXISTING_WINDOWS_LIB_DIR


@pytest.mark.asyncio
async def test_run_pip_in_process_uses_latest_env_when_building_runtime_paths(
    monkeypatch,
    configure_run_pip_in_process_capture,
):
    updated_include = ntpath.join(r"C:\new-toolchain", "include")
    updated_lib = ntpath.join(r"C:\new-toolchain", "lib")
    observed_env = configure_run_pip_in_process_capture(
        platform="win32",
        packaged_runtime=True,
        include_value=EXISTING_WINDOWS_INCLUDE_DIR,
        lib_value=EXISTING_WINDOWS_LIB_DIR,
        existing_runtime_dirs={
            WINDOWS_RUNTIME_INCLUDE_DIR,
            WINDOWS_RUNTIME_LIBS_DIR,
        },
    )

    async def fake_to_thread(func, *args):
        pip_installer_module.os.environ["INCLUDE"] = updated_include
        pip_installer_module.os.environ["LIB"] = updated_lib
        return func(*args)

    monkeypatch.setattr(pip_installer_module.asyncio, "to_thread", fake_to_thread)

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    assert observed_env == {
        "INCLUDE": f"{WINDOWS_RUNTIME_INCLUDE_DIR};{updated_include}",
        "LIB": f"{WINDOWS_RUNTIME_LIBS_DIR};{updated_lib}",
    }
    assert pip_installer_module.os.environ["INCLUDE"] == updated_include
    assert pip_installer_module.os.environ["LIB"] == updated_lib


@pytest.mark.asyncio
async def test_run_pip_in_process_does_not_modify_env_on_non_windows(
    configure_run_pip_in_process_capture,
):
    existing_include = "/toolchain/include"
    existing_lib = "/toolchain/lib"
    observed_env = configure_run_pip_in_process_capture(
        platform="linux",
        packaged_runtime=True,
        include_value=existing_include,
        lib_value=existing_lib,
    )

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    assert observed_env == {"INCLUDE": existing_include, "LIB": existing_lib}
    assert pip_installer_module.os.environ["INCLUDE"] == existing_include
    assert pip_installer_module.os.environ["LIB"] == existing_lib


@pytest.mark.asyncio
async def test_run_pip_in_process_does_not_inject_env_when_not_packaged(
    configure_run_pip_in_process_capture,
):
    observed_env = configure_run_pip_in_process_capture(
        platform="win32",
        packaged_runtime=False,
        existing_runtime_dirs={
            WINDOWS_RUNTIME_INCLUDE_DIR,
            WINDOWS_RUNTIME_LIBS_DIR,
        },
    )

    installer = PipInstaller("")
    result = await installer._run_pip_in_process(["install", "demo-package"])

    assert result == 0
    assert observed_env == {"INCLUDE": None, "LIB": None}
    assert "INCLUDE" not in pip_installer_module.os.environ
    assert "LIB" not in pip_installer_module.os.environ


@pytest.mark.asyncio
async def test_run_pip_in_process_classifies_nonstandard_conflict_output(monkeypatch):
    def fake_pip_main(args):
        del args
        print(
            "Cannot install demo-package and astrbot-core because these package "
            "versions have conflicting dependencies."
        )
        print("The conflict is caused by:")
        print("    demo-package depends on shared-lib>=3.0")
        print("    AstrBot (constraint) depends on shared-lib==2.0")
        return 1

    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._get_pip_main",
        lambda: fake_pip_main,
    )

    installer = PipInstaller("")
    with pytest.raises(pip_installer_module.DependencyConflictError) as exc_info:
        await installer._run_pip_in_process(["install", "demo-package"])

    assert exc_info.value.is_core_conflict is True
    assert "demo-package" in str(exc_info.value)
    assert "demo-package depends on shared-lib>=3.0" in str(exc_info.value)
    assert "AstrBot (constraint) depends on shared-lib==2.0" in str(exc_info.value)
    assert "The conflict is caused by:" in exc_info.value.errors


@pytest.mark.asyncio
async def test_install_raises_dedicated_pip_install_error_on_non_conflict_failure(
    monkeypatch,
):
    async def failing_run_pip(self, args):
        del self, args
        return 2

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", failing_run_pip)

    installer = PipInstaller("")

    with pytest.raises(pip_installer_module.PipInstallError, match="错误码：2"):
        await installer.install(package_name="demo-package")


@pytest.mark.asyncio
async def test_run_pip_with_classification_raises_install_error_on_non_conflict_failure(
    monkeypatch,
):
    async def failing_run_pip(self, args):
        del self, args
        return 3

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", failing_run_pip)

    installer = PipInstaller("")

    with pytest.raises(pip_installer_module.PipInstallError, match="错误码：3"):
        await installer._run_pip_with_classification(["install", "demo-package"])


@pytest.mark.asyncio
async def test_run_pip_in_process_bounds_retained_conflict_lines(monkeypatch):
    def fake_pip_main(args):
        del args
        for index in range(10):
            print(f"noise-{index}")
        print(
            "Cannot install demo-package and astrbot-core because these package "
            "versions have conflicting dependencies."
        )
        print("The conflict is caused by:")
        print("    demo-package depends on shared-lib>=3.0")
        print("    AstrBot (constraint) depends on shared-lib==2.0")
        return 1

    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._get_pip_main",
        lambda: fake_pip_main,
    )
    monkeypatch.setattr("astrbot.core.utils.pip_installer._MAX_PIP_OUTPUT_LINES", 4)

    installer = PipInstaller("")
    with pytest.raises(pip_installer_module.DependencyConflictError) as exc_info:
        await installer._run_pip_in_process(["install", "demo-package"])

    assert len(exc_info.value.errors) == 4
    assert exc_info.value.errors[0].startswith("Cannot install demo-package")
    assert (
        exc_info.value.errors[-1]
        == "    AstrBot (constraint) depends on shared-lib==2.0"
    )


def test_build_pip_args_rejects_package_name_and_requirements_path_together(tmp_path):
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text("demo-package\n", encoding="utf-8")

    installer = PipInstaller("")

    with pytest.raises(ValueError, match="package_name and requirements_path"):
        installer._build_pip_args("requests", str(requirements_path), None)


def _make_fake_distribution(name: str, version: str):
    class FakeDistribution:
        metadata = {"Name": name}

        def __init__(self, version: str):
            self.version = version

    return FakeDistribution(version)


def test_find_missing_requirements_honors_version_specifiers(monkeypatch, tmp_path):
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text("demo-package>=2.0\n", encoding="utf-8")

    monkeypatch.setattr(
        pip_installer_module.importlib_metadata,
        "distributions",
        lambda path: [_make_fake_distribution("demo-package", "1.0")],
    )

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing == {"demo-package"}


def test_find_missing_requirements_skips_unmatched_markers(monkeypatch, tmp_path):
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text(
        'demo-package; sys_platform == "win32"\n',
        encoding="utf-8",
    )

    monkeypatch.setattr(
        pip_installer_module.importlib_metadata,
        "distributions",
        lambda path: [],
    )

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing == set()


def test_find_missing_requirements_follows_nested_requirement_files(
    monkeypatch, tmp_path
):
    base_requirements = tmp_path / "base.txt"
    base_requirements.write_text("demo-package==1.0\n", encoding="utf-8")
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text("-r base.txt\n", encoding="utf-8")

    monkeypatch.setattr(
        pip_installer_module.importlib_metadata,
        "distributions",
        lambda path: [],
    )

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing == {"demo-package"}


def test_find_missing_requirements_follows_equals_form_nested_requirements(
    monkeypatch, tmp_path
):
    base_requirements = tmp_path / "base.txt"
    base_requirements.write_text("demo-package==1.0\n", encoding="utf-8")
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text("--requirement=base.txt\n", encoding="utf-8")

    monkeypatch.setattr(
        pip_installer_module.importlib_metadata,
        "distributions",
        lambda path: [],
    )

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing == {"demo-package"}


def test_find_missing_requirements_returns_none_when_nested_file_missing(tmp_path):
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text("-r base.txt\n", encoding="utf-8")

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing is None


def test_find_missing_requirements_extracts_editable_vcs_requirement(
    monkeypatch, tmp_path
):
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text(
        "-e git+https://example.com/demo.git#egg=demo-package\n",
        encoding="utf-8",
    )

    monkeypatch.setattr(
        pip_installer_module.importlib_metadata,
        "distributions",
        lambda path: [],
    )

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing == {"demo-package"}


def test_find_missing_requirements_prefers_first_search_path_version(
    monkeypatch, tmp_path
):
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text("demo-package>=2.0\n", encoding="utf-8")

    monkeypatch.setattr(
        pip_installer_module.importlib_metadata,
        "distributions",
        lambda path: [
            _make_fake_distribution("demo-package", "1.0"),
            _make_fake_distribution("demo-package", "3.0"),
        ],
    )

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing == {"demo-package"}


def test_find_missing_requirements_returns_none_when_distribution_scan_fails(
    monkeypatch, tmp_path
):
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text("demo-package>=2.0\n", encoding="utf-8")

    def failing_distributions(path):
        del path
        yield _make_fake_distribution("demo-package", "3.0")
        raise RuntimeError("scan failed")

    monkeypatch.setattr(
        pip_installer_module.importlib_metadata,
        "distributions",
        failing_distributions,
    )

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing is None


def test_get_core_constraints_caches_fallback_resolution(monkeypatch):
    distribution_calls = []
    distributions_calls = []

    class FakeFallbackDistribution:
        metadata = {"Name": "AstrBot-App"}
        requires = ["shared-lib>=1.0"]

        def read_text(self, name):
            if name == "top_level.txt":
                return "astrbot\n"
            return ""

    fake_distribution = FakeFallbackDistribution()

    def mock_distribution(name):
        distribution_calls.append(name)
        if name == "AstrBot":
            raise pip_installer_module.importlib_metadata.PackageNotFoundError
        if name == "AstrBot-App":
            return fake_distribution
        raise pip_installer_module.importlib_metadata.PackageNotFoundError

    def mock_distributions(path=None):
        del path
        distributions_calls.append("scan")
        return [fake_distribution]

    monkeypatch.setattr(
        core_constraints_module.importlib_metadata,
        "distribution",
        mock_distribution,
    )
    monkeypatch.setattr(
        core_constraints_module.importlib_metadata,
        "distributions",
        mock_distributions,
    )
    monkeypatch.setattr(
        core_constraints_module,
        "collect_installed_distribution_versions",
        lambda paths: {"shared-lib": "2.0"},
    )

    core_constraints_module._get_core_constraints.cache_clear()
    try:
        first = core_constraints_module._get_core_constraints(None)
        second = core_constraints_module._get_core_constraints(None)
    finally:
        core_constraints_module._get_core_constraints.cache_clear()

    assert first == ("shared-lib==2.0",)
    assert second == ("shared-lib==2.0",)
    assert distribution_calls == ["AstrBot", "AstrBot-App"]
    assert distributions_calls == ["scan"]


def test_get_core_constraints_skips_distributions_with_unreadable_top_level(
    monkeypatch,
):
    class BrokenDistribution:
        metadata = {"Name": "Broken-App"}
        requires = []

        def read_text(self, name):
            if name == "top_level.txt":
                raise OSError("cannot read top_level.txt")
            return ""

    class FakeFallbackDistribution:
        metadata = {"Name": "AstrBot-App"}
        requires = ["shared-lib>=1.0"]

        def read_text(self, name):
            if name == "top_level.txt":
                return "astrbot\n"
            return ""

    broken_distribution = BrokenDistribution()
    fake_distribution = FakeFallbackDistribution()

    def mock_distribution(name):
        if name == "AstrBot":
            raise pip_installer_module.importlib_metadata.PackageNotFoundError
        if name == "AstrBot-App":
            return fake_distribution
        raise pip_installer_module.importlib_metadata.PackageNotFoundError

    def mock_distributions(path=None):
        del path
        return [broken_distribution, fake_distribution]

    monkeypatch.setattr(
        core_constraints_module.importlib_metadata,
        "distribution",
        mock_distribution,
    )
    monkeypatch.setattr(
        core_constraints_module.importlib_metadata,
        "distributions",
        mock_distributions,
    )
    monkeypatch.setattr(
        core_constraints_module,
        "collect_installed_distribution_versions",
        lambda paths: {"shared-lib": "2.0"},
    )

    core_constraints_module._get_core_constraints.cache_clear()
    try:
        constraints = core_constraints_module._get_core_constraints(None)
    finally:
        core_constraints_module._get_core_constraints.cache_clear()

    assert constraints == ("shared-lib==2.0",)


def test_core_constraints_file_propagates_inner_conflict_without_fake_warning(
    monkeypatch,
):
    warning_logs = []
    conflict = pip_installer_module.DependencyConflictError(
        "core conflict",
        [],
        is_core_conflict=True,
    )

    monkeypatch.setattr(
        core_constraints_module,
        "_get_core_constraints",
        lambda core_dist_name: ("aiohttp==3.13.3",),
    )
    monkeypatch.setattr(
        "astrbot.core.utils.core_constraints.logger.warning",
        lambda line, *args: warning_logs.append(line % args if args else line),
    )

    with pytest.raises(
        pip_installer_module.DependencyConflictError,
        match="core conflict",
    ):
        provider = core_constraints_module.CoreConstraintsProvider("AstrBot")
        with provider.constraints_file() as constraints_path:
            assert constraints_path is not None
            raise conflict

    assert warning_logs == []


@pytest.mark.asyncio
async def test_install_adds_desktop_core_lock_constraints_for_packaged_runtime(
    monkeypatch, tmp_path
):
    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.delattr("sys.frozen", raising=False)

    lock_path = tmp_path / "runtime-core-lock.json"
    lock_path.write_text(
        json.dumps(
            {
                "version": 1,
                "distributions": [
                    {
                        "name": "desktop-only-core",
                        "version": "9.9.9",
                        "top_level_modules": ["desktop_only_core"],
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("ASTRBOT_DESKTOP_CORE_LOCK_PATH", str(lock_path))

    site_packages_path = tmp_path / "site-packages"
    captured_constraints = []

    async def capture_pip_args(self, args):
        del self
        constraints_path = args[args.index("-c") + 1]
        captured_constraints.append(Path(constraints_path).read_text(encoding="utf-8"))
        return 0

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", capture_pip_args)
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.get_astrbot_site_packages_path",
        lambda: str(site_packages_path),
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._ensure_plugin_dependencies_preferred",
        lambda path, requirements: None,
    )

    installer = PipInstaller("")
    await installer.install(package_name="Cua")

    assert captured_constraints
    assert "desktop-only-core==9.9.9" in captured_constraints[0]


def test_ensure_plugin_dependencies_preferred_skips_desktop_core_lock_modules(
    monkeypatch, tmp_path
):
    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    lock_path = tmp_path / "runtime-core-lock.json"
    lock_path.write_text(
        json.dumps(
            {
                "version": 1,
                "distributions": [
                    {
                        "name": "openai",
                        "version": "2.32.0",
                        "top_level_modules": ["openai"],
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("ASTRBOT_DESKTOP_CORE_LOCK_PATH", str(lock_path))

    preferred_calls = []

    monkeypatch.setattr(
        pip_installer_module,
        "_collect_candidate_modules",
        lambda requirements, site_packages_path: {"openai", "cua_agent"},
    )
    monkeypatch.setattr(
        pip_installer_module,
        "_ensure_preferred_modules",
        lambda modules, site_packages_path: preferred_calls.append(modules),
    )

    pip_installer_module._ensure_plugin_dependencies_preferred(
        str(tmp_path / "site-packages"),
        {"Cua"},
    )

    assert preferred_calls == [{"cua_agent"}]


def test_iter_requirement_lines_expands_nested_requirement_files(tmp_path):
    base_requirements = tmp_path / "base.txt"
    base_requirements.write_text("demo-package==1.0\n", encoding="utf-8")
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text(
        "# comment\n-r base.txt\n--extra-index-url https://example.com/simple\n",
        encoding="utf-8",
    )

    lines = list(requirements_utils._iter_requirement_lines(str(requirements_path)))

    assert lines == [
        "demo-package==1.0",
        "--extra-index-url https://example.com/simple",
    ]


def test_build_pip_args_extracts_requested_requirements():
    installer = PipInstaller("")

    args, requested = installer._build_pip_args(
        "--index-url https://example.com/simple demo-package",
        None,
        None,
    )

    assert args == [
        "install",
        "--index-url",
        "https://example.com/simple",
        "demo-package",
    ]
    assert requested == {"demo-package"}


def test_build_pip_args_appends_default_index_when_not_overridden():
    installer = PipInstaller("")

    args, requested = installer._build_pip_args("demo-package", None, None)

    assert args == ["install", "demo-package", "-i", "https://pypi.org/simple"]
    assert requested == {"demo-package"}


@pytest.mark.asyncio
async def test_install_splits_space_separated_packages(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name="demo-package another-package>=1.0")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:3] == ["install", "demo-package", "another-package>=1.0"]


@pytest.mark.asyncio
async def test_install_splits_three_space_separated_packages(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(
        package_name="demo-package another-package extra-package>=1.0"
    )

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:4] == [
        "install",
        "demo-package",
        "another-package",
        "extra-package>=1.0",
    ]


@pytest.mark.asyncio
async def test_install_splits_three_bare_packages(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name="demo-package another-package extra-package")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:4] == [
        "install",
        "demo-package",
        "another-package",
        "extra-package",
    ]


@pytest.mark.asyncio
async def test_install_tracks_multiline_packages_for_desktop_client(
    monkeypatch, tmp_path
):
    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.delattr("sys.frozen", raising=False)

    site_packages_path = tmp_path / "site-packages"
    run_pip = _make_run_pip_mock()
    ensure_preferred_calls = []

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.get_astrbot_site_packages_path",
        lambda: str(site_packages_path),
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._prepend_sys_path",
        lambda path: None,
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._ensure_plugin_dependencies_preferred",
        lambda path, requirements: ensure_preferred_calls.append((path, requirements)),
    )

    installer = PipInstaller("")
    await installer.install(package_name="demo-package\nanother-package>=1.0\n")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:3] == ["install", "demo-package", "another-package>=1.0"]
    assert ensure_preferred_calls == [
        (str(site_packages_path), {"demo-package", "another-package"})
    ]


@pytest.mark.asyncio
async def test_install_splits_space_separated_packages_within_multiline_input(
    monkeypatch,
):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(
        package_name="demo-package another-package\nextra-package\n"
    )

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:4] == [
        "install",
        "demo-package",
        "another-package",
        "extra-package",
    ]


@pytest.mark.asyncio
async def test_install_keeps_single_requirement_with_marker_intact(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name="demo-package ; python_version < '4'")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:2] == [
        "install",
        "demo-package ; python_version < '4'",
    ]


@pytest.mark.asyncio
async def test_install_keeps_single_requirement_with_compact_marker_intact(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name='demo-package; python_version < "4"')

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:2] == [
        "install",
        'demo-package; python_version < "4"',
    ]


@pytest.mark.asyncio
async def test_install_keeps_single_requirement_with_version_range_intact(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name="demo-package >= 1.0, < 2.0")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:2] == [
        "install",
        "demo-package >= 1.0, < 2.0",
    ]


@pytest.mark.asyncio
async def test_install_tracks_only_real_requirement_names_for_spaced_single_requirement(
    monkeypatch, tmp_path
):
    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.delattr("sys.frozen", raising=False)

    site_packages_path = tmp_path / "site-packages"
    run_pip = _make_run_pip_mock()
    ensure_preferred_calls = []

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.get_astrbot_site_packages_path",
        lambda: str(site_packages_path),
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._prepend_sys_path",
        lambda path: None,
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._ensure_plugin_dependencies_preferred",
        lambda path, requirements: ensure_preferred_calls.append((path, requirements)),
    )

    installer = PipInstaller("")
    await installer.install(package_name="demo-package >= 1.0, < 2.0")

    assert ensure_preferred_calls == [(str(site_packages_path), {"demo-package"})]


def test_prefer_installed_dependencies_prefers_modules_for_requirements_in_desktop_runtime(
    monkeypatch, tmp_path
):
    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.delattr("sys.frozen", raising=False)

    site_packages_path = tmp_path / "site-packages"
    site_packages_path.mkdir()
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text("demo-package>=1.0\n", encoding="utf-8")

    prepend_calls = []
    preferred_calls = []

    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.get_astrbot_site_packages_path",
        lambda: str(site_packages_path),
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._prepend_sys_path",
        lambda path: prepend_calls.append(path),
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._ensure_plugin_dependencies_preferred",
        lambda path, requirements: preferred_calls.append((path, requirements)),
    )

    installer = PipInstaller("")
    installer.prefer_installed_dependencies(str(requirements_path))

    assert prepend_calls == [str(site_packages_path)]
    assert preferred_calls == [(str(site_packages_path), {"demo-package"})]


@pytest.mark.asyncio
async def test_install_multiline_input_strips_comments_and_splits_options(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(
        package_name=(
            "demo-package==1.0  # pinned\n"
            "--extra-index-url https://example.com/simple\n"
            "another-package\n"
        )
    )

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:5] == [
        "install",
        "demo-package==1.0",
        "--extra-index-url",
        "https://example.com/simple",
        "another-package",
    ]


@pytest.mark.asyncio
async def test_install_single_line_input_strips_inline_comment(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name="requests==2.31.0  # latest")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:2] == ["install", "requests==2.31.0"]


@pytest.mark.asyncio
async def test_install_splits_single_line_editable_option_input(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name="-e .")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:3] == ["install", "-e", "."]


@pytest.mark.asyncio
async def test_install_splits_single_line_option_with_url(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(
        package_name="--index-url https://example.com/simple demo-package"
    )

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:4] == [
        "install",
        "--index-url",
        "https://example.com/simple",
        "demo-package",
    ]
    assert recorded_args.count("--index-url") == 1
    assert "-i" not in recorded_args


@pytest.mark.asyncio
async def test_install_tracks_requirement_name_for_single_line_option_input(
    monkeypatch, tmp_path
):
    monkeypatch.setenv("ASTRBOT_DESKTOP_CLIENT", "1")
    monkeypatch.delattr("sys.frozen", raising=False)

    site_packages_path = tmp_path / "site-packages"
    run_pip = _make_run_pip_mock()
    ensure_preferred_calls = []

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.get_astrbot_site_packages_path",
        lambda: str(site_packages_path),
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._prepend_sys_path",
        lambda path: None,
    )
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer._ensure_plugin_dependencies_preferred",
        lambda path, requirements: ensure_preferred_calls.append((path, requirements)),
    )

    installer = PipInstaller("")
    await installer.install(
        package_name="--index-url https://example.com/simple demo-package"
    )

    assert ensure_preferred_calls == [(str(site_packages_path), {"demo-package"})]


@pytest.mark.asyncio
async def test_install_keeps_equals_form_index_override(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(
        package_name="--index-url=https://example.com/simple demo-package"
    )

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:3] == [
        "install",
        "--index-url=https://example.com/simple",
        "demo-package",
    ]
    assert "-i" not in recorded_args


@pytest.mark.asyncio
async def test_install_keeps_short_form_index_override(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name="-ihttps://example.com/simple demo-package")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:3] == [
        "install",
        "-ihttps://example.com/simple",
        "demo-package",
    ]
    assert "-i" not in recorded_args


@pytest.mark.asyncio
async def test_install_preserves_url_fragment_in_option_input(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(
        package_name="--index-url https://example.com/simple#frag demo-package"
    )

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:4] == [
        "install",
        "--index-url",
        "https://example.com/simple#frag",
        "demo-package",
    ]
    assert "-i" not in recorded_args


def test_find_missing_requirements_returns_none_for_editable_local_path_reference(
    tmp_path,
):
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text("-e ../sharedlib\n", encoding="utf-8")

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing is None


@pytest.mark.parametrize(
    "requirement_line",
    [
        "-e sharedlib\n",
        "--editable=.\\sharedlib\n",
    ],
)
def test_find_missing_requirements_returns_none_for_editable_local_path_variants(
    tmp_path, requirement_line
):
    requirements_path = tmp_path / "requirements.txt"
    requirements_path.write_text(requirement_line, encoding="utf-8")

    missing = requirements_utils.find_missing_requirements(str(requirements_path))

    assert missing is None


@pytest.mark.asyncio
async def test_install_strips_inline_comment_from_option_line(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(
        package_name=(
            "--extra-index-url https://example.com/simple  # mirror\ndemo-package\n"
        )
    )

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:4] == [
        "install",
        "--extra-index-url",
        "https://example.com/simple",
        "demo-package",
    ]


@pytest.mark.asyncio
async def test_install_falls_back_to_raw_input_for_invalid_token_string(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    raw_input = "demo-package !!! another-package"
    await installer.install(package_name=raw_input)

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:4] == ["install", "demo-package", "!!!", "another-package"]


@pytest.mark.asyncio
async def test_install_ignores_whitespace_only_package_string(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name="   ")

    run_pip.assert_not_awaited()


@pytest.mark.asyncio
async def test_install_ignores_missing_package_and_requirements(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install()

    run_pip.assert_not_awaited()


@pytest.mark.asyncio
async def test_install_respects_index_override_in_pip_install_arg(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("--index-url https://example.com/simple")
    await installer.install(package_name="demo-package")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert "install" in recorded_args
    assert "demo-package" in recorded_args
    assert "--index-url" in recorded_args
    assert "https://example.com/simple" in recorded_args
    # Verify that default index overrides are NOT present
    assert "mirrors.aliyun.com" not in recorded_args
    assert "https://pypi.org/simple" not in recorded_args


@pytest.mark.asyncio
async def test_install_respects_no_index_with_find_links(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(
        package_name="--no-index --find-links /tmp/wheels demo-package"
    )

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert recorded_args[0:5] == [
        "install",
        "--no-index",
        "--find-links",
        "/tmp/wheels",
        "demo-package",
    ]
    assert "-i" not in recorded_args


def test_redact_pip_args_for_logging_redacts_inline_url_credentials():
    redacted_args = pip_installer_module._redact_pip_args_for_logging(
        [
            "install",
            "--index-url=https://user:secret@example.com/simple",
            "demo-package",
        ]
    )

    assert redacted_args == [
        "install",
        "--index-url=https://<redacted>@example.com/simple",
        "demo-package",
    ]


def test_redact_pip_args_for_logging_redacts_sensitive_option_value_pairs():
    redacted_args = pip_installer_module._redact_pip_args_for_logging(
        [
            "install",
            "--password",
            "super-secret",
            "--token",
            "opaque-token",
            "demo-package",
        ]
    )

    assert redacted_args == [
        "install",
        "--password",
        "****",
        "--token",
        "****",
        "demo-package",
    ]


def test_redact_pip_args_for_logging_redacts_inline_sensitive_values():
    redacted_args = pip_installer_module._redact_pip_args_for_logging(
        [
            "install",
            "--api-token=super-secret",
            "password=hunter2",
            "demo-package",
        ]
    )

    assert redacted_args == [
        "install",
        "--api-token=****",
        "password=****",
        "demo-package",
    ]


@pytest.mark.asyncio
async def test_install_logs_redacted_pip_argv_when_credentials_present(monkeypatch):
    run_pip = _make_run_pip_mock()
    logged_lines = []

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
    monkeypatch.setattr(
        "astrbot.core.utils.pip_installer.logger.info",
        lambda line, *args: logged_lines.append(line % args if args else line),
    )

    installer = PipInstaller("")
    await installer.install(
        package_name="--index-url https://user:secret@example.com/simple demo-package"
    )

    argv_logs = [line for line in logged_lines if line.startswith("Pip 包管理器 argv:")]

    assert len(argv_logs) == 1
    assert "secret" not in argv_logs[0]
    assert "user:" not in argv_logs[0]
    assert "https://<redacted>@example.com/simple" in argv_logs[0]


@pytest.mark.asyncio
async def test_install_does_not_add_aliyun_trusted_host_for_default_index(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("")
    await installer.install(package_name="demo-package")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert "-i" in recorded_args
    assert "https://pypi.org/simple" in recorded_args
    assert "--trusted-host" not in recorded_args


@pytest.mark.asyncio
async def test_install_adds_aliyun_trusted_host_only_for_aliyun_index(monkeypatch):
    run_pip = _make_run_pip_mock()

    monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)

    installer = PipInstaller("", pypi_index_url="https://mirrors.aliyun.com/simple")
    await installer.install(package_name="demo-package")

    run_pip.assert_awaited_once()
    recorded_args = run_pip.await_args_list[0].args[0]

    assert "-i" in recorded_args
    assert "https://mirrors.aliyun.com/simple" in recorded_args
    trusted_host_index = recorded_args.index("--trusted-host")
    assert recorded_args[trusted_host_index + 1] == "mirrors.aliyun.com"