From 53e073fde464d7761ff1e3852845d6d40b9504a4 Mon Sep 17 00:00:00 2001 From: rex <1073853456@qq.com> Date: Tue, 14 Apr 2026 18:22:56 +0800 Subject: [PATCH 1/3] feat: add mathlib and toolchains cache commands --- README.md | 61 +++++++-- docs/getting-started/quickstart.md | 51 +++++-- docs/index.md | 14 +- leanup/cli/__init__.py | 22 +-- leanup/cli/cache_ops.py | 48 +++++-- leanup/cli/toolchains.py | 141 +++++++++++++++++++ leanup/repo/cache_server.py | 18 +++ leanup/repo/elan.py | 8 +- leanup/repo/toolchain_cache.py | 211 +++++++++++++++++++++++++++++ leanup/utils/basic.py | 8 +- tests/test_cli.py | 7 +- tests/test_mathlib_cache_cli.py | 90 +++++++----- tests/test_toolchains_cli.py | 130 ++++++++++++++++++ 13 files changed, 714 insertions(+), 95 deletions(-) create mode 100644 leanup/cli/toolchains.py create mode 100644 leanup/repo/toolchain_cache.py create mode 100644 tests/test_toolchains_cli.py diff --git a/README.md b/README.md index 84717eb..1f0686a 100644 --- a/README.md +++ b/README.md @@ -88,23 +88,23 @@ leanup repo list -n mathlib ### 快速初始化项目 -`leanup setup` 用于快速创建一个固定 Lean 版本的项目，并按需要为 `mathlib` 依赖准备共享缓存。 +`leanup mathlib setup` 用于快速创建一个固定 Lean 版本的项目，并按需要为 `mathlib` 依赖准备共享缓存。 ```bash # 创建一个带 mathlib 的项目，默认有缓存就复用，没有缓存就自动准备缓存 -leanup setup ./Demo --lean-version v4.27.0 +leanup mathlib setup ./Demo --lean-version v4.27.0 # 使用 copy 模式，把共享缓存复制到项目里 -leanup setup ./DemoCopy --lean-version v4.27.0 --dependency-mode copy +leanup mathlib setup ./DemoCopy --lean-version v4.27.0 --dependency-mode copy # 后续同版本项目可直接软链接复用缓存 -leanup setup ./DemoFast --lean-version v4.27.0 --dependency-mode symlink +leanup mathlib setup ./DemoFast --lean-version v4.27.0 --dependency-mode symlink # 创建不带 mathlib 的纯 Lean 项目 -leanup setup ./PlainDemo --lean-version v4.27.0 --no-mathlib +leanup mathlib setup ./PlainDemo --lean-version v4.27.0 --no-mathlib # 指定 Lake 项目名，并覆盖已存在目录 -leanup setup ./Demo --lean-version v4.27.0 --name MyDemo --force +leanup mathlib setup ./Demo --lean-version v4.27.0 --name MyDemo --force ``` 规则说明： @@ -119,34 +119,37 @@ leanup setup ./Demo --lean-version v4.27.0 --name MyDemo --force ```bash # 查看 LeanUp 已有缓存版本 -leanup cache list +leanup mathlib list --local # 查看远端服务已有缓存版本和下载 URL -leanup cache list --base-url http://127.0.0.1:8000 +leanup mathlib list --remote http://127.0.0.1:8000 # 在临时目录创建某个 Lean 版本对应的共享 mathlib packages 缓存 -leanup cache create v4.22.0 +leanup mathlib create v4.22.0 # 将本地缓存里的 packages//packages 打包成 archives//packages.tar.gz -leanup cache pack v4.22.0 +leanup mathlib pack v4.22.0 + +# 将本地 archive 解压回 packages 目录 +leanup mathlib unpack v4.22.0 # 或者使用指定缓存根 -leanup cache pack v4.22.0 --output-dir /path/to/cache +leanup mathlib pack v4.22.0 --output-dir /path/to/cache # 从 LeanUp cache 服务下载 packages.tar.gz，并解压到本地缓存根 -leanup cache get v4.22.0 --base-url http://127.0.0.1:8000 +leanup mathlib get v4.22.0 --remote http://127.0.0.1:8000 # 启动本地缓存服务： # - /f/... 路由给 lake exe cache get 使用 # - /packages/... 路由给 leanup cache get 使用 -leanup cache serve +leanup serve # 让 mathlib 官方 cache client 改走 LeanUp 服务 export MATHLIB_CACHE_GET_URL=http://127.0.0.1:8000 lake exe cache get # 如需关闭并发压缩，可以显式禁用 pigz -leanup cache pack v4.22.0 --output-dir /path/to/cache --no-pigz +leanup mathlib pack v4.22.0 --output-dir /path/to/cache --no-pigz ``` - 默认会在本机存在 `pigz` 时启用并发压缩 @@ -158,6 +161,36 @@ leanup cache pack v4.22.0 --output-dir /path/to/cache --no-pigz - `leanup cache get` 和 `leanup cache pack` 都通过临时文件 / 临时目录完成后再原子替换，避免中途中断破坏缓存 - `leanup cache serve` 使用 FastAPI/uvicorn 提供服务，并暴露 `/packages/mathlib/index.json` 让其他机器列出远端可用版本 +### 管理 toolchains 缓存 + +```bash +# 查看本地 toolchain 归档 +leanup toolchains list --local + +# 查看远端 toolchain 归档 +leanup toolchains list --remote http://127.0.0.1:8000 + +# 不指定版本时，打包裸的 .elan 基础目录，不包含 toolchains/ +leanup toolchains pack + +# 指定版本时，打包 .elan/toolchains 中对应的 Lean toolchain +leanup toolchains pack v4.28.0 + +# 从本地归档解压 toolchain 到 .elan/toolchains +leanup toolchains unpack v4.28.0 + +# 初始化 .elan：不加 url 时走官方 elan 安装；加 url 时下载 base .elan 归档 +leanup toolchains init +leanup toolchains init --url http://127.0.0.1:8000 + +# 从远端下载并解压指定 toolchain +leanup toolchains get v4.28.0 --remote http://127.0.0.1:8000 +``` + +- toolchain archives 单独存储在 `LEANUP_CACHE_DIR/toolchains/archives` +- 解压后的目录语义沿用 `.elan/` +- 所有下载、压缩、解压都先写临时文件 / 临时目录，成功后再原子替换正式路径 + ### 交互式安装 使用 `leanup repo install -i` 时，您可以配置： diff --git a/docs/getting-started/quickstart.md b/docs/getting-started/quickstart.md index e793bfc..bb55f29 100644 --- a/docs/getting-started/quickstart.md +++ b/docs/getting-started/quickstart.md @@ -24,16 +24,16 @@ leanup --help ```bash # 创建一个带 mathlib 的 Lean 项目，默认有缓存就复用，没有缓存就自动准备缓存 -leanup setup ./Demo --lean-version v4.27.0 +leanup mathlib setup ./Demo --lean-version v4.27.0 # 使用 copy 模式，把共享缓存复制到项目里 -leanup setup ./DemoCopy --lean-version v4.27.0 --dependency-mode copy +leanup mathlib setup ./DemoCopy --lean-version v4.27.0 --dependency-mode copy # 后续项目直接复用共享依赖缓存 -leanup setup ./DemoFast --lean-version v4.27.0 --dependency-mode symlink +leanup mathlib setup ./DemoFast --lean-version v4.27.0 --dependency-mode symlink # 创建不依赖 mathlib 的纯 Lean 项目 -leanup setup ./PlainDemo --lean-version v4.27.0 --no-mathlib +leanup mathlib setup ./PlainDemo --lean-version v4.27.0 --no-mathlib ``` 说明： @@ -50,32 +50,35 @@ leanup setup ./PlainDemo --lean-version v4.27.0 --no-mathlib ```bash # 查看 LeanUp 已有缓存版本 -leanup cache list +leanup mathlib list --local # 查看远端服务已有缓存版本和下载 URL -leanup cache list --base-url http://127.0.0.1:8000 +leanup mathlib list --remote http://127.0.0.1:8000 # 在 tempfile 临时工作目录中创建某个 Lean 版本的共享 mathlib packages 缓存 -leanup cache create v4.22.0 +leanup mathlib create v4.22.0 # 将本地缓存里的 packages//packages 打包成 archives//packages.tar.gz -leanup cache pack v4.22.0 +leanup mathlib pack v4.22.0 + +# 将本地 archive 解压回 packages 目录 +leanup mathlib unpack v4.22.0 # 或者使用指定缓存根 -leanup cache pack v4.22.0 --output-dir /path/to/cache +leanup mathlib pack v4.22.0 --output-dir /path/to/cache # 启动缓存服务：/f/... 给 lake exe cache get，/packages/... 给 leanup cache get -leanup cache serve +leanup serve # 让 mathlib 官方 cache client 改走 LeanUp 服务 export MATHLIB_CACHE_GET_URL=http://127.0.0.1:8000 lake exe cache get # 从 LeanUp cache 服务下载 packages.tar.gz，并解压到本地缓存根 -leanup cache get v4.22.0 --base-url http://127.0.0.1:8000 +leanup mathlib get v4.22.0 --remote http://127.0.0.1:8000 # 如需关闭并发压缩，可以显式禁用 pigz -leanup cache pack v4.22.0 --output-dir /path/to/cache --no-pigz +leanup mathlib pack v4.22.0 --output-dir /path/to/cache --no-pigz ``` - 默认会在本机存在 `pigz` 时启用并发压缩 @@ -88,6 +91,30 @@ leanup cache pack v4.22.0 --output-dir /path/to/cache --no-pigz - `leanup cache get` 从远端下载 `packages.tar.gz` 到 `mathlib/archives//packages.tar.gz`，并解压到 `mathlib/packages//packages` - `leanup cache pack` 和 `leanup cache get` 都先写临时文件 / 临时目录，成功后再原子替换正式路径，避免中断损坏缓存 +### 管理 toolchains 缓存 + +```bash +# 查看本地和远端 toolchain 归档 +leanup toolchains list --local +leanup toolchains list --remote http://127.0.0.1:8000 + +# 打包裸的 .elan 基础目录，不包含 toolchains/ +leanup toolchains pack + +# 打包、解压、下载具体 Lean toolchain +leanup toolchains pack v4.28.0 +leanup toolchains unpack v4.28.0 +leanup toolchains get v4.28.0 --remote http://127.0.0.1:8000 + +# 初始化 .elan：默认走官方 elan；加 url 时下载 base .elan 归档 +leanup toolchains init +leanup toolchains init --url http://127.0.0.1:8000 +``` + +- toolchain archives 单独存储在 `LEANUP_CACHE_DIR/toolchains/archives` +- 解压后的目录语义沿用 `.elan/` +- 所有下载、压缩、解压都先写临时文件 / 临时目录，成功后再原子替换正式路径 + ### 仓库管理 ```bash diff --git a/docs/index.md b/docs/index.md index 4cb4473..bcd36d8 100644 --- a/docs/index.md +++ b/docs/index.md @@ -4,12 +4,14 @@ ## 功能特性 -- `leanup setup`：快速创建固定 Lean 版本项目，支持 mathlib 共享缓存 -- `leanup cache list`：查看本地或远端 mathlib 共享缓存版本 -- `leanup cache create `：在临时目录创建某个 Lean 版本的 mathlib packages 缓存 -- `leanup cache pack `：将本地缓存里的 packages 目录打包为共享缓存归档 -- `leanup cache get `：从 LeanUp cache 服务下载 `packages.tar.gz` 并解压到本地缓存 -- `leanup cache serve`：提供 `.ltar` 兼容路由和 LeanUp packages 归档下载服务 +- `leanup mathlib setup`：快速创建固定 Lean 版本项目，支持 mathlib 共享缓存 +- `leanup mathlib list`：查看本地或远端 mathlib 共享缓存版本 +- `leanup mathlib create `：在临时目录创建某个 Lean 版本的 mathlib packages 缓存 +- `leanup mathlib pack `：将本地缓存里的 packages 目录打包为共享缓存归档 +- `leanup mathlib unpack `：从本地 archive 解压回 packages 目录 +- `leanup mathlib get `：从 LeanUp 服务下载 `packages.tar.gz` 并解压到本地缓存 +- `leanup serve`：提供 `.ltar` 兼容路由和 LeanUp packages 归档下载服务 +- `leanup toolchains`：管理 `.elan` 基础包和 Lean toolchain 归档 - `leanup repo install`：安装 Lean 仓库，支持命令优先、交互补参 - `leanup repo list`：查看已安装仓库 diff --git a/leanup/cli/__init__.py b/leanup/cli/__init__.py index 70564eb..a63d89c 100644 --- a/leanup/cli/__init__.py +++ b/leanup/cli/__init__.py @@ -1,8 +1,9 @@ import click -from leanup.cli.cache_ops import create_cache, get_cache, list_cache, pack_cache, serve_cache +from leanup.cli.cache_ops import create_cache, get_cache, list_cache, pack_cache, serve_cache, unpack_cache from leanup.cli.repo import repo from leanup.cli.setup import setup_project +from leanup.cli.toolchains import toolchains from leanup.utils.custom_logger import setup_logger logger = setup_logger("leanup_cli") @@ -17,19 +18,22 @@ def cli(ctx): @click.group() -def cache() -> None: - """Manage reusable caches.""" +def mathlib() -> None: + """Manage mathlib projects and package caches.""" -cache.add_command(serve_cache) -cache.add_command(pack_cache) -cache.add_command(list_cache) -cache.add_command(get_cache) -cache.add_command(create_cache) +mathlib.add_command(setup_project) +mathlib.add_command(pack_cache) +mathlib.add_command(unpack_cache) +mathlib.add_command(list_cache) +mathlib.add_command(get_cache) +mathlib.add_command(create_cache) cli.add_command(setup_project) -cli.add_command(cache) +cli.add_command(mathlib) +cli.add_command(serve_cache) +cli.add_command(toolchains) cli.add_command(repo) diff --git a/leanup/cli/cache_ops.py b/leanup/cli/cache_ops.py index bff83dd..2b05425 100644 --- a/leanup/cli/cache_ops.py +++ b/leanup/cli/cache_ops.py @@ -47,42 +47,72 @@ def pack_cache(lean_version: str, output_dir: Path, pigz: bool) -> None: click.echo(str(packed)) +@click.command(name="unpack") +@click.argument("lean_version") +@click.option( + "--cache-dir", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=PACKAGES_CACHE_ROOT, + help="Mathlib cache root containing packages//packages and archives//packages.tar.gz.", +) +def unpack_cache(lean_version: str, cache_dir: Path) -> None: + """Unpack archives//packages.tar.gz into packages//packages.""" + manager = MathlibCacheManager(cache_root=cache_dir) + version = normalize_lean_version(lean_version) + archive = manager.get_local_archive_path(version) + + try: + packages_dir = manager.extract_archive(archive, manager.get_local_packages_dir(version)) + except ValueError as exc: + raise click.ClickException(str(exc)) from exc + + click.echo(str(packages_dir)) + + @click.command(name="list") @click.option( - "--base-url", - help="Print packages.tar.gz URLs using this base URL, for example http://127.0.0.1:8000.", + "--local", + "source", + flag_value="local", + default=True, + help="List local mathlib package caches.", +) +@click.option( + "--remote", + "remote_url", + help="List remote mathlib package caches using this base URL.", ) -def list_cache(base_url: str | None) -> None: +def list_cache(source: str, remote_url: str | None) -> None: """List available mathlib package caches.""" manager = MathlibCacheManager() - entries = manager.list_remote_entries(base_url) if base_url else manager.list_entries() + entries = manager.list_remote_entries(remote_url) if remote_url else manager.list_entries() if not entries: click.echo("No mathlib caches found.") return for entry in entries: - if base_url: - click.echo(f"{entry.version} {manager.build_archive_url(entry.version, base_url)}") + if remote_url: + click.echo(f"{entry.version} {manager.build_archive_url(entry.version, remote_url)}") else: click.echo(entry.version) @click.command(name="get") @click.argument("lean_version") -@click.option("--base-url", required=True, help="Base URL serving /packages/mathlib//packages.tar.gz.") +@click.option("--remote", "remote_url", required=True, help="Base URL serving /packages/mathlib//packages.tar.gz.") @click.option( "--cache-dir", type=click.Path(path_type=Path, file_okay=False, dir_okay=True), default=PACKAGES_CACHE_ROOT, help="Mathlib cache root containing packages//packages and archives//packages.tar.gz.", ) -def get_cache(lean_version: str, base_url: str, cache_dir: Path) -> None: +def get_cache(lean_version: str, remote_url: str, cache_dir: Path) -> None: """Download packages.tar.gz into local cache and extract packages//packages.""" manager = MathlibCacheManager(cache_root=cache_dir) try: - packages_dir = manager.fetch_packages(lean_version, base_url) + packages_dir = manager.fetch_packages(lean_version, remote_url) except (ValueError, requests.RequestException) as exc: raise click.ClickException(str(exc)) from exc diff --git a/leanup/cli/toolchains.py b/leanup/cli/toolchains.py new file mode 100644 index 0000000..e83f12f --- /dev/null +++ b/leanup/cli/toolchains.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +from pathlib import Path +import tempfile + +import click +import requests + +from leanup.const import LEANUP_CACHE_DIR +from leanup.repo.toolchain_cache import ToolchainCacheManager + + +TOOLCHAIN_CACHE_ROOT = LEANUP_CACHE_DIR / "toolchains" + + +@click.group(name="toolchains") +def toolchains() -> None: + """Manage elan and Lean toolchain archives.""" + + +@toolchains.command(name="list") +@click.option("--local", "mode", flag_value="local", default=True, help="List local toolchain archives.") +@click.option("--remote", "remote_url", help="List remote toolchain archives using this base URL.") +@click.option( + "--cache-dir", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=TOOLCHAIN_CACHE_ROOT, + help="Toolchain cache root.", +) +def list_toolchains(mode: str, remote_url: str | None, cache_dir: Path) -> None: + manager = ToolchainCacheManager(cache_root=cache_dir) + if remote_url: + has_base, versions = manager.list_remote(remote_url) + if has_base: + click.echo(f"base {manager.build_base_url(remote_url)}") + for version in versions: + click.echo(f"{version} {manager.build_toolchain_url(version, remote_url)}") + if not has_base and not versions: + click.echo("No toolchain archives found.") + return + + entries = manager.list_local_versions() + if manager.has_local_base_archive(): + click.echo("base") + for version in entries: + click.echo(version) + if not manager.has_local_base_archive() and not entries: + click.echo("No toolchain archives found.") + + +@toolchains.command(name="init") +@click.option("--url", help="Download base .elan archive from this service URL instead of official elan installer.") +@click.option( + "--cache-dir", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=TOOLCHAIN_CACHE_ROOT, + help="Toolchain cache root.", +) +@click.option( + "--elan-home", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=Path(tempfile.gettempdir()) / "leanup-elan", + help="Target .elan directory.", +) +def init_toolchains(url: str | None, cache_dir: Path, elan_home: Path) -> None: + manager = ToolchainCacheManager(cache_root=cache_dir, elan_home=elan_home) + try: + result = manager.init_base(url) + except (ValueError, RuntimeError, requests.RequestException) as exc: + raise click.ClickException(str(exc)) from exc + click.echo(str(result)) + + +@toolchains.command(name="pack") +@click.argument("lean_version", required=False) +@click.option( + "--cache-dir", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=TOOLCHAIN_CACHE_ROOT, + help="Toolchain cache root.", +) +@click.option( + "--elan-home", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=Path(tempfile.gettempdir()) / "leanup-elan", + help="Source .elan directory.", +) +def pack_toolchain(lean_version: str | None, cache_dir: Path, elan_home: Path) -> None: + manager = ToolchainCacheManager(cache_root=cache_dir, elan_home=elan_home) + try: + archive = manager.pack_base_archive() if lean_version is None else manager.pack_toolchain_archive(lean_version) + except ValueError as exc: + raise click.ClickException(str(exc)) from exc + click.echo(str(archive)) + + +@toolchains.command(name="unpack") +@click.argument("lean_version") +@click.option( + "--cache-dir", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=TOOLCHAIN_CACHE_ROOT, + help="Toolchain cache root.", +) +@click.option( + "--elan-home", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=Path(tempfile.gettempdir()) / "leanup-elan", + help="Target .elan directory.", +) +def unpack_toolchain(lean_version: str, cache_dir: Path, elan_home: Path) -> None: + manager = ToolchainCacheManager(cache_root=cache_dir, elan_home=elan_home) + try: + path = manager.unpack_toolchain_archive(lean_version) + except ValueError as exc: + raise click.ClickException(str(exc)) from exc + click.echo(str(path)) + + +@toolchains.command(name="get") +@click.argument("lean_version") +@click.option("--remote", required=True, help="Base URL serving /toolchains//toolchain.tar.gz.") +@click.option( + "--cache-dir", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=TOOLCHAIN_CACHE_ROOT, + help="Toolchain cache root.", +) +@click.option( + "--elan-home", + type=click.Path(path_type=Path, file_okay=False, dir_okay=True), + default=Path(tempfile.gettempdir()) / "leanup-elan", + help="Target .elan directory.", +) +def get_toolchain(lean_version: str, remote: str, cache_dir: Path, elan_home: Path) -> None: + manager = ToolchainCacheManager(cache_root=cache_dir, elan_home=elan_home) + try: + path = manager.fetch_toolchain(lean_version, remote) + except (ValueError, requests.RequestException) as exc: + raise click.ClickException(str(exc)) from exc + click.echo(str(path)) diff --git a/leanup/repo/cache_server.py b/leanup/repo/cache_server.py index 52f22e8..60f4c02 100644 --- a/leanup/repo/cache_server.py +++ b/leanup/repo/cache_server.py @@ -11,6 +11,7 @@ def create_cache_app(ltar_root: Path, packages_root: Path) -> FastAPI: ltar_root = ltar_root.resolve() packages_root = packages_root.resolve() archives_root = packages_root / "archives" + toolchain_archives_root = (packages_root.parent / "toolchains" / "archives").resolve() app = FastAPI(title="LeanUp Cache Server") @@ -31,6 +32,23 @@ def ltar_file(filename: str) -> FileResponse: def package_archive(version: str) -> FileResponse: return file_response(archives_root / version / "packages.tar.gz") + @app.get("/toolchains/index.json") + def toolchain_index() -> JSONResponse: + return JSONResponse( + { + "has_base": (toolchain_archives_root / "base-elan.tar.gz").exists(), + "versions": list_package_versions(toolchain_archives_root), + } + ) + + @app.get("/toolchains/base-elan.tar.gz") + def toolchain_base_archive() -> FileResponse: + return file_response(toolchain_archives_root / "base-elan.tar.gz") + + @app.get("/toolchains/{version}/toolchain.tar.gz") + def toolchain_archive(version: str) -> FileResponse: + return file_response(toolchain_archives_root / version / "toolchain.tar.gz") + return app diff --git a/leanup/repo/elan.py b/leanup/repo/elan.py index 67c4db1..91bb4f7 100644 --- a/leanup/repo/elan.py +++ b/leanup/repo/elan.py @@ -16,8 +16,8 @@ class ElanManager: """Elan toolchain manager""" - def __init__(self): - self.elan_home = Path(os.environ.get("ELAN_HOME", Path.home() / ".elan")) + def __init__(self, elan_home: Path | None = None): + self.elan_home = Path(elan_home or os.environ.get("ELAN_HOME", Path.home() / ".elan")) self.elan_bin_dir = self.elan_home / "bin" self._elan_exe = None @@ -133,7 +133,9 @@ def install_elan(self, force: bool = False) -> bool: logger.info("Running elan installation script...") cmd = ["sh", str(installer_path), "-y"] - output, error, code = execute_command(cmd, cwd=str(temp_dir)) + env = os.environ.copy() + env["ELAN_HOME"] = str(self.elan_home) + output, error, code = execute_command(cmd, cwd=str(temp_dir), env=env) if code != 0: logger.error(f"Installation failed: {error}") return False diff --git a/leanup/repo/toolchain_cache.py b/leanup/repo/toolchain_cache.py new file mode 100644 index 0000000..3b6bfd0 --- /dev/null +++ b/leanup/repo/toolchain_cache.py @@ -0,0 +1,211 @@ +from __future__ import annotations + +import os +from pathlib import Path +import tarfile +import tempfile +from urllib.parse import urljoin + +import requests + +from leanup.const import LEANUP_CACHE_DIR +from leanup.repo.elan import ElanManager +from leanup.repo.mathlib_cache import normalize_lean_version, remove_path +from leanup.utils.custom_logger import setup_logger + +logger = setup_logger("toolchain_cache") + + +class ToolchainCacheManager: + def __init__(self, cache_root: Path | None = None, elan_home: Path | None = None): + self.cache_root = cache_root or (LEANUP_CACHE_DIR / "toolchains") + self.archives_root = self.cache_root / "archives" + self.elan_home = elan_home or Path(os.environ.get("ELAN_HOME", Path.home() / ".elan")) + + def get_base_archive_path(self) -> Path: + return self.archives_root / "base-elan.tar.gz" + + def get_toolchain_archive_path(self, version: str) -> Path: + return self.archives_root / normalize_lean_version(version) / "toolchain.tar.gz" + + def list_local_versions(self) -> list[str]: + if not self.archives_root.exists(): + return [] + versions: list[str] = [] + for child in sorted(self.archives_root.iterdir()): + if child.is_dir() and (child / "toolchain.tar.gz").exists(): + versions.append(child.name) + return versions + + def has_local_base_archive(self) -> bool: + return self.get_base_archive_path().exists() + + def list_remote(self, base_url: str) -> tuple[bool, list[str]]: + index_url = urljoin(base_url.rstrip("/") + "/", "toolchains/index.json") + try: + response = requests.get(index_url, timeout=10) + response.raise_for_status() + payload = response.json() + except Exception: + return False, [] + versions = [normalize_lean_version(version) for version in payload.get("versions", [])] + has_base = bool(payload.get("has_base")) + return has_base, versions + + def build_base_url(self, base_url: str) -> str: + return urljoin(base_url.rstrip("/") + "/", "toolchains/base-elan.tar.gz") + + def build_toolchain_url(self, version: str, base_url: str) -> str: + normalized = normalize_lean_version(version) + return urljoin(base_url.rstrip("/") + "/", f"toolchains/{normalized}/toolchain.tar.gz") + + def download_base_archive(self, url: str) -> Path: + return self._download_to(url, self.get_base_archive_path()) + + def download_toolchain_archive(self, version: str, url: str) -> Path: + return self._download_to(url, self.get_toolchain_archive_path(version)) + + def init_base(self, url: str | None = None) -> Path: + if url: + archive = self.download_base_archive(self.build_base_url(url)) + return self.unpack_base_archive(archive) + + manager = ElanManager(elan_home=self.elan_home) + if not manager.install_elan(): + raise RuntimeError("Failed to install elan.") + self.pack_base_archive() + return self.elan_home + + def pack_base_archive(self) -> Path: + if not self.elan_home.exists(): + raise ValueError(f"Elan home not found: {self.elan_home}") + + output_file = self.get_base_archive_path() + output_file.parent.mkdir(parents=True, exist_ok=True) + temp_output = output_file.parent / f".{output_file.name}.tmp" + remove_path(temp_output) + + with tarfile.open(temp_output, "w:gz", dereference=False) as tar: + for child in sorted(self.elan_home.iterdir()): + if child.name == "toolchains": + continue + tar.add(child, arcname=f".elan/{child.name}", recursive=True) + + remove_path(output_file) + temp_output.replace(output_file) + logger.info(f"Packed base elan archive -> {output_file}") + return output_file + + def unpack_base_archive(self, archive_path: Path | None = None) -> Path: + archive_path = archive_path or self.get_base_archive_path() + temp_root = Path(tempfile.mkdtemp(prefix=".elan-base.", dir=self.elan_home.parent)) + try: + with tarfile.open(archive_path, "r:gz") as tar: + self._safe_extract(tar, temp_root) + extracted = temp_root / ".elan" + if not extracted.exists(): + raise ValueError(f"Archive does not contain top-level .elan/ directory: {archive_path}") + final_temp = self.elan_home.parent / f".{self.elan_home.name}.replace-{os.getpid()}" + remove_path(final_temp) + extracted.replace(final_temp) + remove_path(self.elan_home) + final_temp.replace(self.elan_home) + remove_path(temp_root) + logger.info(f"Unpacked base elan archive -> {self.elan_home}") + return self.elan_home + except Exception: + remove_path(temp_root) + raise + + def pack_toolchain_archive(self, version: str) -> Path: + toolchain_dir = self._resolve_installed_toolchain_dir(version) + if toolchain_dir is None: + raise ValueError(f"Installed toolchain not found for {normalize_lean_version(version)}") + + output_file = self.get_toolchain_archive_path(version) + output_file.parent.mkdir(parents=True, exist_ok=True) + temp_output = output_file.parent / f".{output_file.name}.tmp" + remove_path(temp_output) + + with tarfile.open(temp_output, "w:gz", dereference=False) as tar: + tar.add(toolchain_dir, arcname=f".elan/toolchains/{toolchain_dir.name}", recursive=True) + + remove_path(output_file) + temp_output.replace(output_file) + logger.info(f"Packed toolchain {version} -> {output_file}") + return output_file + + def unpack_toolchain_archive(self, version: str, archive_path: Path | None = None) -> Path: + archive_path = archive_path or self.get_toolchain_archive_path(version) + temp_root = Path(tempfile.mkdtemp(prefix=".elan-toolchain.", dir=self.elan_home.parent)) + try: + with tarfile.open(archive_path, "r:gz") as tar: + self._safe_extract(tar, temp_root) + toolchains_root = temp_root / ".elan" / "toolchains" + toolchain_dirs = [path for path in toolchains_root.iterdir() if path.is_dir()] if toolchains_root.exists() else [] + if len(toolchain_dirs) != 1: + raise ValueError(f"Archive must contain exactly one toolchain directory: {archive_path}") + source_dir = toolchain_dirs[0] + target_dir = self.elan_home / "toolchains" / source_dir.name + target_dir.parent.mkdir(parents=True, exist_ok=True) + final_temp = target_dir.parent / f".{target_dir.name}.replace-{os.getpid()}" + remove_path(final_temp) + source_dir.replace(final_temp) + remove_path(target_dir) + final_temp.replace(target_dir) + remove_path(temp_root) + logger.info(f"Unpacked toolchain archive -> {target_dir}") + return target_dir + except Exception: + remove_path(temp_root) + raise + + def fetch_toolchain(self, version: str, base_url: str) -> Path: + archive = self.download_toolchain_archive(version, self.build_toolchain_url(version, base_url)) + return self.unpack_toolchain_archive(version, archive) + + def _resolve_installed_toolchain_dir(self, version: str) -> Path | None: + normalized = normalize_lean_version(version) + toolchains_root = self.elan_home / "toolchains" + if not toolchains_root.exists(): + return None + for child in sorted(toolchains_root.iterdir()): + if child.is_dir() and normalized in child.name: + return child + return None + + def _download_to(self, url: str, output_file: Path) -> Path: + output_file.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + dir=output_file.parent, + prefix=f".{output_file.name}.", + suffix=".tmp", + delete=False, + ) as handle: + temp_output = Path(handle.name) + + try: + with requests.get(url, stream=True, timeout=120) as response: + response.raise_for_status() + with temp_output.open("wb") as output_handle: + for chunk in response.iter_content(chunk_size=1024 * 1024): + if chunk: + output_handle.write(chunk) + remove_path(output_file) + temp_output.replace(output_file) + logger.info(f"Downloaded {url} -> {output_file}") + return output_file + except Exception: + remove_path(temp_output) + raise + + def _safe_extract(self, tar: tarfile.TarFile, target_dir: Path) -> None: + target_dir = target_dir.resolve() + for member in tar.getmembers(): + member_path = (target_dir / member.name).resolve() + if not str(member_path).startswith(str(target_dir)): + raise ValueError(f"Archive contains unsafe path: {member.name}") + try: + tar.extractall(target_dir, filter="data") + except TypeError: + tar.extractall(target_dir) diff --git a/leanup/utils/basic.py b/leanup/utils/basic.py index e25a9bf..6b4a12f 100644 --- a/leanup/utils/basic.py +++ b/leanup/utils/basic.py @@ -2,7 +2,7 @@ import subprocess from leanup.const import OS_TYPE, TMP_DIR import shlex -from typing import Optional, Union, Tuple, List, Generator +from typing import Optional, Union, Tuple, List, Generator, Dict from contextlib import contextmanager from pathlib import Path import os @@ -12,7 +12,8 @@ def execute_command(command: Union[str, List[str]], text: bool = True, input: Union[str, None] = None, capture_output: bool = True, - timeout: Optional[int] = None) -> Tuple[str, str, int]: + timeout: Optional[int] = None, + env: Optional[Dict[str, str]] = None) -> Tuple[str, str, int]: """ Execute command with subprocess.Popen. @@ -40,7 +41,8 @@ def execute_command(command: Union[str, List[str]], stdout=stdout_pipe, stderr=stderr_pipe, shell=OS_TYPE == 'Windows', - text=text + text=text, + env=env, ) stdout, stderr = process.communicate(input=input, timeout=timeout) returncode = process.returncode diff --git a/tests/test_cli.py b/tests/test_cli.py index 01cdc30..5aeed74 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -15,12 +15,13 @@ def test_cli_help(self): assert result.exit_code == 0 assert "LeanUp - Lean project management tool" in result.output - def test_cache_help_lists_top_level_commands(self): - result = self.runner.invoke(cli, ["cache", "--help"]) + def test_mathlib_help_lists_top_level_commands(self): + result = self.runner.invoke(cli, ["mathlib", "--help"]) assert result.exit_code == 0 assert "list" in result.output assert "get" in result.output assert "pack" in result.output - assert "serve" in result.output assert "create" in result.output + assert "unpack" in result.output + assert "setup" in result.output diff --git a/tests/test_mathlib_cache_cli.py b/tests/test_mathlib_cache_cli.py index a7a5ff7..0b4e3af 100644 --- a/tests/test_mathlib_cache_cli.py +++ b/tests/test_mathlib_cache_cli.py @@ -22,7 +22,7 @@ def test_mathlib_cache_pack_archives_current_repo(tmp_path): result = runner.invoke( cli, [ - "cache", + "mathlib", "pack", "v4.22.0", "--output-dir", @@ -52,7 +52,7 @@ def test_mathlib_subcommand_pack_defaults_to_mathlib_cache_root(tmp_path): sys.executable, "-c", "from leanup.cli import cli; cli()", - "cache", + "mathlib", "pack", "v4.28.0", "--no-pigz", @@ -80,7 +80,7 @@ def test_mathlib_pack_output_is_listable_via_server_url(tmp_path): pack_result = runner.invoke( cli, [ - "cache", + "mathlib", "pack", "v4.28.0", "--output-dir", @@ -96,11 +96,10 @@ def test_mathlib_pack_output_is_listable_via_server_url(tmp_path): command = [ sys.executable, "-c", - "from leanup.cli import cli; cli()", - "cache", - "serve", - "--host", - "127.0.0.1", + "from leanup.cli import cli; cli()", + "serve", + "--host", + "127.0.0.1", "--port", "18083", "--ltar-root", @@ -122,7 +121,7 @@ def test_mathlib_pack_output_is_listable_via_server_url(tmp_path): proc.stdout.readline() time.sleep(1) base_url = "http://127.0.0.1:18083" - list_result = runner.invoke(cli, ["cache", "list", "--base-url", base_url]) + list_result = runner.invoke(cli, ["mathlib", "list", "--remote", base_url]) finally: proc.terminate() proc.wait(timeout=5) @@ -141,7 +140,7 @@ def test_mathlib_cache_pack_follows_root_packages_symlink(tmp_path): result = runner.invoke( cli, [ - "cache", + "mathlib", "pack", "v4.29.0", "--output-dir", @@ -178,7 +177,7 @@ def fake_pack(self, packages_dir, output_file, use_pigz=False): result = runner.invoke( cli, [ - "cache", + "mathlib", "pack", "v4.22.0", "--output-dir", @@ -213,7 +212,7 @@ def fake_pack(self, packages_dir, output_file, use_pigz=False): result = runner.invoke( cli, [ - "cache", + "mathlib", "pack", "v4.22.0", "--output-dir", @@ -239,9 +238,9 @@ def test_mathlib_cache_list_prints_package_urls(monkeypatch, tmp_path): result = runner.invoke( cli, [ - "cache", + "mathlib", "list", - "--base-url", + "--remote", "http://127.0.0.1:8000/", ], ) @@ -265,11 +264,10 @@ def test_mathlib_cache_list_reads_remote_index(tmp_path): command = [ sys.executable, "-c", - "from leanup.cli import cli; cli()", - "cache", - "serve", - "--host", - "127.0.0.1", + "from leanup.cli import cli; cli()", + "serve", + "--host", + "127.0.0.1", "--port", "18081", "--ltar-root", @@ -293,7 +291,7 @@ def test_mathlib_cache_list_reads_remote_index(tmp_path): base_url = "http://127.0.0.1:18081" result = runner.invoke( cli, - ["cache", "list", "--base-url", base_url], + ["mathlib", "list", "--remote", base_url], ) finally: proc.terminate() @@ -319,7 +317,7 @@ def fake_cache_init(self, cache_root_arg=None): result = runner.invoke( cli, - ["cache", "list", "--base-url", "http://127.0.0.1:8000"], + ["mathlib", "list", "--remote", "http://127.0.0.1:8000"], ) assert result.exit_code == 0 @@ -329,7 +327,7 @@ def fake_cache_init(self, cache_root_arg=None): def test_cache_pack_reports_missing_project_packages(tmp_path): runner = CliRunner() - result = runner.invoke(cli, ["cache", "pack", "v4.28.0", "--output-dir", str(tmp_path)]) + result = runner.invoke(cli, ["mathlib", "pack", "v4.28.0", "--output-dir", str(tmp_path)]) assert result.exit_code == 1 assert "Run 'leanup cache create v4.28.0' or 'leanup cache get v4.28.0 --base-url ...' first." in result.output @@ -357,11 +355,10 @@ def test_cache_get_downloads_and_extracts_packages_archive(tmp_path): command = [ sys.executable, "-c", - "from leanup.cli import cli; cli()", - "cache", - "serve", - "--host", - "127.0.0.1", + "from leanup.cli import cli; cli()", + "serve", + "--host", + "127.0.0.1", "--port", "18082", "--ltar-root", @@ -383,12 +380,12 @@ def test_cache_get_downloads_and_extracts_packages_archive(tmp_path): proc.stdout.readline() time.sleep(1) result = runner.invoke( - cli, - [ - "cache", - "get", - "v4.22.0", - "--base-url", + cli, + [ + "mathlib", + "get", + "v4.22.0", + "--remote", "http://127.0.0.1:18082", "--cache-dir", str(tmp_path / "local-cache"), @@ -428,7 +425,7 @@ def test_cache_pack_honors_cleanup_cache_dir_env_in_subprocess(tmp_path): sys.executable, "-c", "from leanup.cli import cli; cli()", - "cache", + "mathlib", "pack", "v4.22.0", "--no-pigz", @@ -458,7 +455,6 @@ def test_cache_serve_honors_explicit_roots_in_subprocess(tmp_path): sys.executable, "-c", "from leanup.cli import cli; cli()", - "cache", "serve", "--host", "127.0.0.1", @@ -516,7 +512,7 @@ def fake_cache_init(self, cache_root=None): monkeypatch.setattr(MathlibCacheManager, "__init__", fake_cache_init) - result = runner.invoke(cli, ["cache", "create", "v4.22.0", "--no-pigz"]) + result = runner.invoke(cli, ["mathlib", "create", "v4.22.0", "--no-pigz"]) expected_packages = custom_cache_root / "packages" / "v4.22.0" / "packages" expected_archive = custom_cache_root / "archives" / "v4.22.0" / "packages.tar.gz" @@ -525,3 +521,25 @@ def fake_cache_init(self, cache_root=None): assert expected_archive.exists() assert str(expected_packages) in result.output assert str(expected_archive) in result.output + + +def test_mathlib_unpack_extracts_local_archive(tmp_path): + runner = CliRunner() + cache_root = tmp_path / "cache-root" + archive_dir = cache_root / "archives" / "v4.28.0" + archive_dir.mkdir(parents=True, exist_ok=True) + + source_packages = tmp_path / "source-packages" / "mathlib" + source_packages.mkdir(parents=True, exist_ok=True) + (source_packages / "README.md").write_text("cached\n", encoding="utf-8") + archive = archive_dir / "packages.tar.gz" + MathlibCacheManager(cache_root=cache_root).pack_packages_archive(source_packages.parent, archive) + + result = runner.invoke( + cli, + ["mathlib", "unpack", "v4.28.0", "--cache-dir", str(cache_root)], + ) + + assert result.exit_code == 0 + extracted = cache_root / "packages" / "v4.28.0" / "packages" / "mathlib" / "README.md" + assert extracted.read_text(encoding="utf-8") == "cached\n" diff --git a/tests/test_toolchains_cli.py b/tests/test_toolchains_cli.py new file mode 100644 index 0000000..66057b2 --- /dev/null +++ b/tests/test_toolchains_cli.py @@ -0,0 +1,130 @@ +from pathlib import Path +import tarfile + +from click.testing import CliRunner + +from leanup.cli import cli + + +def _create_base_elan(elan_home: Path) -> None: + (elan_home / "bin").mkdir(parents=True, exist_ok=True) + (elan_home / "settings.toml").write_text("default_toolchain = 'stable'\n", encoding="utf-8") + (elan_home / "bin" / "elan").write_text("binary\n", encoding="utf-8") + + +def _create_toolchain(elan_home: Path, version: str) -> Path: + toolchain_dir = elan_home / "toolchains" / f"leanprover--lean4---{version}" + toolchain_dir.mkdir(parents=True, exist_ok=True) + (toolchain_dir / "VERSION").write_text(version, encoding="utf-8") + return toolchain_dir + + +def test_toolchains_list_local(tmp_path): + runner = CliRunner() + cache_dir = tmp_path / "toolchains" + (cache_dir / "archives").mkdir(parents=True, exist_ok=True) + (cache_dir / "archives" / "base-elan.tar.gz").write_bytes(b"ok") + (cache_dir / "archives" / "v4.28.0").mkdir(parents=True, exist_ok=True) + (cache_dir / "archives" / "v4.28.0" / "toolchain.tar.gz").write_bytes(b"ok") + + result = runner.invoke(cli, ["toolchains", "list", "--cache-dir", str(cache_dir)]) + + assert result.exit_code == 0 + assert "base" in result.output + assert "v4.28.0" in result.output + + +def test_toolchains_pack_without_version_packs_base_elan(tmp_path): + runner = CliRunner() + cache_dir = tmp_path / "toolchains" + elan_home = tmp_path / "leanup-elan" + _create_base_elan(elan_home) + + result = runner.invoke( + cli, + ["toolchains", "pack", "--cache-dir", str(cache_dir), "--elan-home", str(elan_home)], + ) + + archive = cache_dir / "archives" / "base-elan.tar.gz" + assert result.exit_code == 0 + assert archive.exists() + with tarfile.open(archive, "r:gz") as tar: + names = tar.getnames() + assert ".elan/settings.toml" in names + assert ".elan/bin/elan" in names + + +def test_toolchains_pack_and_unpack_version_archive(tmp_path): + runner = CliRunner() + cache_dir = tmp_path / "toolchains" + elan_home = tmp_path / "leanup-elan" + _create_base_elan(elan_home) + _create_toolchain(elan_home, "v4.28.0") + + pack_result = runner.invoke( + cli, + [ + "toolchains", + "pack", + "v4.28.0", + "--cache-dir", + str(cache_dir), + "--elan-home", + str(elan_home), + ], + ) + assert pack_result.exit_code == 0 + + extracted_home = tmp_path / "restored-elan" + unpack_result = runner.invoke( + cli, + [ + "toolchains", + "unpack", + "v4.28.0", + "--cache-dir", + str(cache_dir), + "--elan-home", + str(extracted_home), + ], + ) + assert unpack_result.exit_code == 0 + assert (extracted_home / "toolchains" / "leanprover--lean4---v4.28.0" / "VERSION").read_text(encoding="utf-8") == "v4.28.0" + + +def test_toolchains_init_with_url_downloads_base_archive(tmp_path, monkeypatch): + runner = CliRunner() + cache_dir = tmp_path / "toolchains" + elan_home = tmp_path / "restored-elan" + + archive_source = tmp_path / "source-elan" + _create_base_elan(archive_source) + manager_runner = CliRunner() + manager_runner.invoke( + cli, + ["toolchains", "pack", "--cache-dir", str(cache_dir), "--elan-home", str(archive_source)], + ) + + def fake_download(self, url: str): + return self.get_base_archive_path() + + from leanup.repo.toolchain_cache import ToolchainCacheManager + + monkeypatch.setattr(ToolchainCacheManager, "download_base_archive", fake_download) + + result = runner.invoke( + cli, + [ + "toolchains", + "init", + "--url", + "http://127.0.0.1:8000", + "--cache-dir", + str(cache_dir), + "--elan-home", + str(elan_home), + ], + ) + + assert result.exit_code == 0 + assert (elan_home / "settings.toml").exists() From 37f474c0c43ed86f50979ade9dd3d80e9ff3bc18 Mon Sep 17 00:00:00 2001 From: rex <1073853456@qq.com> Date: Tue, 14 Apr 2026 18:30:51 +0800 Subject: [PATCH 2/3] fix: align toolchain cli with ci and py39 --- .github/workflows/ci.yaml | 4 +++- leanup/__init__.py | 2 +- leanup/cli/toolchains.py | 7 ++++--- leanup/repo/elan.py | 2 +- leanup/repo/toolchain_cache.py | 17 +++++++++-------- 5 files changed, 18 insertions(+), 14 deletions(-) diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index b7a7e16..2c9c92d 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -35,7 +35,9 @@ jobs: run: | # Test basic CLI commands leanup --help - leanup cache --help + leanup mathlib --help + leanup serve --help + leanup toolchains --help leanup repo --help leanup repo install lean-zh/leanup leanup repo install lean-zh/repl diff --git a/leanup/__init__.py b/leanup/__init__.py index 8877138..5bd4a9f 100644 --- a/leanup/__init__.py +++ b/leanup/__init__.py @@ -2,7 +2,7 @@ __author__ = """Lean-zh Community""" __email__ = 'leanprover@outlook.com' -__version__ = '0.2.2' +__version__ = '0.2.3' from .repo import ( RepoManager, diff --git a/leanup/cli/toolchains.py b/leanup/cli/toolchains.py index e83f12f..906dc5a 100644 --- a/leanup/cli/toolchains.py +++ b/leanup/cli/toolchains.py @@ -2,6 +2,7 @@ from pathlib import Path import tempfile +from typing import Optional import click import requests @@ -27,7 +28,7 @@ def toolchains() -> None: default=TOOLCHAIN_CACHE_ROOT, help="Toolchain cache root.", ) -def list_toolchains(mode: str, remote_url: str | None, cache_dir: Path) -> None: +def list_toolchains(mode: str, remote_url: Optional[str], cache_dir: Path) -> None: manager = ToolchainCacheManager(cache_root=cache_dir) if remote_url: has_base, versions = manager.list_remote(remote_url) @@ -62,7 +63,7 @@ def list_toolchains(mode: str, remote_url: str | None, cache_dir: Path) -> None: default=Path(tempfile.gettempdir()) / "leanup-elan", help="Target .elan directory.", ) -def init_toolchains(url: str | None, cache_dir: Path, elan_home: Path) -> None: +def init_toolchains(url: Optional[str], cache_dir: Path, elan_home: Path) -> None: manager = ToolchainCacheManager(cache_root=cache_dir, elan_home=elan_home) try: result = manager.init_base(url) @@ -85,7 +86,7 @@ def init_toolchains(url: str | None, cache_dir: Path, elan_home: Path) -> None: default=Path(tempfile.gettempdir()) / "leanup-elan", help="Source .elan directory.", ) -def pack_toolchain(lean_version: str | None, cache_dir: Path, elan_home: Path) -> None: +def pack_toolchain(lean_version: Optional[str], cache_dir: Path, elan_home: Path) -> None: manager = ToolchainCacheManager(cache_root=cache_dir, elan_home=elan_home) try: archive = manager.pack_base_archive() if lean_version is None else manager.pack_toolchain_archive(lean_version) diff --git a/leanup/repo/elan.py b/leanup/repo/elan.py index 91bb4f7..61db8e9 100644 --- a/leanup/repo/elan.py +++ b/leanup/repo/elan.py @@ -16,7 +16,7 @@ class ElanManager: """Elan toolchain manager""" - def __init__(self, elan_home: Path | None = None): + def __init__(self, elan_home: Optional[Path] = None): self.elan_home = Path(elan_home or os.environ.get("ELAN_HOME", Path.home() / ".elan")) self.elan_bin_dir = self.elan_home / "bin" self._elan_exe = None diff --git a/leanup/repo/toolchain_cache.py b/leanup/repo/toolchain_cache.py index 3b6bfd0..bcd1120 100644 --- a/leanup/repo/toolchain_cache.py +++ b/leanup/repo/toolchain_cache.py @@ -4,6 +4,7 @@ from pathlib import Path import tarfile import tempfile +from typing import List, Optional, Tuple from urllib.parse import urljoin import requests @@ -17,7 +18,7 @@ class ToolchainCacheManager: - def __init__(self, cache_root: Path | None = None, elan_home: Path | None = None): + def __init__(self, cache_root: Optional[Path] = None, elan_home: Optional[Path] = None): self.cache_root = cache_root or (LEANUP_CACHE_DIR / "toolchains") self.archives_root = self.cache_root / "archives" self.elan_home = elan_home or Path(os.environ.get("ELAN_HOME", Path.home() / ".elan")) @@ -28,10 +29,10 @@ def get_base_archive_path(self) -> Path: def get_toolchain_archive_path(self, version: str) -> Path: return self.archives_root / normalize_lean_version(version) / "toolchain.tar.gz" - def list_local_versions(self) -> list[str]: + def list_local_versions(self) -> List[str]: if not self.archives_root.exists(): return [] - versions: list[str] = [] + versions: List[str] = [] for child in sorted(self.archives_root.iterdir()): if child.is_dir() and (child / "toolchain.tar.gz").exists(): versions.append(child.name) @@ -40,7 +41,7 @@ def list_local_versions(self) -> list[str]: def has_local_base_archive(self) -> bool: return self.get_base_archive_path().exists() - def list_remote(self, base_url: str) -> tuple[bool, list[str]]: + def list_remote(self, base_url: str) -> Tuple[bool, List[str]]: index_url = urljoin(base_url.rstrip("/") + "/", "toolchains/index.json") try: response = requests.get(index_url, timeout=10) @@ -65,7 +66,7 @@ def download_base_archive(self, url: str) -> Path: def download_toolchain_archive(self, version: str, url: str) -> Path: return self._download_to(url, self.get_toolchain_archive_path(version)) - def init_base(self, url: str | None = None) -> Path: + def init_base(self, url: Optional[str] = None) -> Path: if url: archive = self.download_base_archive(self.build_base_url(url)) return self.unpack_base_archive(archive) @@ -96,7 +97,7 @@ def pack_base_archive(self) -> Path: logger.info(f"Packed base elan archive -> {output_file}") return output_file - def unpack_base_archive(self, archive_path: Path | None = None) -> Path: + def unpack_base_archive(self, archive_path: Optional[Path] = None) -> Path: archive_path = archive_path or self.get_base_archive_path() temp_root = Path(tempfile.mkdtemp(prefix=".elan-base.", dir=self.elan_home.parent)) try: @@ -135,7 +136,7 @@ def pack_toolchain_archive(self, version: str) -> Path: logger.info(f"Packed toolchain {version} -> {output_file}") return output_file - def unpack_toolchain_archive(self, version: str, archive_path: Path | None = None) -> Path: + def unpack_toolchain_archive(self, version: str, archive_path: Optional[Path] = None) -> Path: archive_path = archive_path or self.get_toolchain_archive_path(version) temp_root = Path(tempfile.mkdtemp(prefix=".elan-toolchain.", dir=self.elan_home.parent)) try: @@ -164,7 +165,7 @@ def fetch_toolchain(self, version: str, base_url: str) -> Path: archive = self.download_toolchain_archive(version, self.build_toolchain_url(version, base_url)) return self.unpack_toolchain_archive(version, archive) - def _resolve_installed_toolchain_dir(self, version: str) -> Path | None: + def _resolve_installed_toolchain_dir(self, version: str) -> Optional[Path]: normalized = normalize_lean_version(version) toolchains_root = self.elan_home / "toolchains" if not toolchains_root.exists(): From b9cdd41cdb164199567964c2287fe6478e592b0f Mon Sep 17 00:00:00 2001 From: rex <1073853456@qq.com> Date: Tue, 14 Apr 2026 18:33:03 +0800 Subject: [PATCH 3/3] fix: preserve execute_command compatibility --- leanup/utils/basic.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/leanup/utils/basic.py b/leanup/utils/basic.py index 6b4a12f..92b11a9 100644 --- a/leanup/utils/basic.py +++ b/leanup/utils/basic.py @@ -35,15 +35,16 @@ def execute_command(command: Union[str, List[str]], # Handle string commands if isinstance(command, str) and OS_TYPE != 'Windows': command = shlex.split(command) - process = subprocess.Popen( - command, - cwd=cwd, - stdout=stdout_pipe, - stderr=stderr_pipe, - shell=OS_TYPE == 'Windows', - text=text, - env=env, - ) + popen_kwargs = { + "cwd": cwd, + "stdout": stdout_pipe, + "stderr": stderr_pipe, + "shell": OS_TYPE == 'Windows', + "text": text, + } + if env is not None: + popen_kwargs["env"] = env + process = subprocess.Popen(command, **popen_kwargs) stdout, stderr = process.communicate(input=input, timeout=timeout) returncode = process.returncode stdout = stdout or ""