pyflowx/tests/cli/test_envdev.py

"""Tests for ops.dev 模块 envdev/dockercmd 函数 (镜像源配置/Docker 登录)."""

from __future__ import annotations

import subprocess
from pathlib import Path
from unittest.mock import MagicMock

import pytest

from pyflowx.conditions import Constants
from pyflowx.ops.dev import (
    docker_login_tencent,
    download_rustup_script,
    install_linux_docker,
    install_linux_fonts,
    install_linux_qt_libs,
    install_rust_toolchain,
    setup_conda_mirror,
    setup_linux_system_mirror,
    setup_python_mirror,
    setup_rust_mirror,
)


# ---------------------------------------------------------------------- #
# setup_python_mirror
# ---------------------------------------------------------------------- #
class TestSetupPythonMirror:
    """``setup_python_mirror`` 函数测试."""

    def test_unknown_mirror_skips(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
        """未知镜像源应打印提示并跳过."""
        monkeypatch.setattr(subprocess, "run", lambda *_, **__: MagicMock())
        setup_python_mirror("unknown_mirror")
        captured = capsys.readouterr()
        assert "未知 Python 镜像源" in captured.out

    def test_known_mirror_writes_config(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
        """已知镜像源应写入配置文件."""
        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        setup_python_mirror("tsinghua")
        # Linux 平台默认配置路径
        config_path = tmp_path / ".pip" / "pip.conf"
        if not config_path.exists():
            config_path = tmp_path / "pip" / "pip.ini"
        assert config_path.exists()
        content = config_path.read_text(encoding="utf-8")
        assert "pypi.tuna.tsinghua.edu.cn" in content

    def test_linux_uses_pip_conf(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
        """Linux 平台应写入 ~/.pip/pip.conf."""
        monkeypatch.setattr(Constants, "IS_LINUX", True)
        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        setup_python_mirror("tsinghua")
        config_path = tmp_path / ".pip" / "pip.conf"
        assert config_path.exists()

    def test_non_linux_uses_pip_ini(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
        """非 Linux 平台应写入 ~/pip/pip.ini."""
        monkeypatch.setattr(Constants, "IS_LINUX", False)
        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        setup_python_mirror("tsinghua")
        config_path = tmp_path / "pip" / "pip.ini"
        assert config_path.exists()

    def test_sets_env_vars(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
        """应设置 PIP_INDEX_URL 等环境变量."""
        import os

        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        monkeypatch.setattr(os, "environ", {})

        setup_python_mirror("aliyun")

        assert "PIP_INDEX_URL" in os.environ
        assert "mirrors.aliyun.com" in os.environ["PIP_INDEX_URL"]


# ---------------------------------------------------------------------- #
# setup_conda_mirror
# ---------------------------------------------------------------------- #
class TestSetupCondaMirror:
    """``setup_conda_mirror`` 函数测试."""

    def test_unknown_mirror_skips(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
        """未知镜像源应跳过."""
        setup_conda_mirror("unknown")
        captured = capsys.readouterr()
        assert "未知 Conda 镜像源" in captured.out

    def test_known_mirror_writes_condarc(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
        """已知镜像源应写入 ~/.condarc."""
        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        setup_conda_mirror("tsinghua")
        condarc = tmp_path / ".condarc"
        assert condarc.exists()
        content = condarc.read_text(encoding="utf-8")
        assert "tsinghua" in content
        assert "channels:" in content


# ---------------------------------------------------------------------- #
# setup_rust_mirror
# ---------------------------------------------------------------------- #
class TestSetupRustMirror:
    """``setup_rust_mirror`` 函数测试."""

    def test_unknown_mirror_skips(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
        """未知镜像源应跳过."""
        setup_rust_mirror("unknown")
        captured = capsys.readouterr()
        assert "未知 Rust 镜像源" in captured.out

    def test_known_mirror_writes_config(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
        """已知镜像源应写入 ~/.cargo/config.toml."""
        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        setup_rust_mirror("ustc", "nightly")
        config = tmp_path / ".cargo" / "config.toml"
        assert config.exists()
        content = config.read_text(encoding="utf-8")
        assert "ustc" in content

    def test_creates_sccache_dir(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
        """应创建 sccache 缓存目录."""
        from pyflowx.ops import dev as dev_module

        fake_sccache = tmp_path / ".cargo" / "sccache"
        monkeypatch.setattr(dev_module, "_RUST_SCCACHE_DIR", fake_sccache)
        monkeypatch.setattr(Path, "home", lambda: tmp_path)
        setup_rust_mirror("tsinghua")
        assert fake_sccache.exists()


# ---------------------------------------------------------------------- #
# docker_login_tencent
# ---------------------------------------------------------------------- #
class TestDockerLoginTencent:
    """``docker_login_tencent`` 函数测试."""

    def test_default_username_uses_getpass(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """未提供 username 时应使用 getpass.getuser."""
        monkeypatch.setattr("getpass.getuser", lambda: "testuser")
        ran_cmds: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())

        docker_login_tencent()

        assert ran_cmds[0][0] == "docker"
        assert ran_cmds[0][1] == "login"
        assert "testuser" in ran_cmds[0]
        assert "ccr.ccs.tencentyun.com" in ran_cmds[0]

    def test_custom_username(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """提供 username 时应使用自定义用户名."""
        ran_cmds: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())

        docker_login_tencent("myuser")

        assert "myuser" in ran_cmds[0]


# ---------------------------------------------------------------------- #
# setup_linux_system_mirror
# ---------------------------------------------------------------------- #
class TestSetupLinuxSystemMirror:
    """``setup_linux_system_mirror`` 函数测试."""

    def test_non_linux_skips(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
        """非 Linux 平台应跳过."""
        monkeypatch.setattr(Constants, "IS_LINUX", False)
        called: list[str] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: called.append(cmd))

        setup_linux_system_mirror()

        captured = capsys.readouterr()
        assert "仅在 Linux 上执行" in captured.out
        assert called == []

    def test_linux_already_configured_skips(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """Linux 上已配置国内镜像时应跳过."""
        monkeypatch.setattr(Constants, "IS_LINUX", True)

        def fake_read_text(self: Path, encoding: str = "utf-8") -> str:
            return "tsinghua mirror configured"

        monkeypatch.setattr(Path, "read_text", fake_read_text)
        called: list[str] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: called.append(cmd))

        setup_linux_system_mirror()

        captured = capsys.readouterr()
        assert "已配置" in captured.out
        assert called == []

    def test_linux_not_configured_runs_script(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """Linux 上未配置镜像时应执行下载与安装脚本."""
        monkeypatch.setattr(Constants, "IS_LINUX", True)

        def fake_read_text(self: Path, encoding: str = "utf-8") -> str:
            raise OSError("file not found")

        monkeypatch.setattr(Path, "read_text", fake_read_text)
        ran_cmds: list[str] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())

        setup_linux_system_mirror()

        captured = capsys.readouterr()
        assert "下载" in captured.out
        assert "安装" in captured.out
        assert len(ran_cmds) == 2

    def test_linux_content_without_mirror_runs_script(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """Linux 上文件存在但不包含镜像关键词时应执行脚本."""
        monkeypatch.setattr(Constants, "IS_LINUX", True)

        def fake_read_text(self: Path, encoding: str = "utf-8") -> str:
            return "deb http://archive.ubuntu.com/ubuntu/ jammy main"

        monkeypatch.setattr(Path, "read_text", fake_read_text)
        ran_cmds: list[str] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())

        setup_linux_system_mirror()

        captured = capsys.readouterr()
        assert "下载" in captured.out
        assert len(ran_cmds) == 2


# ---------------------------------------------------------------------- #
# install_linux_qt_libs / install_linux_fonts / install_linux_docker
# ---------------------------------------------------------------------- #
class TestLinuxInstallers:
    """Linux 专用安装函数测试."""

    def test_qt_libs_non_linux_skips(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
        """非 Linux 上 install_linux_qt_libs 应跳过."""
        monkeypatch.setattr(Constants, "IS_LINUX", False)
        called: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: called.append(cmd))
        install_linux_qt_libs()
        captured = capsys.readouterr()
        assert "仅在 Linux" in captured.out
        assert called == []

    def test_qt_libs_linux_runs_apt(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Linux 上应执行 apt install."""
        monkeypatch.setattr(Constants, "IS_LINUX", True)
        ran_cmds: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())
        install_linux_qt_libs()
        assert ran_cmds[0][0] == "sudo"
        assert "apt" in ran_cmds[0]
        assert "install" in ran_cmds[0]

    def test_fonts_non_linux_skips(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
        """非 Linux 上 install_linux_fonts 应跳过."""
        monkeypatch.setattr(Constants, "IS_LINUX", False)
        called: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: called.append(cmd))
        install_linux_fonts()
        captured = capsys.readouterr()
        assert "仅在 Linux" in captured.out
        assert called == []

    def test_fonts_linux_runs_apt(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Linux 上应执行 apt install 字体包."""
        monkeypatch.setattr(Constants, "IS_LINUX", True)
        ran_cmds: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())
        install_linux_fonts()
        assert "fonts-noto-cjk" in ran_cmds[0]

    def test_docker_non_linux_skips(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
        """非 Linux 上 install_linux_docker 应跳过."""
        monkeypatch.setattr(Constants, "IS_LINUX", False)
        called: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: called.append(cmd))
        install_linux_docker()
        captured = capsys.readouterr()
        assert "仅在 Linux" in captured.out
        assert called == []

    def test_docker_linux_runs_install_and_usermod(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Linux 上应执行 apt install docker-compose-v2 和 usermod."""
        monkeypatch.setattr(Constants, "IS_LINUX", True)
        monkeypatch.setattr("getpass.getuser", lambda: "testuser")
        ran_cmds: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())

        install_linux_docker()

        assert any("docker-compose-v2" in cmd for cmd in ran_cmds)
        assert any("usermod" in cmd and "docker" in cmd for cmd in ran_cmds)


# ---------------------------------------------------------------------- #
# download_rustup_script
# ---------------------------------------------------------------------- #
class TestDownloadRustupScript:
    """``download_rustup_script`` 函数测试."""

    def test_rustup_installed_skips(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
        """rustup 已安装时应跳过."""
        monkeypatch.setattr("shutil.which", lambda _cmd: "/usr/bin/rustup")
        called: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: called.append(cmd))

        download_rustup_script()

        captured = capsys.readouterr()
        assert "已安装" in captured.out
        assert called == []

    def test_windows_downloads_exe(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Windows 上应下载 rustup-init.exe."""
        monkeypatch.setattr("shutil.which", lambda _cmd: None)
        monkeypatch.setattr(Constants, "IS_WINDOWS", True)
        ran_cmds: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())

        download_rustup_script()

        assert "powershell" in ran_cmds[0]
        assert "rustup-init.exe" in ran_cmds[0]

    def test_non_windows_downloads_sh(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """非 Windows 上应下载 rustup-init.sh."""
        monkeypatch.setattr("shutil.which", lambda _cmd: None)
        monkeypatch.setattr(Constants, "IS_WINDOWS", False)
        ran_cmds: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())

        download_rustup_script()

        assert "curl" in ran_cmds[0]
        assert "rustup-init.sh" in ran_cmds[0]


# ---------------------------------------------------------------------- #
# install_rust_toolchain
# ---------------------------------------------------------------------- #
class TestInstallRustToolchain:
    """``install_rust_toolchain`` 函数测试."""

    def test_rustup_not_installed_skips(
        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
    ) -> None:
        """rustup 未安装时应跳过."""
        monkeypatch.setattr("shutil.which", lambda _cmd: None)
        called: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: called.append(cmd))

        install_rust_toolchain()

        captured = capsys.readouterr()
        assert "未安装" in captured.out
        assert called == []

    def test_rustup_installed_runs_toolchain_install(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """rustup 已安装时应执行 toolchain install."""
        monkeypatch.setattr("shutil.which", lambda _cmd: "/usr/bin/rustup")
        ran_cmds: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())

        install_rust_toolchain("nightly")

        assert ran_cmds == [["rustup", "toolchain", "install", "nightly"]]

    def test_default_version_stable(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """默认版本应为 stable."""
        monkeypatch.setattr("shutil.which", lambda _cmd: "/usr/bin/rustup")
        ran_cmds: list[list[str]] = []
        monkeypatch.setattr(subprocess, "run", lambda cmd, **_: ran_cmds.append(cmd) or MagicMock())

        install_rust_toolchain()

        assert "stable" in ran_cmds[0]