diff --git a/pyproject.toml b/pyproject.toml index a5507c5..d0a4f21 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,8 +29,6 @@ bumpversion = "pyflowx.cli.bumpversion:main" clr = "pyflowx.cli.clearscreen:main" emlman = "pyflowx.cli.emlmanager:main" envdev = "pyflowx.cli.envdev:main" -envpy = "pyflowx.cli.envpy:main" -envrs = "pyflowx.cli.envrs:main" filedate = "pyflowx.cli.filedate:main" filelvl = "pyflowx.cli.filelevel:main" foldback = "pyflowx.cli.folderback:main" diff --git a/src/pyflowx/cli/envdev.py b/src/pyflowx/cli/envdev.py index 04241b6..8a79032 100644 --- a/src/pyflowx/cli/envdev.py +++ b/src/pyflowx/cli/envdev.py @@ -127,6 +127,37 @@ CHINESE_FONTS: list[str] = [ "fonts-noto-color-emoji", ] +# ============================================================================ +# Rust 配置 +# ============================================================================ +RustMirrorType = Literal["tsinghua", "ustc", "aliyun"] +RustVersionType = Literal["stable", "nightly", "beta"] +DEFAULT_RUST_VERSION: RustVersionType = "stable" +DEFAULT_MIRROR: RustMirrorType = "tsinghua" + +RUSTUP_MIRRORS: dict[RustMirrorType, dict[str, str]] = { + "tsinghua": { + "RUSTUP_DIST_SERVER": "https://mirrors.tuna.tsinghua.edu.cn/rustup", + "RUSTUP_UPDATE_ROOT": "https://mirrors.tuna.tsinghua.edu.cn/rustup/rustup", + "TOML_REGISTRY": "https://mirrors.tuna.tsinghua.edu.cn/crates.io-index/", + }, + "aliyun": { + "RUSTUP_DIST_SERVER": "https://mirrors.aliyun.com/rustup", + "RUSTUP_UPDATE_ROOT": "https://mirrors.aliyun.com/rustup/rustup", + "TOML_REGISTRY": "https://mirrors.aliyun.com/crates.io-index/", + }, + "ustc": { + "RUSTUP_DIST_SERVER": "https://mirrors.ustc.edu.cn/rust-static", + "RUSTUP_UPDATE_ROOT": "https://mirrors.ustc.edu.cn/rust-static/rustup", + "TOML_REGISTRY": "https://mirrors.ustc.edu.cn/crates.io-index/", + }, +} +RUSTUP_DOWNLOAD_URL_LINUX = "https://mirrors.aliyun.com/repo/rust/rustup-init.sh" +RUSTUP_DOWNLOAD_URL_WINDOWS = "https://static.rust-lang.org/rustup/dist/x86_64-pc-windows-msvc/rustup-init.exe" +RUST_CONFIG_PATH = Path.home() / ".cargo" / "config.toml" +RUST_SCCACHE_DIR: Path = Path.home() / ".cargo" / "sccache" +RUST_SCCACHE_CACHE_SIZE: str = "20G" + def main() -> None: """主函数.""" @@ -147,14 +178,34 @@ def main() -> None: choices=get_args(CondaMirrorType), help="Conda 镜镜像源", ) + parser.add_argument( + "--rust-mirror", + nargs="?", + type=str, + default=DEFAULT_MIRROR, + choices=get_args(RustMirrorType), + help="Rust 镜像源", + ) + parser.add_argument( + "--rust-version", + nargs="?", + type=str, + default=DEFAULT_RUST_VERSION, + choices=get_args(RustVersionType), + help=f"Rust 版本, 推荐: {get_args(RustVersionType)}", + ) args = parser.parse_args() python_mirror = args.python_mirror conda_mirror_urls = CONDA_MIRROR_URLS[args.conda_mirror] + rust_mirror = args.rust_mirror + rust_version = args.rust_version # 确保配置文件目录存在 PIP_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) CONDA_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) + RUST_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) + RUST_SCCACHE_DIR.mkdir(parents=True, exist_ok=True) # 使用 conditions 自动控制任务执行 graph = px.Graph.from_specs([ @@ -222,5 +273,59 @@ def main() -> None: str(CONDA_CONFIG_PATH), "show_channel_urls: true\nchannels:\n - " + "\n - ".join(conda_mirror_urls) + "\n - defaults", ), + # 设置 Rust 镜像源 + *setenv_group({ + "RUSTUP_DIST_SERVER": RUSTUP_MIRRORS[rust_mirror]["RUSTUP_DIST_SERVER"], + "RUSTUP_UPDATE_ROOT": RUSTUP_MIRRORS[rust_mirror]["RUSTUP_UPDATE_ROOT"], + "RUST_SCCACHE_DIR": str(RUST_SCCACHE_DIR), + "RUST_SCCACHE_CACHE_SIZE": RUST_SCCACHE_CACHE_SIZE, + }), + # 写入 Rust 配置（仅当未配置） + write_file( + str(RUST_CONFIG_PATH), + f""" +[source.crates-io] +replace-with = '{rust_mirror}' + +[source.{rust_mirror}] +registry = "sparse+{RUSTUP_MIRRORS[rust_mirror]["TOML_REGISTRY"]}" + +[registries.{rust_mirror}] +index = "sparse+{RUSTUP_MIRRORS[rust_mirror]["TOML_REGISTRY"]}" +""", + ), + # 下载 Rustup 安装脚本 + px.TaskSpec( + "download_rustup", + cmd=["curl", "-fsSL", RUSTUP_DOWNLOAD_URL_LINUX, "-o", "rustup-init.sh"], + conditions=(BuiltinConditions.IS_LINUX(), BuiltinConditions.NOT(BuiltinConditions.HAS_INSTALLED("rustup"))), + verbose=True, + ), + px.TaskSpec( + "download_rustup_win", + cmd=[ + "powershell", + "-Command", + "Invoke-WebRequest", + "-Uri", + RUSTUP_DOWNLOAD_URL_WINDOWS, + "-OutFile", + "rustup-init.exe", + ], + conditions=( + BuiltinConditions.IS_WINDOWS(), + BuiltinConditions.NOT(BuiltinConditions.HAS_INSTALLED("rustup")), + ), + verbose=True, + ), + # 安装 Rust 工具链 + px.TaskSpec( + "install_rust", + cmd=["rustup", "toolchain", "install", rust_version], + conditions=(BuiltinConditions.HAS_INSTALLED("rustup"),), + depends_on=("setenv_rustup_dist_server",), + allow_upstream_skip=True, + verbose=True, + ), ]) px.run(graph, strategy="thread", verbose=True) diff --git a/src/pyflowx/cli/envpy.py b/src/pyflowx/cli/envpy.py deleted file mode 100644 index 3f51238..0000000 --- a/src/pyflowx/cli/envpy.py +++ /dev/null @@ -1,122 +0,0 @@ -"""Python 环境配置工具. - -用于设置 pip 镜像源, 支持清华和阿里云等国内镜像源, -同时配置 UV 和 Conda 的镜像源. -""" - -from __future__ import annotations - -import argparse -import os -from pathlib import Path - -import pyflowx as px -from pyflowx.conditions import Constants - -# ============================================================================ -# 配置 -# ============================================================================ - -PIP_INDEX_URLS: dict[str, str] = { - "tsinghua": "https://pypi.tuna.tsinghua.edu.cn/simple", - "aliyun": "https://mirrors.aliyun.com/pypi/simple/", -} - -PIP_TRUSTED_HOSTS: dict[str, str] = { - "tsinghua": "pypi.tuna.tsinghua.edu.cn", - "aliyun": "mirrors.aliyun.com", -} - -UV_INDEX_URL: str = "https://mirrors.aliyun.com/pypi/simple/" -UV_PYTHON_INSTALL_MIRROR: str = "https://registry.npmmirror.com/-/binary/python-build-standalone" - -CONDA_MIRROR_URLS: dict[str, list[str]] = { - "tsinghua": [ - "https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/", - "https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/", - "https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge/", - ], - "aliyun": [ - "https://mirrors.aliyun.com/anaconda/pkgs/main/", - "https://mirrors.aliyun.com/anaconda/pkgs/free/", - "https://mirrors.aliyun.com/anaconda/cloud/conda-forge/", - ], -} - - -# ============================================================================ -# 辅助函数 -# ============================================================================ - - -def set_pip_mirror(mirror: str = "tsinghua", token: str | None = None) -> None: - """设置 pip 镜像源. - - Parameters - ---------- - mirror : str - 镜像源名称: tsinghua, aliyun - token : str | None - PyPI token for publishing - """ - index_url = PIP_INDEX_URLS.get(mirror, PIP_INDEX_URLS["tsinghua"]) - trusted_host = PIP_TRUSTED_HOSTS.get(mirror, "") - - # 设置环境变量 - os.environ["PIP_INDEX_URL"] = index_url - os.environ["UV_INDEX_URL"] = UV_INDEX_URL - os.environ["UV_DEFAULT_INDEX"] = UV_INDEX_URL - os.environ["UV_PYTHON_INSTALL_MIRROR"] = UV_PYTHON_INSTALL_MIRROR - - # 写入 pip 配置文件 - pip_dir = Path.home() / "pip" - pip_dir.mkdir(exist_ok=True) - pip_conf = pip_dir / ("pip.ini" if Constants.IS_WINDOWS else "pip.conf") - pip_conf.write_text(f"[global]\nindex-url = {index_url}\n[install]\ntrusted-host = {trusted_host}\n") - - # 写入 conda 配置文件 - condarc = Path.home() / ".condarc" - conda_urls = CONDA_MIRROR_URLS.get(mirror, CONDA_MIRROR_URLS["tsinghua"]) - condarc.write_text( - "show_channel_urls: true\nchannels:\n" + "\n".join(f" - {url}" for url in conda_urls) + "\n - defaults\n" - ) - - # 写入 pypirc 配置文件 (如果有 token) - if token: - pypirc = Path.home() / ".pypirc" - pypirc.write_text( - f"[pypi]\nrepository: https://upload.pypi.org/legacy/\nusername: __token__\npassword: {token}\n" - ) - - print(f"已设置 pip 镜像源: {mirror} ({index_url})") - - -# ============================================================================ -# CLI Runner -# ============================================================================ - - -def main() -> None: - """Python 环境配置工具主函数.""" - parser = argparse.ArgumentParser( - description="EnvPy - Python 环境配置工具", - usage="envpy [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - # 设置镜像源命令 - mirror_parser = subparsers.add_parser("mirror", help="设置 pip 镜像源") - mirror_parser.add_argument("name", choices=["tsinghua", "aliyun"], help="镜像源名称") - mirror_parser.add_argument("--token", type=str, help="PyPI token for publishing") - - args = parser.parse_args() - - if args.command == "mirror": - graph = px.Graph.from_specs([ - px.TaskSpec("set_pip_mirror", fn=set_pip_mirror, args=(args.name,), kwargs={"token": args.token}) - ]) - else: - parser.print_help() - return - - px.run(graph, strategy="thread") diff --git a/src/pyflowx/cli/envrs.py b/src/pyflowx/cli/envrs.py deleted file mode 100644 index 3a45566..0000000 --- a/src/pyflowx/cli/envrs.py +++ /dev/null @@ -1,150 +0,0 @@ -"""Rust 环境配置工具. - -配置 Rustup 和 Cargo 的国内镜像源, -加速 Rust 工具链和依赖包的下载. -""" - -from __future__ import annotations - -import argparse -import os -import subprocess -from pathlib import Path -from typing import Literal, get_args - -import pyflowx as px - -# ============================================================================ -# 配置 -# ============================================================================ - -RUSTUP_MIRRORS: dict[str, dict[str, str]] = { - "aliyun": { - "RUSTUP_DIST_SERVER": "https://mirrors.aliyun.com/rustup", - "RUSTUP_UPDATE_ROOT": "https://mirrors.aliyun.com/rustup/rustup", - "TOML_REGISTRY": "https://mirrors.aliyun.com/crates.io-index/", - }, - "ustc": { - "RUSTUP_DIST_SERVER": "https://mirrors.ustc.edu.cn/rust-static", - "RUSTUP_UPDATE_ROOT": "https://mirrors.ustc.edu.cn/rust-static/rustup", - "TOML_REGISTRY": "https://mirrors.ustc.edu.cn/crates.io-index/", - }, - "tsinghua": { - "RUSTUP_DIST_SERVER": "https://mirrors.tuna.tsinghua.edu.cn/rustup", - "RUSTUP_UPDATE_ROOT": "https://mirrors.tuna.tsinghua.edu.cn/rustup/rustup", - "TOML_REGISTRY": "https://mirrors.tuna.tsinghua.edu.cn/crates.io-index/", - }, -} - -UsableRustVersion = Literal["stable", "nightly", "beta"] -UsableMirror = Literal["aliyun", "ustc", "tsinghua"] - -DEFAULT_RUST_VERSION: UsableRustVersion = "stable" -DEFAULT_MIRROR: UsableMirror = "tsinghua" - - -# ============================================================================ -# 辅助函数 -# ============================================================================ - - -def set_rust_mirror(mirror: UsableMirror = DEFAULT_MIRROR) -> None: - """设置 Rust 镜像源. - - Parameters - ---------- - mirror : str - 镜像源名称: aliyun, ustc, tsinghua - """ - mirror_dict = RUSTUP_MIRRORS.get(mirror, RUSTUP_MIRRORS[DEFAULT_MIRROR]) - server = mirror_dict["RUSTUP_DIST_SERVER"] - update_root = mirror_dict["RUSTUP_UPDATE_ROOT"] - toml_registry = mirror_dict["TOML_REGISTRY"] - - # 设置环境变量 - os.environ["RUSTUP_DIST_SERVER"] = server - os.environ["RUSTUP_UPDATE_ROOT"] = update_root - - # 写入 cargo 配置 - cargo_dir = Path.home() / ".cargo" - cargo_dir.mkdir(exist_ok=True) - cargo_config = cargo_dir / "config.toml" - cargo_config.write_text( - f"""[source.crates-io] -replace-with = '{mirror}' - -[source.{mirror}] -registry = "sparse+{toml_registry}" - -[registries.{mirror}] -index = "sparse+{toml_registry}" -""" - ) - - print(f"已设置 Rust 镜像源: {mirror}") - - -def install_rust(version: UsableRustVersion = DEFAULT_RUST_VERSION) -> None: - """安装 Rust 工具链. - - Parameters - ---------- - version : str - Rust 版本: stable, nightly, beta - """ - try: - subprocess.run(["rustup", "toolchain", "install", version], check=True) - print(f"已安装 Rust {version}") - except FileNotFoundError: - print("未找到 rustup，请先安装 Rust: https://rustup.rs") - raise - - -# ============================================================================ -# CLI Runner -# ============================================================================ - - -def main() -> None: - """Rust 环境配置工具主函数.""" - parser = argparse.ArgumentParser( - description="EnvRs - Rust 环境配置工具", - usage="envrs [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - # 设置镜像源命令 - mirror_parser = subparsers.add_parser("mirror", help="设置 Rust 镜像源") - mirror_parser.add_argument( - "name", - nargs="?", - default=DEFAULT_MIRROR, - choices=get_args(UsableMirror), - help=f"镜像源名称 ({get_args(UsableMirror)})", - ) - - # 安装 Rust 命令 - install_parser = subparsers.add_parser("install", help="安装 Rust 工具链") - install_parser.add_argument( - "version", - nargs="?", - default=DEFAULT_RUST_VERSION, - choices=get_args(UsableRustVersion), - help=f"Rust 版本 ({get_args(UsableRustVersion)})", - ) - - args = parser.parse_args() - - if args.command == "mirror": - graph = px.Graph.from_specs([ - px.TaskSpec("set_rust_mirror", fn=set_rust_mirror, args=(args.name,), verbose=True) - ]) - elif args.command == "install": - graph = px.Graph.from_specs([ - px.TaskSpec("install_rust", cmd=["rustup", "toolchain", "install", args.version], verbose=True) - ]) - else: - parser.print_help() - return - - px.run(graph, strategy="thread", verbose=True) diff --git a/src/pyflowx/graph.py b/src/pyflowx/graph.py index 0e8a673..0fddbd4 100644 --- a/src/pyflowx/graph.py +++ b/src/pyflowx/graph.py @@ -12,6 +12,11 @@ from __future__ import annotations +__all__ = [ + "Graph", + "GraphDefaults", +] + import sys from dataclasses import dataclass, field, replace from typing import Any, Callable, Iterable, Mapping, Sequence @@ -73,6 +78,7 @@ class Graph: specs: dict[str, TaskSpec[Any]] = field(default_factory=dict) deps: dict[str, tuple[str, ...]] = field(default_factory=dict) defaults: GraphDefaults = field(default_factory=GraphDefaults) + # 待解析的字符串引用列表（由 GraphComposer 消费）；为空表示无引用。 _pending_refs: list[str] = field(default_factory=list) diff --git a/tests/cli/test_envpy.py b/tests/cli/test_envpy.py deleted file mode 100644 index 3d95fab..0000000 --- a/tests/cli/test_envpy.py +++ /dev/null @@ -1,110 +0,0 @@ -"""Tests for cli.envpy module.""" - -from __future__ import annotations - -from pathlib import Path -from unittest.mock import patch - -import pytest - -import pyflowx as px -from pyflowx.cli import envpy - - -# ---------------------------------------------------------------------- # -# set_pip_mirror -# ---------------------------------------------------------------------- # -class TestSetPipMirror: - """Test set_pip_mirror function.""" - - def test_set_pip_mirror_tsinghua(self, tmp_path: Path) -> None: - """Should set tsinghua mirror.""" - with patch.object(Path, "home", return_value=tmp_path): - envpy.set_pip_mirror("tsinghua") - # Check pip config - pip_config = tmp_path / "pip" / "pip.ini" - if envpy.Constants.IS_WINDOWS: - assert pip_config.exists() or (tmp_path / "pip" / "pip.conf").exists() - - def test_set_pip_mirror_aliyun(self, tmp_path: Path) -> None: - """Should set aliyun mirror.""" - with patch.object(Path, "home", return_value=tmp_path): - envpy.set_pip_mirror("aliyun") - # Check pip config - pip_dir = tmp_path / "pip" - assert pip_dir.exists() - - def test_set_pip_mirror_with_token(self, tmp_path: Path) -> None: - """Should set mirror with token.""" - with patch.object(Path, "home", return_value=tmp_path): - envpy.set_pip_mirror("tsinghua", token="test_token") - # Check that token is set - - def test_set_pip_mirror_creates_pip_dir(self, tmp_path: Path) -> None: - """Should create pip directory if it doesn't exist.""" - pip_dir = tmp_path / "pip" - with patch.object(Path, "home", return_value=tmp_path): - envpy.set_pip_mirror("tsinghua") - assert pip_dir.exists() - assert pip_dir.is_dir() - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_mirror_tsinghua(self) -> None: - """main() should handle mirror tsinghua command.""" - with patch("sys.argv", ["envpy", "mirror", "tsinghua"]), patch.object(px, "run") as mock_run, patch.object( - envpy, "set_pip_mirror" - ): - envpy.main() - assert mock_run.called - - def test_main_mirror_aliyun(self) -> None: - """main() should handle mirror aliyun command.""" - with patch("sys.argv", ["envpy", "mirror", "aliyun"]), patch.object(px, "run") as mock_run, patch.object( - envpy, "set_pip_mirror" - ): - envpy.main() - assert mock_run.called - - def test_main_mirror_with_token(self) -> None: - """main() should handle mirror with token.""" - with patch("sys.argv", ["envpy", "mirror", "tsinghua", "--token", "test_token"]), patch.object( - px, "run" - ) as mock_run, patch.object(envpy, "set_pip_mirror"): - envpy.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help and return.""" - with patch("sys.argv", ["envpy"]): - envpy.main() - # Should print help and return - - def test_main_invalid_mirror_shows_error(self) -> None: - """main() with invalid mirror should show error.""" - with patch("sys.argv", ["envpy", "mirror", "invalid"]), pytest.raises(SystemExit) as exc_info: - envpy.main() - assert exc_info.value.code == 2 - - def test_main_creates_task_spec_with_correct_name(self) -> None: - """main() should create TaskSpec with correct name.""" - with patch("sys.argv", ["envpy", "mirror", "tsinghua"]), patch.object(px, "run") as mock_run, patch.object( - envpy, "set_pip_mirror" - ): - envpy.main() - graph = mock_run.call_args[0][0] - task_names = list(graph.all_specs().keys()) - assert "set_pip_mirror" in task_names - - def test_main_uses_thread_strategy(self) -> None: - """main() should use thread strategy.""" - with patch("sys.argv", ["envpy", "mirror", "tsinghua"]), patch.object(px, "run") as mock_run, patch.object( - envpy, "set_pip_mirror" - ): - envpy.main() - assert mock_run.call_args[1]["strategy"] == "thread" diff --git a/tests/cli/test_envrs.py b/tests/cli/test_envrs.py deleted file mode 100644 index ce3d086..0000000 --- a/tests/cli/test_envrs.py +++ /dev/null @@ -1,210 +0,0 @@ -"""Tests for cli.envrs module.""" - -from __future__ import annotations - -import os -from pathlib import Path -from unittest.mock import MagicMock, patch - -import pytest - -import pyflowx as px -from pyflowx.cli import envrs - - -# ---------------------------------------------------------------------- # -# set_rust_mirror -# ---------------------------------------------------------------------- # -class TestSetRustMirror: - """Test set_rust_mirror function.""" - - def test_set_rust_mirror_aliyun(self, tmp_path: Path) -> None: - """Should set aliyun mirror.""" - with patch.object(Path, "home", return_value=tmp_path): - envrs.set_rust_mirror("aliyun") - # Check environment variables - assert os.environ.get("RUSTUP_DIST_SERVER") == "https://mirrors.aliyun.com/rustup" - assert os.environ.get("RUSTUP_UPDATE_ROOT") == "https://mirrors.aliyun.com/rustup/rustup" - # Check cargo config - cargo_config = tmp_path / ".cargo" / "config.toml" - assert cargo_config.exists() - content = cargo_config.read_text() - assert "aliyun" in content - - def test_set_rust_mirror_ustc(self, tmp_path: Path) -> None: - """Should set ustc mirror.""" - with patch.object(Path, "home", return_value=tmp_path): - envrs.set_rust_mirror("ustc") - assert os.environ.get("RUSTUP_DIST_SERVER") == "https://mirrors.ustc.edu.cn/rust-static" - assert os.environ.get("RUSTUP_UPDATE_ROOT") == "https://mirrors.ustc.edu.cn/rust-static/rustup" - - def test_set_rust_mirror_tsinghua(self, tmp_path: Path) -> None: - """Should set tsinghua mirror.""" - with patch.object(Path, "home", return_value=tmp_path): - envrs.set_rust_mirror("tsinghua") - assert os.environ.get("RUSTUP_DIST_SERVER") == "https://mirrors.tuna.tsinghua.edu.cn/rustup" - assert os.environ.get("RUSTUP_UPDATE_ROOT") == "https://mirrors.tuna.tsinghua.edu.cn/rustup/rustup" - - def test_set_rust_mirror_unknown_uses_default(self, tmp_path: Path) -> None: - """Should use default mirror for unknown mirror name.""" - with patch.object(Path, "home", return_value=tmp_path): - # pyrefly: ignore [bad-argument-type] - envrs.set_rust_mirror("unknown") - # Should use default mirror (tsinghua) - assert os.environ.get("RUSTUP_DIST_SERVER") == "https://mirrors.tuna.tsinghua.edu.cn/rustup" - - def test_set_rust_mirror_creates_cargo_dir(self, tmp_path: Path) -> None: - """Should create .cargo directory if it doesn't exist.""" - cargo_dir = tmp_path / ".cargo" - with patch.object(Path, "home", return_value=tmp_path): - envrs.set_rust_mirror("aliyun") - assert cargo_dir.exists() - assert cargo_dir.is_dir() - - def test_set_rust_mirror_prints_message(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: - """Should print mirror name.""" - with patch.object(Path, "home", return_value=tmp_path): - envrs.set_rust_mirror("aliyun") - captured = capsys.readouterr() - assert "已设置 Rust 镜像源: aliyun" in captured.out - - -# ---------------------------------------------------------------------- # -# install_rust -# ---------------------------------------------------------------------- # -class TestInstallRust: - """Test install_rust function.""" - - def test_install_rust_stable(self) -> None: - """Should install stable Rust.""" - with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock(returncode=0) - envrs.install_rust("stable") - mock_run.assert_called_once_with(["rustup", "toolchain", "install", "stable"], check=True) - - def test_install_rust_nightly(self) -> None: - """Should install nightly Rust.""" - with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock(returncode=0) - envrs.install_rust("nightly") - mock_run.assert_called_once_with(["rustup", "toolchain", "install", "nightly"], check=True) - - def test_install_rust_beta(self) -> None: - """Should install beta Rust.""" - with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock(returncode=0) - envrs.install_rust("beta") - mock_run.assert_called_once_with(["rustup", "toolchain", "install", "beta"], check=True) - - def test_install_rust_file_not_found(self) -> None: - """Should raise FileNotFoundError when rustup not found.""" - with patch("subprocess.run", side_effect=FileNotFoundError), pytest.raises(FileNotFoundError): - envrs.install_rust("stable") - - def test_install_rust_prints_message(self, capsys: pytest.CaptureFixture[str]) -> None: - """Should print installation message.""" - with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock(returncode=0) - envrs.install_rust("stable") - captured = capsys.readouterr() - assert "已安装 Rust stable" in captured.out - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_mirror_aliyun(self) -> None: - """main() should handle mirror aliyun command.""" - with patch("sys.argv", ["envrs", "mirror", "aliyun"]), patch.object(px, "run") as mock_run, patch.object( - envrs, "set_rust_mirror" - ): - envrs.main() - assert mock_run.called - - def test_main_mirror_ustc(self) -> None: - """main() should handle mirror ustc command.""" - with patch("sys.argv", ["envrs", "mirror", "ustc"]), patch.object(px, "run") as mock_run, patch.object( - envrs, "set_rust_mirror" - ): - envrs.main() - assert mock_run.called - - def test_main_mirror_tsinghua(self) -> None: - """main() should handle mirror tsinghua command.""" - with patch("sys.argv", ["envrs", "mirror", "tsinghua"]), patch.object(px, "run") as mock_run, patch.object( - envrs, "set_rust_mirror" - ): - envrs.main() - assert mock_run.called - - def test_main_mirror_default(self) -> None: - """main() should use default mirror when not specified.""" - with patch("sys.argv", ["envrs", "mirror"]), patch.object(px, "run") as mock_run, patch.object( - envrs, "set_rust_mirror" - ): - envrs.main() - assert mock_run.called - - def test_main_install_stable(self) -> None: - """main() should handle install stable command.""" - with patch("sys.argv", ["envrs", "install", "stable"]), patch.object(px, "run") as mock_run: - envrs.main() - assert mock_run.called - - def test_main_install_nightly(self) -> None: - """main() should handle install nightly command.""" - with patch("sys.argv", ["envrs", "install", "nightly"]), patch.object(px, "run") as mock_run: - envrs.main() - assert mock_run.called - - def test_main_install_beta(self) -> None: - """main() should handle install beta command.""" - with patch("sys.argv", ["envrs", "install", "beta"]), patch.object(px, "run") as mock_run: - envrs.main() - assert mock_run.called - - def test_main_install_default(self) -> None: - """main() should use default version when not specified.""" - with patch("sys.argv", ["envrs", "install"]), patch.object(px, "run") as mock_run: - envrs.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help and return.""" - with patch("sys.argv", ["envrs"]): - envrs.main() - # Should print help and return - - def test_main_invalid_version_shows_error(self) -> None: - """main() with invalid version should show error.""" - with patch("sys.argv", ["envrs", "install", "invalid"]), pytest.raises(SystemExit) as exc_info: - envrs.main() - assert exc_info.value.code == 2 - - def test_main_invalid_mirror_shows_error(self) -> None: - """main() with invalid mirror should show error.""" - with patch("sys.argv", ["envrs", "mirror", "invalid"]), pytest.raises(SystemExit) as exc_info: - envrs.main() - assert exc_info.value.code == 2 - - def test_main_creates_task_spec_with_verbose(self) -> None: - """main() should create TaskSpec with verbose=True.""" - with patch("sys.argv", ["envrs", "mirror", "aliyun"]), patch.object(px, "run") as mock_run, patch.object( - envrs, "set_rust_mirror" - ): - envrs.main() - graph = mock_run.call_args[0][0] - specs = graph.all_specs() - for spec in specs.values(): - assert spec.verbose is True - - def test_main_uses_thread_strategy(self) -> None: - """main() should use thread strategy.""" - with patch("sys.argv", ["envrs", "mirror", "aliyun"]), patch.object(px, "run") as mock_run, patch.object( - envrs, "set_rust_mirror" - ): - envrs.main() - assert mock_run.call_args[1]["strategy"] == "thread"