From de368ea810a92b7893106c1522ad6592065d5b3e Mon Sep 17 00:00:00 2001 From: gooker_young Date: Sun, 5 Jul 2026 08:39:20 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=88=A0=E9=99=A4=E5=86=97?= =?UTF-8?q?=E4=BD=99=20cli=20=E5=85=A5=E5=8F=A3=E8=84=9A=E6=9C=AC,=20gitto?= =?UTF-8?q?ol=20=E7=94=A8=E6=95=B0=E7=BB=84=E9=85=8D=E7=BD=AE=20clean=20ex?= =?UTF-8?q?cludes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 删除 13 个已有 YAML 配置的 cli .py 入口脚本, 统一通过 pf 调用 2. gittool.yaml 用 CLEAN_EXCLUDES 数组变量配置 git clean 的 -e 参数, 保留 .venv/.tox/node_modules/.idea 等目录避免误删 3. run_cli 执行前打印调用信息: [gittool] 执行: c 4. 更新 pyproject.toml 移除 13 个冗余 entry points, 仅保留 pf 5. 清理测试文件中的 TestMain 类 (测 _ops 模块的测试保留) --- pyproject.toml | 13 --- src/pyflowx/__init__.py | 1 + src/pyflowx/cli/autofmt.py | 52 ----------- src/pyflowx/cli/bumpversion.py | 43 --------- src/pyflowx/cli/configs/gittool.yaml | 9 +- src/pyflowx/cli/filedate.py | 41 -------- src/pyflowx/cli/filelevel.py | 39 -------- src/pyflowx/cli/folderback.py | 32 ------- src/pyflowx/cli/folderzip.py | 30 ------ src/pyflowx/cli/gittool.py | 39 -------- src/pyflowx/cli/lscalc.py | 51 ---------- src/pyflowx/cli/packtool.py | 76 --------------- src/pyflowx/cli/pdftool.py | 134 --------------------------- src/pyflowx/cli/piptool.py | 59 ------------ src/pyflowx/cli/screenshot.py | 40 -------- src/pyflowx/cli/sshcopyid.py | 42 --------- src/pyflowx/yaml_loader.py | 19 ++-- tests/cli/test_autofmt.py | 84 ----------------- tests/cli/test_bumpversion.py | 57 ------------ tests/cli/test_filedate.py | 34 ------- tests/cli/test_filelevel.py | 38 -------- tests/cli/test_folderback.py | 19 ---- tests/cli/test_folderzip.py | 17 ---- tests/cli/test_gittool.py | 34 ------- tests/cli/test_lscalc.py | 48 ---------- tests/cli/test_packtool.py | 66 ------------- tests/cli/test_pdftool.py | 45 --------- tests/cli/test_piptool.py | 51 ---------- tests/cli/test_screenshot.py | 27 ------ tests/cli/test_sshcopyid.py | 68 -------------- 30 files changed, 20 insertions(+), 1288 deletions(-) delete mode 100644 src/pyflowx/cli/autofmt.py delete mode 100644 src/pyflowx/cli/bumpversion.py delete mode 100644 src/pyflowx/cli/filedate.py delete mode 100644 src/pyflowx/cli/filelevel.py delete mode 100644 src/pyflowx/cli/folderback.py delete mode 100644 src/pyflowx/cli/folderzip.py delete mode 100644 src/pyflowx/cli/gittool.py delete mode 100644 src/pyflowx/cli/lscalc.py delete mode 100644 src/pyflowx/cli/packtool.py delete mode 100644 src/pyflowx/cli/pdftool.py delete mode 100644 src/pyflowx/cli/piptool.py delete mode 100644 src/pyflowx/cli/screenshot.py delete mode 100644 src/pyflowx/cli/sshcopyid.py diff --git a/pyproject.toml b/pyproject.toml index 3310f67..cce82f2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,27 +25,14 @@ requires-python = ">=3.8" version = "0.3.5" [project.scripts] -autofmt = "pyflowx.cli.autofmt:main" -bumpversion = "pyflowx.cli.bumpversion:main" dockercmd = "pyflowx.cli.dev.dockercmd:main" emlman = "pyflowx.cli.emlmanager:main" -filedate = "pyflowx.cli.filedate:main" -filelvl = "pyflowx.cli.filelevel:main" -foldback = "pyflowx.cli.folderback:main" -foldzip = "pyflowx.cli.folderzip:main" -gitt = "pyflowx.cli.gittool:main" -lscalc = "pyflowx.cli.lscalc:main" msdown = "pyflowx.cli.llm.msdownload:main" -packtool = "pyflowx.cli.packtool:main" -pdftool = "pyflowx.cli.pdftool:main" pf = "pyflowx.cli.pf:main" -piptool = "pyflowx.cli.piptool:main" pxp = "pyflowx.cli.profiler:main" pymake = "pyflowx.cli.pymake:main" reseticon = "pyflowx.cli.reseticoncache:main" -scrcap = "pyflowx.cli.screenshot:main" sglang = "pyflowx.cli.llm.sglang:main" -sshcopy = "pyflowx.cli.sshcopyid:main" yamlrun = "pyflowx.cli.yamlrun:main" # dev envdev = "pyflowx.cli.dev.envdev:main" diff --git a/src/pyflowx/__init__.py b/src/pyflowx/__init__.py index fc630be..2894cc1 100644 --- a/src/pyflowx/__init__.py +++ b/src/pyflowx/__init__.py @@ -144,6 +144,7 @@ __all__ = [ "TaskTimeoutError", "YamlLoadError", "build_call_args", + "build_cli_parser", "cmd", "compose", "describe_injection", diff --git a/src/pyflowx/cli/autofmt.py b/src/pyflowx/cli/autofmt.py deleted file mode 100644 index 0116078..0000000 --- a/src/pyflowx/cli/autofmt.py +++ /dev/null @@ -1,52 +0,0 @@ -"""自动格式化工具模块. - -提供 Python 代码自动格式化的常用功能封装, -支持 docstring 自动生成、pyproject.toml 配置同步等功能. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "autofmt.yaml" - - -def main() -> None: - """自动格式化工具主函数.""" - parser = argparse.ArgumentParser( - description="AutoFmt - 自动格式化工具", - usage="autofmt [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - format_parser = subparsers.add_parser("fmt", help="使用 ruff 格式化代码") - format_parser.add_argument("--target", type=str, default=".", help="目标路径") - - lint_parser = subparsers.add_parser("lint", help="使用 ruff 检查代码") - lint_parser.add_argument("--target", type=str, default=".", help="目标路径") - lint_parser.add_argument("--fix", action="store_true", help="自动修复") - - doc_parser = subparsers.add_parser("doc", help="自动添加 docstring") - doc_parser.add_argument("--root-dir", type=str, default=".", help="根目录") - - sync_parser = subparsers.add_parser("sync", help="同步 pyproject.toml 配置") - sync_parser.add_argument("--root-dir", type=str, default=".", help="根目录") - - args = parser.parse_args() - - if args.command not in ("fmt", "lint", "doc", "sync"): - parser.print_help() - return - - if args.command == "fmt": - px.run_yaml(_CONFIG, jobs="fmt", variables={"TARGET": args.target}) - elif args.command == "lint": - job = "lint_fix" if args.fix else "lint" - px.run_yaml(_CONFIG, jobs=job, variables={"TARGET": args.target}) - elif args.command == "doc": - px.run_yaml(_CONFIG, jobs="doc", variables={"ROOT_DIR": args.root_dir}) - elif args.command == "sync": - px.run_yaml(_CONFIG, jobs="sync", variables={"ROOT_DIR": args.root_dir}) diff --git a/src/pyflowx/cli/bumpversion.py b/src/pyflowx/cli/bumpversion.py deleted file mode 100644 index 0a1add8..0000000 --- a/src/pyflowx/cli/bumpversion.py +++ /dev/null @@ -1,43 +0,0 @@ -"""版本号自动管理工具. - -通过 YAML 配置调用 ``bump_project_version``, 搜索项目中所有 -``__init__.py`` 和 ``pyproject.toml`` 文件, 更新版本号并提交. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path -from typing import Literal, get_args - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "bumpversion.yaml" - -BumpVersionType = Literal["patch", "minor", "major"] - - -def main() -> None: - """版本号管理工具主函数.""" - parser = argparse.ArgumentParser(description="BumpVersion - 版本号自动管理工具") - parser.add_argument( - "part", - type=str, - nargs="?", - default="patch", - choices=get_args(BumpVersionType), - help=f"版本部分: {get_args(BumpVersionType)}", - ) - parser.add_argument( - "--no-tag", - action="store_true", - help="提交后不创建 git tag", - ) - - args = parser.parse_args() - - px.run_yaml( - _CONFIG, - jobs="bump", - variables={"PART": args.part, "NO_TAG": args.no_tag}, - ) diff --git a/src/pyflowx/cli/configs/gittool.yaml b/src/pyflowx/cli/configs/gittool.yaml index cf6f0d1..2d5afeb 100644 --- a/src/pyflowx/cli/configs/gittool.yaml +++ b/src/pyflowx/cli/configs/gittool.yaml @@ -7,6 +7,11 @@ # pf gittool p # pf gittool pl strategy: thread +variables: + # git clean -e 参数列表 (展开为 cmd 数组元素) + CLEAN_EXCLUDES: ["-e", ".venv", "-e", ".tox", "-e", "node_modules", + "-e", ".idea", "-e", "idea.config", + "-e", "idea_modules.xml", "-e", "vcs.xml"] cli: description: "GitTool - Git 执行工具" usage: "pf gittool " @@ -14,7 +19,7 @@ cli: a: help: "添加并提交" c: - help: "查看状态" + help: "清理并查看状态" i: help: "初始化并提交" isub: @@ -28,7 +33,7 @@ jobs: fn: git_add_commit args: ["chore: update"] clean: - cmd: ["git", "clean", "-xfd", "-e", "idea.config", "-e", "idea_modules.xml", "-e", "vcs.xml"] + cmd: ["git", "clean", "-xfd", "${CLEAN_EXCLUDES}"] c: needs: [clean] cmd: ["git", "status", "--porcelain"] diff --git a/src/pyflowx/cli/filedate.py b/src/pyflowx/cli/filedate.py deleted file mode 100644 index 5003de4..0000000 --- a/src/pyflowx/cli/filedate.py +++ /dev/null @@ -1,41 +0,0 @@ -"""文件日期处理工具. - -自动检测文件名的日期前缀, -并根据文件的实际创建或修改时间重命名文件. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "filedate.yaml" - - -def main() -> None: - """文件日期处理工具主函数.""" - parser = argparse.ArgumentParser( - description="FileDate - 文件日期处理工具", - usage="filedate [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - add_parser = subparsers.add_parser("add", help="添加日期前缀") - add_parser.add_argument("files", nargs="+", help="文件路径") - - clear_parser = subparsers.add_parser("clear", help="清除日期前缀") - clear_parser.add_argument("files", nargs="+", help="文件路径") - - args = parser.parse_args() - - if args.command not in ("add", "clear"): - parser.print_help() - return - - px.run_yaml( - _CONFIG, - jobs=args.command, - variables={"FILES": [Path(f) for f in args.files]}, - ) diff --git a/src/pyflowx/cli/filelevel.py b/src/pyflowx/cli/filelevel.py deleted file mode 100644 index fa76864..0000000 --- a/src/pyflowx/cli/filelevel.py +++ /dev/null @@ -1,39 +0,0 @@ -"""文件等级重命名工具. - -根据文件等级配置自动重命名文件, -支持多种等级标识和括号格式. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "filelevel.yaml" - - -def main() -> None: - """文件等级重命名工具主函数.""" - parser = argparse.ArgumentParser( - description="FileLevel - 文件等级重命名工具", - usage="filelevel [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - level_parser = subparsers.add_parser("set", help="设置文件等级") - level_parser.add_argument("files", nargs="+", help="文件路径") - level_parser.add_argument("--level", type=int, choices=[0, 1, 2, 3, 4], required=True, help="文件等级 (0-4)") - - args = parser.parse_args() - - if args.command != "set": - parser.print_help() - return - - px.run_yaml( - _CONFIG, - jobs="set", - variables={"FILES": [Path(f) for f in args.files], "LEVEL": args.level}, - ) diff --git a/src/pyflowx/cli/folderback.py b/src/pyflowx/cli/folderback.py deleted file mode 100644 index eb4563f..0000000 --- a/src/pyflowx/cli/folderback.py +++ /dev/null @@ -1,32 +0,0 @@ -"""文件夹备份工具. - -备份文件和文件夹为 zip 文件, -自动删除超过最大数量的旧备份文件. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "folderback.yaml" - - -def main() -> None: - """文件夹备份工具主函数.""" - parser = argparse.ArgumentParser( - description="FolderBack - 文件夹备份工具", - usage="folderback [--src SRC] [--dst DST] [--max-zip MAX]", - ) - parser.add_argument("--src", type=str, default=".", help="源文件夹路径 (默认: 当前目录)") - parser.add_argument("--dst", type=str, default="./backup", help="目标文件夹路径 (默认: ./backup)") - parser.add_argument("--max-zip", type=int, default=5, help="最大备份数量 (默认: 5)") - args = parser.parse_args() - - px.run_yaml( - _CONFIG, - jobs="backup", - variables={"SRC": args.src, "DST": args.dst, "MAX_ZIP": args.max_zip}, - ) diff --git a/src/pyflowx/cli/folderzip.py b/src/pyflowx/cli/folderzip.py deleted file mode 100644 index cac292e..0000000 --- a/src/pyflowx/cli/folderzip.py +++ /dev/null @@ -1,30 +0,0 @@ -"""文件夹压缩工具. - -压缩目录下的所有文件/文件夹为 zip 文件, -默认压缩当前目录下的所有子文件夹. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "folderzip.yaml" - - -def main() -> None: - """文件夹压缩工具主函数.""" - parser = argparse.ArgumentParser( - description="FolderZip - 文件夹压缩工具", - usage="folderzip [--cwd DIR]", - ) - parser.add_argument("--cwd", type=str, default=".", help="工作目录 (默认: 当前目录)") - args = parser.parse_args() - - px.run_yaml( - _CONFIG, - jobs="zip", - variables={"CWD": args.cwd}, - ) diff --git a/src/pyflowx/cli/gittool.py b/src/pyflowx/cli/gittool.py deleted file mode 100644 index 83ce482..0000000 --- a/src/pyflowx/cli/gittool.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Git 工具模块. - -提供 Git 仓库管理的常用操作封装, -支持初始化、提交、清理、推送等功能. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "gittool.yaml" - - -def main() -> None: - """Git工具主函数.""" - parser = argparse.ArgumentParser( - description="Gittool - Git 执行工具", - usage="gitt ", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - subparsers.add_parser("a", help="添加并提交") - subparsers.add_parser("c", help="清理 (git clean -xfd + status)") - subparsers.add_parser("i", help="初始化、添加并提交") - subparsers.add_parser("isub", help="初始化子目录的 Git 仓库") - subparsers.add_parser("p", help="推送") - subparsers.add_parser("pl", help="拉取") - - args = parser.parse_args() - - valid_commands = ("a", "c", "i", "isub", "p", "pl") - if args.command not in valid_commands: - parser.print_help() - return - - px.run_yaml(_CONFIG, jobs=args.command) diff --git a/src/pyflowx/cli/lscalc.py b/src/pyflowx/cli/lscalc.py deleted file mode 100644 index c434e79..0000000 --- a/src/pyflowx/cli/lscalc.py +++ /dev/null @@ -1,51 +0,0 @@ -"""LS-DYNA 计算工具. - -用于管理 LS-DYNA 仿真计算任务, -支持启动、监控和管理计算进程. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "lscalc.yaml" - -_DEFAULT_NCPU = 4 - - -def main() -> None: - """LS-DYNA 计算工具主函数.""" - parser = argparse.ArgumentParser( - description="LSCalc - LS-DYNA 计算工具", - usage="lscalc [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - run_parser = subparsers.add_parser("run", help="运行 LS-DYNA 计算") - run_parser.add_argument("input_file", help="输入文件路径") - run_parser.add_argument("--ncpu", type=int, default=_DEFAULT_NCPU, help="CPU 核心数") - - mpi_parser = subparsers.add_parser("mpi", help="运行 LS-DYNA MPI 计算") - mpi_parser.add_argument("input_file", help="输入文件路径") - mpi_parser.add_argument("--ncpu", type=int, default=_DEFAULT_NCPU, help="CPU 核心数") - - subparsers.add_parser("status", help="检查 LS-DYNA 进程状态") - - args = parser.parse_args() - - if args.command not in ("run", "mpi", "status"): - parser.print_help() - return - - variables: dict[str, object] = {"NCPU": getattr(args, "ncpu", _DEFAULT_NCPU)} - if args.command in ("run", "mpi"): - variables["INPUT_FILE"] = args.input_file - - px.run_yaml( - _CONFIG, - jobs=args.command, - variables=variables, - ) diff --git a/src/pyflowx/cli/packtool.py b/src/pyflowx/cli/packtool.py deleted file mode 100644 index 8feb0c4..0000000 --- a/src/pyflowx/cli/packtool.py +++ /dev/null @@ -1,76 +0,0 @@ -"""Python 打包工具模块. - -提供 Python 项目打包的常用功能封装, -支持源码打包、依赖打包、嵌入式 Python 安装等功能. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "packtool.yaml" - -_DEFAULT_BUILD_DIR = ".pypack" -_DEFAULT_DIST_DIR = "dist" -_DEFAULT_LIB_DIR = "libs" - - -def main() -> None: - """Python 打包工具主函数.""" - parser = argparse.ArgumentParser( - description="PackTool - Python 打包工具", - usage="packtool [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - src_parser = subparsers.add_parser("src", help="打包项目源码") - src_parser.add_argument("--project-dir", type=str, default=".", help="项目目录") - src_parser.add_argument("--output-dir", type=str, default=_DEFAULT_BUILD_DIR, help="输出目录") - - deps_parser = subparsers.add_parser("deps", help="打包项目依赖") - deps_parser.add_argument("--lib-dir", type=str, default=_DEFAULT_LIB_DIR, help="依赖库目录") - deps_parser.add_argument("dependencies", nargs="*", help="依赖列表") - - wheel_parser = subparsers.add_parser("wheel", help="打包项目为 wheel 文件") - wheel_parser.add_argument("--project-dir", type=str, default=".", help="项目目录") - wheel_parser.add_argument("--output-dir", type=str, default=_DEFAULT_DIST_DIR, help="输出目录") - - embed_parser = subparsers.add_parser("embed", help="安装嵌入式 Python") - embed_parser.add_argument("--version", type=str, default="3.10", help="Python 版本") - embed_parser.add_argument("--output-dir", type=str, default="python", help="输出目录") - - zip_parser = subparsers.add_parser("zip", help="创建 ZIP 打包文件") - zip_parser.add_argument("--source-dir", type=str, default=".", help="源目录") - zip_parser.add_argument("--output-file", type=str, default="package.zip", help="输出文件") - - subparsers.add_parser("clean", help="清理构建目录") - - args = parser.parse_args() - - if args.command not in ("src", "deps", "wheel", "embed", "zip", "clean"): - parser.print_help() - return - - variables: dict[str, object] = {} - - if args.command == "src": - variables = {"PROJECT_DIR": Path(args.project_dir), "OUTPUT_DIR": Path(args.output_dir)} - elif args.command == "deps": - variables = {"LIB_DIR": Path(args.lib_dir), "DEPENDENCIES": args.dependencies} - elif args.command == "wheel": - variables = {"PROJECT_DIR": Path(args.project_dir), "OUTPUT_DIR": Path(args.output_dir)} - elif args.command == "embed": - variables = {"VERSION": args.version, "OUTPUT_DIR": Path(args.output_dir)} - elif args.command == "zip": - variables = {"SOURCE_DIR": Path(args.source_dir), "OUTPUT_FILE": Path(args.output_file)} - elif args.command == "clean": - variables = {"OUTPUT_DIR": Path(_DEFAULT_BUILD_DIR)} - - px.run_yaml( - _CONFIG, - jobs=args.command, - variables=variables, - ) diff --git a/src/pyflowx/cli/pdftool.py b/src/pyflowx/cli/pdftool.py deleted file mode 100644 index 65d982e..0000000 --- a/src/pyflowx/cli/pdftool.py +++ /dev/null @@ -1,134 +0,0 @@ -"""PDF 工具模块. - -提供 PDF 文件操作的常用功能封装, -支持合并、拆分、压缩、加密、水印、OCR等功能. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "pdftool.yaml" - - -def main() -> None: # noqa: PLR0912 - """PDF 工具主函数.""" - parser = argparse.ArgumentParser( - description="PDFTool - PDF 文件工具集", - usage="pdftool [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - merge_parser = subparsers.add_parser("m", help="合并 PDF 文件") - merge_parser.add_argument("inputs", nargs="+", help="输入 PDF 文件路径") - merge_parser.add_argument("--output", type=str, default="merged.pdf", help="输出文件路径") - - split_parser = subparsers.add_parser("s", help="拆分 PDF 文件为单页") - split_parser.add_argument("input", help="输入 PDF 文件路径") - split_parser.add_argument("--output-dir", type=str, default="split", help="输出目录") - - compress_parser = subparsers.add_parser("c", help="压缩 PDF 文件") - compress_parser.add_argument("input", help="输入 PDF 文件路径") - compress_parser.add_argument("--output", type=str, default="compressed.pdf", help="输出文件路径") - - encrypt_parser = subparsers.add_parser("e", help="加密 PDF 文件") - encrypt_parser.add_argument("input", help="输入 PDF 文件路径") - encrypt_parser.add_argument("--output", type=str, default="encrypted.pdf", help="输出文件路径") - encrypt_parser.add_argument("--password", type=str, required=True, help="密码") - - decrypt_parser = subparsers.add_parser("d", help="解密 PDF 文件") - decrypt_parser.add_argument("input", help="输入 PDF 文件路径") - decrypt_parser.add_argument("--output", type=str, default="decrypted.pdf", help="输出文件路径") - decrypt_parser.add_argument("--password", type=str, required=True, help="密码") - - extract_text_parser = subparsers.add_parser("xt", help="提取 PDF 文本") - extract_text_parser.add_argument("input", help="输入 PDF 文件路径") - extract_text_parser.add_argument("--output", type=str, default="output.txt", help="输出文件路径") - - extract_images_parser = subparsers.add_parser("xi", help="提取 PDF 图片") - extract_images_parser.add_argument("input", help="输入 PDF 文件路径") - extract_images_parser.add_argument("--output-dir", type=str, default="images", help="输出目录") - - watermark_parser = subparsers.add_parser("w", help="添加 PDF 水印") - watermark_parser.add_argument("input", help="输入 PDF 文件路径") - watermark_parser.add_argument("--output", type=str, default="watermarked.pdf", help="输出文件路径") - watermark_parser.add_argument("--text", type=str, default="CONFIDENTIAL", help="水印文本") - - rotate_parser = subparsers.add_parser("r", help="旋转 PDF 页面") - rotate_parser.add_argument("input", help="输入 PDF 文件路径") - rotate_parser.add_argument("--output", type=str, default="rotated.pdf", help="输出文件路径") - rotate_parser.add_argument("--rotation", type=int, default=90, help="旋转角度 (90, 180, 270)") - - crop_parser = subparsers.add_parser("crop", help="裁剪 PDF 页面") - crop_parser.add_argument("input", help="输入 PDF 文件路径") - crop_parser.add_argument("--output", type=str, default="cropped.pdf", help="输出文件路径") - crop_parser.add_argument("--left", type=int, default=10, help="左边裁剪") - crop_parser.add_argument("--top", type=int, default=10, help="顶部裁剪") - crop_parser.add_argument("--right", type=int, default=10, help="右边裁剪") - crop_parser.add_argument("--bottom", type=int, default=10, help="底部裁剪") - - info_parser = subparsers.add_parser("i", help="显示 PDF 信息") - info_parser.add_argument("input", help="输入 PDF 文件路径") - - ocr_parser = subparsers.add_parser("ocr", help="PDF OCR 识别") - ocr_parser.add_argument("input", help="输入 PDF 文件路径") - ocr_parser.add_argument("--output", type=str, default="ocr.pdf", help="输出文件路径") - ocr_parser.add_argument("--lang", type=str, default="chi_sim+eng", help="OCR 语言") - - to_images_parser = subparsers.add_parser("img", help="PDF 转图片") - to_images_parser.add_argument("input", help="输入 PDF 文件路径") - to_images_parser.add_argument("--output-dir", type=str, default="images", help="输出目录") - to_images_parser.add_argument("--dpi", type=int, default=300, help="图片 DPI") - - repair_parser = subparsers.add_parser("repair", help="修复 PDF 文件") - repair_parser.add_argument("input", help="输入 PDF 文件路径") - repair_parser.add_argument("--output", type=str, default="repaired.pdf", help="输出文件路径") - - args = parser.parse_args() - - valid_commands = ("m", "s", "c", "e", "d", "xt", "xi", "w", "r", "crop", "i", "ocr", "img", "repair") - if args.command not in valid_commands: - parser.print_help() - return - - variables: dict[str, object] = {} - - if args.command == "m": - variables = {"INPUTS": [Path(p) for p in args.inputs], "OUTPUT": Path(args.output)} - elif args.command == "s": - variables = {"INPUT": Path(args.input), "OUTPUT_DIR": Path(args.output_dir)} - elif args.command == "c": - variables = {"INPUT": Path(args.input), "OUTPUT": Path(args.output)} - elif args.command in {"e", "d"}: - variables = {"INPUT": Path(args.input), "OUTPUT": Path(args.output), "PASSWORD": args.password} - elif args.command == "xt": - variables = {"INPUT": Path(args.input), "OUTPUT": Path(args.output)} - elif args.command == "xi": - variables = {"INPUT": Path(args.input), "OUTPUT_DIR": Path(args.output_dir)} - elif args.command == "w": - variables = {"INPUT": Path(args.input), "OUTPUT": Path(args.output), "TEXT": args.text} - elif args.command == "r": - variables = {"INPUT": Path(args.input), "OUTPUT": Path(args.output), "ROTATION": args.rotation} - elif args.command == "crop": - variables = { - "INPUT": Path(args.input), - "OUTPUT": Path(args.output), - "MARGINS": [args.left, args.top, args.right, args.bottom], - } - elif args.command == "i": - variables = {"INPUT": Path(args.input)} - elif args.command == "ocr": - variables = {"INPUT": Path(args.input), "OUTPUT": Path(args.output), "LANG": args.lang} - elif args.command == "img": - variables = {"INPUT": Path(args.input), "OUTPUT_DIR": Path(args.output_dir), "DPI": args.dpi} - elif args.command == "repair": - variables = {"INPUT": Path(args.input), "OUTPUT": Path(args.output)} - - px.run_yaml( - _CONFIG, - jobs=args.command, - variables=variables, - ) diff --git a/src/pyflowx/cli/piptool.py b/src/pyflowx/cli/piptool.py deleted file mode 100644 index 29faf69..0000000 --- a/src/pyflowx/cli/piptool.py +++ /dev/null @@ -1,59 +0,0 @@ -"""pip 包管理工具模块. - -提供 pip 包管理操作的封装, -支持安装、卸载、下载等功能. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "piptool.yaml" - - -def main() -> None: - """pip 工具主函数.""" - parser = argparse.ArgumentParser( - description="PipTool - pip 包管理工具", - usage="piptool [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - install_parser = subparsers.add_parser("i", help="安装包") - install_parser.add_argument("packages", nargs="+", help="要安装的包名") - - uninstall_parser = subparsers.add_parser("u", help="卸载包") - uninstall_parser.add_argument("packages", nargs="+", help="要卸载的包名 (支持通配符)") - - reinstall_parser = subparsers.add_parser("r", help="重新安装包") - reinstall_parser.add_argument("packages", nargs="+", help="要重装的包名") - reinstall_parser.add_argument("--offline", action="store_true", help="使用离线模式") - - download_parser = subparsers.add_parser("d", help="下载包") - download_parser.add_argument("packages", nargs="+", help="要下载的包名") - download_parser.add_argument("--offline", action="store_true", help="使用离线模式") - - subparsers.add_parser("up", help="升级 pip") - - subparsers.add_parser("f", help="冻结依赖到 requirements.txt") - - args = parser.parse_args() - - if args.command not in ("i", "u", "r", "d", "up", "f"): - parser.print_help() - return - - variables: dict[str, object] = {} - if args.command in ("i", "u", "r", "d"): - variables["PACKAGES"] = args.packages - if args.command in ("r", "d"): - variables["OFFLINE"] = args.offline - - px.run_yaml( - _CONFIG, - jobs=args.command, - variables=variables, - ) diff --git a/src/pyflowx/cli/screenshot.py b/src/pyflowx/cli/screenshot.py deleted file mode 100644 index 8c399fc..0000000 --- a/src/pyflowx/cli/screenshot.py +++ /dev/null @@ -1,40 +0,0 @@ -"""截图工具. - -跨平台截图工具, 支持全屏截图和区域截图. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "screenshot.yaml" - - -def main() -> None: - """截图工具主函数.""" - parser = argparse.ArgumentParser( - description="Screenshot - 截图工具", - usage="screenshot [options]", - ) - subparsers = parser.add_subparsers(dest="command", help="可用命令") - - full_parser = subparsers.add_parser("full", help="全屏截图") - full_parser.add_argument("--filename", type=str, help="文件名") - - area_parser = subparsers.add_parser("area", help="区域截图") - area_parser.add_argument("--filename", type=str, help="文件名") - - args = parser.parse_args() - - if args.command not in ("full", "area"): - parser.print_help() - return - - px.run_yaml( - _CONFIG, - jobs=args.command, - variables={"FILENAME": args.filename}, - ) diff --git a/src/pyflowx/cli/sshcopyid.py b/src/pyflowx/cli/sshcopyid.py deleted file mode 100644 index bebc2a8..0000000 --- a/src/pyflowx/cli/sshcopyid.py +++ /dev/null @@ -1,42 +0,0 @@ -"""SSH 密钥部署工具. - -类似 ssh-copy-id, 自动将 SSH 公钥部署到远程服务器, -支持密码认证和密钥认证两种方式. -""" - -from __future__ import annotations - -import argparse -from pathlib import Path - -import pyflowx as px - -_CONFIG = Path(__file__).parent / "configs" / "sshcopyid.yaml" - - -def main() -> None: - """SSH 密钥部署工具主函数.""" - parser = argparse.ArgumentParser( - description="SSHCopyID - SSH 密钥部署工具", - usage="sshcopyid [--port PORT] [--keypath KEYPATH]", - ) - parser.add_argument("hostname", type=str, help="远程服务器主机名或 IP 地址") - parser.add_argument("username", type=str, help="远程服务器用户名") - parser.add_argument("password", type=str, help="远程服务器密码") - parser.add_argument("--port", type=int, default=22, help="SSH 端口 (默认: 22)") - parser.add_argument("--keypath", type=str, default="~/.ssh/id_rsa.pub", help="公钥文件路径") - parser.add_argument("--timeout", type=int, default=30, help="SSH 操作超时秒数 (默认: 30)") - args = parser.parse_args() - - px.run_yaml( - _CONFIG, - jobs="deploy", - variables={ - "HOSTNAME": args.hostname, - "USERNAME": args.username, - "PASSWORD": args.password, - "PORT": args.port, - "KEYPATH": args.keypath, - "TIMEOUT": args.timeout, - }, - ) diff --git a/src/pyflowx/yaml_loader.py b/src/pyflowx/yaml_loader.py index ccf6796..a5e0466 100644 --- a/src/pyflowx/yaml_loader.py +++ b/src/pyflowx/yaml_loader.py @@ -137,13 +137,15 @@ _ENV_VAR_EQUALS_RE = re.compile(r"^env\.(\w+)\s*==\s*['\"]([^'\"]*)['\"]$") _ENV_VAR_NOT_EQUALS_RE = re.compile(r"^env\.(\w+)\s*!=\s*['\"]([^'\"]*)['\"]$") # 支持的 job 字段名（连字符形式） -_JOB_FIELDS_HYPHEN = frozenset({ - "continue-on-error", - "skip-if-missing", - "allow-upstream-skip", - "concurrency-key", - "runs-on", -}) +_JOB_FIELDS_HYPHEN = frozenset( + { + "continue-on-error", + "skip-if-missing", + "allow-upstream-skip", + "concurrency-key", + "runs-on", + } +) class YamlLoadError(px.PyFlowXError): @@ -1028,6 +1030,9 @@ def run_cli(path: str | Path, argv: Sequence[str] | None = None) -> int: # noqa variables = _extract_variables(args, cli_config) + job_display = job_name if job_name else "全部任务" + print(f"[{p.stem}] 执行: {job_display}", flush=True) + try: run_yaml( p, diff --git a/tests/cli/test_autofmt.py b/tests/cli/test_autofmt.py index 09f8791..38d8ac9 100644 --- a/tests/cli/test_autofmt.py +++ b/tests/cli/test_autofmt.py @@ -5,8 +5,6 @@ from __future__ import annotations from pathlib import Path from unittest.mock import MagicMock, patch -import pyflowx as px -from pyflowx.cli import autofmt from pyflowx.cli._ops import dev @@ -218,85 +216,3 @@ class TestFormatAll: dev.format_all(tmp_path) # Should call ruff format and ruff check assert mock_run.call_count == 2 - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_fmt_default_target(self) -> None: - """main() should handle fmt command with default target.""" - with patch("sys.argv", ["autofmt", "fmt"]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_fmt_custom_target(self) -> None: - """main() should handle fmt command with custom target.""" - with patch("sys.argv", ["autofmt", "fmt", "--target", "src"]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_lint_default_target(self) -> None: - """main() should handle lint command with default target.""" - with patch("sys.argv", ["autofmt", "lint"]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_lint_with_fix(self) -> None: - """main() should handle lint command with fix.""" - with patch("sys.argv", ["autofmt", "lint", "--fix"]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_lint_custom_target(self) -> None: - """main() should handle lint command with custom target.""" - with patch("sys.argv", ["autofmt", "lint", "--target", "src"]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_doc_default_root(self) -> None: - """main() should handle doc command with default root.""" - with patch("sys.argv", ["autofmt", "doc"]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_doc_custom_root(self) -> None: - """main() should handle doc command with custom root.""" - with patch("sys.argv", ["autofmt", "doc", "--root-dir", "src"]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_sync_default_root(self) -> None: - """main() should handle sync command with default root.""" - with patch("sys.argv", ["autofmt", "sync"]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_sync_custom_root(self) -> None: - """main() should handle sync command with custom root.""" - with patch("sys.argv", ["autofmt", "sync", "--root-dir", "."]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help.""" - with patch("sys.argv", ["autofmt"]), patch.object(autofmt, "main"): - # Just call main, it should show help and return - autofmt.main() - # main() should return without calling px.run - assert True - - def test_main_creates_task_specs_with_verbose(self) -> None: - """main() should create TaskSpecs with verbose=True.""" - with patch("sys.argv", ["autofmt", "fmt"]), patch.object(px, "run") as mock_run: - autofmt.main() - assert mock_run.called - - def test_main_uses_thread_strategy(self) -> None: - """main() should use thread strategy.""" - with patch("sys.argv", ["autofmt", "fmt"]), patch.object(px, "run") as mock_run: - autofmt.main() - # Check that strategy="thread" was used - assert mock_run.called diff --git a/tests/cli/test_bumpversion.py b/tests/cli/test_bumpversion.py index 6cebc2a..4696361 100644 --- a/tests/cli/test_bumpversion.py +++ b/tests/cli/test_bumpversion.py @@ -4,12 +4,9 @@ from __future__ import annotations import os from pathlib import Path -from unittest.mock import patch import pytest -import pyflowx as px -from pyflowx.cli import bumpversion from pyflowx.cli._ops import dev @@ -273,57 +270,3 @@ class TestEdgeCases: # 验证最终结果 assert test_file.read_text(encoding="utf-8") == '__version__ = "2.0.0"' - - -class TestBumpVersionCli: - """Test bumpversion CLI.""" - - def test_minor(self, tmp_path: Path) -> None: - """Should handle minor version bump.""" - test_file = tmp_path / "__init__.py" - test_file.write_text('__version__ = "1.0.0"', encoding="utf-8") - - # Mock px.run: 只真正执行第一次调用(版本更新),其余返回空 dict - # run_yaml 内部调用 px.run 时传递 max_workers/dry_run 等 kwargs, 故用 *args/**kwargs 兼容 - with patch("sys.argv", ["bumpversion", "minor", "--no-tag"]), patch("pyflowx.run") as mock_run: - - def run_side_effect(*args: object, **kwargs: object) -> dict[str, object]: - # 执行实际版本更新任务 - graph = args[0] if args else kwargs.get("graph") - assert isinstance(graph, px.Graph) - results: dict[str, object] = {} - for spec in graph.specs.values(): - if spec.fn is not None and spec.args: - results[spec.name] = spec.fn(*spec.args) - return results - - mock_run.side_effect = run_side_effect - bumpversion.main() - - # 验证版本号已更新 - assert test_file.read_text(encoding="utf-8") == '__version__ = "1.1.0"' - - def test_no_valid_files(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: - """Should handle no valid files.""" - test_file = tmp_path / "test.txt" - test_file.write_text("这是一个测试文件", encoding="utf-8") - - # run_yaml 内部调用 px.run 时传递 max_workers/dry_run 等 kwargs, 故用 *args/**kwargs 兼容 - with patch("sys.argv", ["bumpversion", "minor", "--no-tag"]), patch("pyflowx.run") as mock_run: - - def run_side_effect(*args: object, **kwargs: object) -> dict[str, object]: - # 执行实际版本更新任务 - graph = args[0] if args else kwargs.get("graph") - assert isinstance(graph, px.Graph) - results: dict[str, object] = {} - for spec in graph.specs.values(): - if spec.fn is not None and spec.args: - results[spec.name] = spec.fn(*spec.args) - return results - - mock_run.side_effect = run_side_effect - bumpversion.main() - - # 验证未更新任何文件 - assert test_file.read_text(encoding="utf-8") == "这是一个测试文件" - assert "未找到包含版本号的文件" in capsys.readouterr().out diff --git a/tests/cli/test_filedate.py b/tests/cli/test_filedate.py index aa3c2c0..98c3fbd 100644 --- a/tests/cli/test_filedate.py +++ b/tests/cli/test_filedate.py @@ -3,10 +3,7 @@ from __future__ import annotations from pathlib import Path -from unittest.mock import patch -import pyflowx as px -from pyflowx.cli import filedate from pyflowx.cli._ops import files @@ -104,34 +101,3 @@ class TestProcessFilesDate: files.process_files_date(file_list, clear=False) # All files should be processed - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_add_command(self, tmp_path: Path) -> None: - """main() should handle add command.""" - test_file = tmp_path / "test.txt" - test_file.write_text("test content") - - with patch("sys.argv", ["filedate", "add", str(test_file)]), patch.object(px, "run") as mock_run: - filedate.main() - assert mock_run.called - - def test_main_clear_command(self, tmp_path: Path) -> None: - """main() should handle clear command.""" - test_file = tmp_path / "20240101_test.txt" - test_file.write_text("test content") - - with patch("sys.argv", ["filedate", "clear", str(test_file)]), patch.object(px, "run") as mock_run: - filedate.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help.""" - with patch("sys.argv", ["filedate"]): - filedate.main() - # Should print help and return diff --git a/tests/cli/test_filelevel.py b/tests/cli/test_filelevel.py index 4282413..4e8205e 100644 --- a/tests/cli/test_filelevel.py +++ b/tests/cli/test_filelevel.py @@ -3,10 +3,7 @@ from __future__ import annotations from pathlib import Path -from unittest.mock import patch -import pyflowx as px -from pyflowx.cli import filelevel from pyflowx.cli._ops import files @@ -97,38 +94,3 @@ class TestProcessFilesLevel: files.process_files_level(file_list, level=1) # All files should be processed - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_set_command(self, tmp_path: Path) -> None: - """main() should handle set command.""" - test_file = tmp_path / "test.txt" - test_file.write_text("test content") - - with patch("sys.argv", ["filelevel", "set", str(test_file), "--level", "1"]), patch.object( - px, "run" - ) as mock_run: - filelevel.main() - assert mock_run.called - - def test_main_set_command_level_2(self, tmp_path: Path) -> None: - """main() should handle set command with level 2.""" - test_file = tmp_path / "test.txt" - test_file.write_text("test content") - - with patch("sys.argv", ["filelevel", "set", str(test_file), "--level", "2"]), patch.object( - px, "run" - ) as mock_run: - filelevel.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help.""" - with patch("sys.argv", ["filelevel"]): - filelevel.main() - # Should print help and return diff --git a/tests/cli/test_folderback.py b/tests/cli/test_folderback.py index 050c68b..aea40a3 100644 --- a/tests/cli/test_folderback.py +++ b/tests/cli/test_folderback.py @@ -5,8 +5,6 @@ from __future__ import annotations from pathlib import Path from unittest.mock import patch -import pyflowx as px -from pyflowx.cli import folderback from pyflowx.cli._ops import files @@ -159,20 +157,3 @@ class TestRegisteredFunctions: """folderback_default should be a registered callable.""" # folderback_default 现在是通过 @px.register_fn 注册的普通函数, 不是 TaskSpec assert callable(files.folderback_default) - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_calls_run_yaml(self) -> None: - """main() should parse args and call px.run_yaml().""" - with patch("sys.argv", ["folderback", "--src", ".", "--dst", "./backup"]), patch.object( - px, "run_yaml" - ) as mock_run_yaml: - folderback.main() - assert mock_run_yaml.called - # 验证 jobs 参数为 "backup" (YAML 中的 job 名) - assert mock_run_yaml.call_args.kwargs["jobs"] == "backup" diff --git a/tests/cli/test_folderzip.py b/tests/cli/test_folderzip.py index 636ffd1..ea21556 100644 --- a/tests/cli/test_folderzip.py +++ b/tests/cli/test_folderzip.py @@ -5,8 +5,6 @@ from __future__ import annotations from pathlib import Path from unittest.mock import patch -import pyflowx as px -from pyflowx.cli import folderzip from pyflowx.cli._ops import files @@ -61,18 +59,3 @@ class TestRegisteredFunctions: """folderzip_default should be a registered callable.""" # folderzip_default 现在是通过 @px.register_fn 注册的普通函数, 不是 TaskSpec assert callable(files.folderzip_default) - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_calls_run_yaml(self) -> None: - """main() should parse args and call px.run_yaml().""" - with patch("sys.argv", ["folderzip", "--cwd", "."]), patch.object(px, "run_yaml") as mock_run_yaml: - folderzip.main() - assert mock_run_yaml.called - # 验证 jobs 参数为 "zip" (YAML 中的 job 名) - assert mock_run_yaml.call_args.kwargs["jobs"] == "zip" diff --git a/tests/cli/test_gittool.py b/tests/cli/test_gittool.py index f76ad4c..bc37d53 100644 --- a/tests/cli/test_gittool.py +++ b/tests/cli/test_gittool.py @@ -8,7 +8,6 @@ from unittest.mock import patch import pytest import pyflowx as px -from pyflowx.cli import gittool from pyflowx.cli._ops import dev @@ -93,36 +92,3 @@ class TestInitSubDirs: dev.init_sub_dirs() # Should not call px.run assert mock_run.call_count == 0 - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_calls_run_yaml(self) -> None: - """main() should parse args and call px.run_yaml().""" - # 使用有效命令 p (推送), mock run_yaml 避免实际执行 git 命令 - with patch("sys.argv", ["gittool", "p"]), patch.object(px, "run_yaml") as mock_run_yaml: - gittool.main() - assert mock_run_yaml.called - # 验证 jobs 参数为 "p" (YAML 中的 job 名) - assert mock_run_yaml.call_args.kwargs["jobs"] == "p" - - def test_main_with_valid_command(self) -> None: - """main() should handle valid commands.""" - # 测试各个有效命令均能调用 run_yaml - for cmd in ("a", "c", "i", "isub", "p", "pl"): - with patch("sys.argv", ["gittool", cmd]), patch.object(px, "run_yaml") as mock_run_yaml: - gittool.main() - assert mock_run_yaml.called - assert mock_run_yaml.call_args.kwargs["jobs"] == cmd - - def test_main_with_no_args_shows_help(self, capsys: pytest.CaptureFixture[str]) -> None: - """main() with no args should show help and return.""" - # 新的 main() 在无有效命令时打印 help 并 return, 不再 sys.exit - with patch("sys.argv", ["gittool"]): - gittool.main() - captured = capsys.readouterr() - assert "Gittool" in captured.out or "usage" in captured.out diff --git a/tests/cli/test_lscalc.py b/tests/cli/test_lscalc.py index 46b80a0..00b67bc 100644 --- a/tests/cli/test_lscalc.py +++ b/tests/cli/test_lscalc.py @@ -5,8 +5,6 @@ from __future__ import annotations from pathlib import Path from unittest.mock import MagicMock, patch -import pyflowx as px -from pyflowx.cli import lscalc from pyflowx.cli._ops import system from pyflowx.conditions import Constants @@ -110,49 +108,3 @@ class TestCheckLsDynaStatus: mock_run.return_value = MagicMock(stdout="1234", returncode=0) system.check_ls_dyna_status() assert mock_run.called - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_run_command(self, tmp_path: Path) -> None: - """main() should handle run command.""" - input_file = tmp_path / "input.k" - input_file.write_text("LS-DYNA input") - - with patch("sys.argv", ["lscalc", "run", str(input_file)]), patch.object(px, "run") as mock_run: - lscalc.main() - assert mock_run.called - - def test_main_run_command_with_ncpu(self, tmp_path: Path) -> None: - """main() should handle run command with ncpu.""" - input_file = tmp_path / "input.k" - input_file.write_text("LS-DYNA input") - - with patch("sys.argv", ["lscalc", "run", str(input_file), "--ncpu", "8"]), patch.object(px, "run") as mock_run: - lscalc.main() - assert mock_run.called - - def test_main_mpi_command(self, tmp_path: Path) -> None: - """main() should handle mpi command.""" - input_file = tmp_path / "input.k" - input_file.write_text("LS-DYNA input") - - with patch("sys.argv", ["lscalc", "mpi", str(input_file)]), patch.object(px, "run") as mock_run: - lscalc.main() - assert mock_run.called - - def test_main_status_command(self) -> None: - """main() should handle status command.""" - with patch("sys.argv", ["lscalc", "status"]), patch.object(px, "run") as mock_run: - lscalc.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help.""" - with patch("sys.argv", ["lscalc"]): - lscalc.main() - # Should print help and return diff --git a/tests/cli/test_packtool.py b/tests/cli/test_packtool.py index c6a3f9a..1daec6b 100644 --- a/tests/cli/test_packtool.py +++ b/tests/cli/test_packtool.py @@ -7,8 +7,6 @@ from unittest.mock import MagicMock, patch import pytest -import pyflowx as px -from pyflowx.cli import packtool from pyflowx.cli._ops import system @@ -259,67 +257,3 @@ class TestCleanBuildDir: system.clean_build_dir(build_dir) # Should print message - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_src_command(self, tmp_path: Path) -> None: - """main() should handle src command.""" - project_dir = tmp_path / "project" - project_dir.mkdir() - - with patch("sys.argv", ["packtool", "src", "--project-dir", str(project_dir)]), patch.object( - px, "run" - ) as mock_run: - packtool.main() - assert mock_run.called - - def test_main_deps_command(self, tmp_path: Path) -> None: - """main() should handle deps command.""" - with patch("sys.argv", ["packtool", "deps", "numpy", "pandas"]), patch.object(px, "run") as mock_run: - packtool.main() - assert mock_run.called - - def test_main_wheel_command(self, tmp_path: Path) -> None: - """main() should handle wheel command.""" - project_dir = tmp_path / "project" - project_dir.mkdir() - - with patch("sys.argv", ["packtool", "wheel", "--project-dir", str(project_dir)]), patch.object( - px, "run" - ) as mock_run: - packtool.main() - assert mock_run.called - - def test_main_embed_command(self, tmp_path: Path) -> None: - """main() should handle embed command.""" - with patch("sys.argv", ["packtool", "embed", "--version", "3.10"]), patch.object(px, "run") as mock_run: - packtool.main() - assert mock_run.called - - def test_main_zip_command(self, tmp_path: Path) -> None: - """main() should handle zip command.""" - source_dir = tmp_path / "source" - source_dir.mkdir() - - with patch("sys.argv", ["packtool", "zip", "--source-dir", str(source_dir)]), patch.object( - px, "run" - ) as mock_run: - packtool.main() - assert mock_run.called - - def test_main_clean_command(self) -> None: - """main() should handle clean command.""" - with patch("sys.argv", ["packtool", "clean"]), patch.object(px, "run") as mock_run: - packtool.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help.""" - with patch("sys.argv", ["packtool"]): - packtool.main() - # Should print help and return diff --git a/tests/cli/test_pdftool.py b/tests/cli/test_pdftool.py index feea3dc..8e0e29c 100644 --- a/tests/cli/test_pdftool.py +++ b/tests/cli/test_pdftool.py @@ -8,8 +8,6 @@ from unittest.mock import MagicMock, patch import pytest -import pyflowx as px -from pyflowx.cli import pdftool from pyflowx.cli._ops import media @@ -280,46 +278,3 @@ class TestPdfRepair: mock_fitz_open.return_value = mock_doc media.pdf_repair(input_file, output_file) assert mock_doc.save.called - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_merge_command(self, tmp_path: Path) -> None: - """main() should handle merge command.""" - input_files = [tmp_path / "input1.pdf", tmp_path / "input2.pdf"] - for f in input_files: - f.write_bytes(b"PDF content") - - with patch("sys.argv", ["pdftool", "m", str(input_files[0]), str(input_files[1])]), patch.object( - px, "run" - ) as mock_run: - pdftool.main() - assert mock_run.called - - def test_main_split_command(self, tmp_path: Path) -> None: - """main() should handle split command.""" - input_file = tmp_path / "input.pdf" - input_file.write_bytes(b"PDF content") - - with patch("sys.argv", ["pdftool", "s", str(input_file)]), patch.object(px, "run") as mock_run: - pdftool.main() - assert mock_run.called - - def test_main_compress_command(self, tmp_path: Path) -> None: - """main() should handle compress command.""" - input_file = tmp_path / "input.pdf" - input_file.write_bytes(b"PDF content") - - with patch("sys.argv", ["pdftool", "c", str(input_file)]), patch.object(px, "run") as mock_run: - pdftool.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help.""" - with patch("sys.argv", ["pdftool"]): - pdftool.main() - # Should print help and return diff --git a/tests/cli/test_piptool.py b/tests/cli/test_piptool.py index eb70699..1c32826 100644 --- a/tests/cli/test_piptool.py +++ b/tests/cli/test_piptool.py @@ -6,8 +6,6 @@ import subprocess from pathlib import Path from unittest.mock import MagicMock, patch -import pyflowx as px -from pyflowx.cli import piptool from pyflowx.cli._ops import dev @@ -204,52 +202,3 @@ class TestPipFreeze: mock_run.return_value = MagicMock(stdout="numpy==1.0.0\npandas==2.0.0", returncode=0) dev.pip_freeze() assert mock_run.called - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_install_command(self) -> None: - """main() should handle install command.""" - with patch("sys.argv", ["piptool", "i", "numpy", "pandas"]), patch.object(px, "run") as mock_run: - piptool.main() - assert mock_run.called - - def test_main_uninstall_command(self) -> None: - """main() should handle uninstall command.""" - with patch("sys.argv", ["piptool", "u", "numpy"]), patch.object(px, "run") as mock_run: - piptool.main() - assert mock_run.called - - def test_main_reinstall_command(self) -> None: - """main() should handle reinstall command.""" - with patch("sys.argv", ["piptool", "r", "numpy"]), patch.object(px, "run") as mock_run: - piptool.main() - assert mock_run.called - - def test_main_download_command(self) -> None: - """main() should handle download command.""" - with patch("sys.argv", ["piptool", "d", "numpy"]), patch.object(px, "run") as mock_run: - piptool.main() - assert mock_run.called - - def test_main_upgrade_command(self) -> None: - """main() should handle upgrade command.""" - with patch("sys.argv", ["piptool", "up"]), patch.object(px, "run") as mock_run: - piptool.main() - assert mock_run.called - - def test_main_freeze_command(self) -> None: - """main() should handle freeze command.""" - with patch("sys.argv", ["piptool", "f"]), patch.object(px, "run") as mock_run: - piptool.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help.""" - with patch("sys.argv", ["piptool"]): - piptool.main() - # Should print help and return diff --git a/tests/cli/test_screenshot.py b/tests/cli/test_screenshot.py index 41a2072..c5ea7f6 100644 --- a/tests/cli/test_screenshot.py +++ b/tests/cli/test_screenshot.py @@ -5,8 +5,6 @@ from __future__ import annotations from pathlib import Path from unittest.mock import MagicMock, patch -import pyflowx as px -from pyflowx.cli import screenshot from pyflowx.cli._ops import media from pyflowx.conditions import Constants @@ -97,28 +95,3 @@ class TestTakeScreenshotArea: mock_run.return_value = MagicMock(returncode=0) media.take_screenshot_area() assert mock_run.called - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_full_command(self, tmp_path: Path) -> None: - """main() should handle full command.""" - with patch("sys.argv", ["screenshot", "full"]), patch.object(px, "run") as mock_run: - screenshot.main() - assert mock_run.called - - def test_main_area_command(self, tmp_path: Path) -> None: - """main() should handle area command.""" - with patch("sys.argv", ["screenshot", "area"]), patch.object(px, "run") as mock_run: - screenshot.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help.""" - with patch("sys.argv", ["screenshot"]): - screenshot.main() - # Should print help and return diff --git a/tests/cli/test_sshcopyid.py b/tests/cli/test_sshcopyid.py index 610e6ac..bfc2375 100644 --- a/tests/cli/test_sshcopyid.py +++ b/tests/cli/test_sshcopyid.py @@ -8,8 +8,6 @@ from unittest.mock import MagicMock, patch import pytest -import pyflowx as px -from pyflowx.cli import sshcopyid from pyflowx.cli._ops import system @@ -97,69 +95,3 @@ class TestSshCopyId: # Verify timeout is used in ConnectTimeout option call_args = mock_run.call_args[0][0] assert "ConnectTimeout=60" in call_args - - -# ---------------------------------------------------------------------- # -# main function -# ---------------------------------------------------------------------- # -class TestMain: - """Test main function.""" - - def test_main_with_required_args(self) -> None: - """main() should handle required arguments.""" - with patch("sys.argv", ["sshcopyid", "localhost", "user", "password"]), patch.object( - px, "run" - ) as mock_run, patch.object(system, "ssh_copy_id"): - sshcopyid.main() - assert mock_run.called - graph = mock_run.call_args[0][0] - assert isinstance(graph, px.Graph) - - def test_main_with_custom_port(self) -> None: - """main() should handle custom port argument.""" - with patch("sys.argv", ["sshcopyid", "localhost", "user", "password", "--port", "2222"]), patch.object( - px, "run" - ) as mock_run, patch.object(system, "ssh_copy_id"): - sshcopyid.main() - assert mock_run.called - - def test_main_with_custom_keypath(self) -> None: - """main() should handle custom keypath argument.""" - with patch( - "sys.argv", ["sshcopyid", "localhost", "user", "password", "--keypath", "/custom/key.pub"] - ), patch.object(px, "run") as mock_run, patch.object(system, "ssh_copy_id"): - sshcopyid.main() - assert mock_run.called - - def test_main_with_custom_timeout(self) -> None: - """main() should handle custom timeout argument.""" - with patch("sys.argv", ["sshcopyid", "localhost", "user", "password", "--timeout", "60"]), patch.object( - px, "run" - ) as mock_run, patch.object(system, "ssh_copy_id"): - sshcopyid.main() - assert mock_run.called - - def test_main_with_no_args_shows_help(self) -> None: - """main() with no args should show help and exit.""" - with patch("sys.argv", ["sshcopyid"]), pytest.raises(SystemExit) as exc_info: - sshcopyid.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", ["sshcopyid", "localhost", "user", "password"]), patch.object( - px, "run" - ) as mock_run, patch.object(system, "ssh_copy_id"): - sshcopyid.main() - graph = mock_run.call_args[0][0] - task_names = list(graph.all_specs().keys()) - # YAML 中 job 名为 "deploy" (不再是 "ssh_deploy") - assert "deploy" in task_names - - def test_main_uses_thread_strategy(self) -> None: - """main() should use thread strategy.""" - with patch("sys.argv", ["sshcopyid", "localhost", "user", "password"]), patch.object( - px, "run" - ) as mock_run, patch.object(system, "ssh_copy_id"): - sshcopyid.main() - assert mock_run.call_args[1]["strategy"] == "thread"