refactor: 将 cli/system、cli/llm、cli/dev 脚本迁移为 YAML 配置 + register_fn 模式

删除 cli/system (clearscreen/taskkill/which)、cli/llm (msdownload/sglang)、cli/dev (dockercmd/envdev) 三个目录; 新增 7 个 YAML 配置; 新增 ops/llm.py 模块; 扩展 ops/system.py 和 ops/dev.py; pf.py 添加工具别名; 删除 examples/ 目录

tests/cli/test_envdev.py

@@ -0,0 +1,389 @@
+"""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]