zhou/pyflowx
1
0
Fork 0
Code Issues Pull Requests Actions 107 Packages Projects Releases 6 Wiki Activity
Files
8cfdef7516925ac29dbfdc58d852f48bbd0f45f2
pyflowx/tests/test_tools.py
T
zhou fca6f17a5c chore: 升级项目适配 Python 3.10+ 并重构代码 
1.  移除对 Python 3.8/3.9 的支持，更新 tox、pyproject.toml 配置
2.  替换 typing 导入为 collections.abc 标准库类型
3.  重构 CLI 系统从 argparse 迁移到 typer
4.  优化代码格式与类型注解，修复多处类型兼容问题
5.  更新依赖声明，移除 graphlib_backport 兼容包
2026-07-06 22:14:59 +08:00

1144 lines
32 KiB
Python
Raw Blame History

"""tests/test_tools.py - @px.tool 装饰器核心模块测试.
覆盖 tools.py 全部分支:注册/重复检测/typer CLI 生成/单命令执行/
cmd/fn/聚合/DAG/hidden/全局选项/退出码三态.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
import pytest
from pyflowx import tools
from pyflowx.executors import RunReport
from pyflowx.task import RetryPolicy
from pyflowx.tools import (
ToolExitCode,
_build_task_spec,
_build_typer_app,
_collect_with_deps,
_global_option_params,
_handle_list_rich,
_has_function_logic,
_is_aggregate,
_make_dispatcher,
_noop,
clear_tool_registry,
get_tool,
list_subcommands,
list_tools,
run_tool,
tool,
)
# ---------------------------------------------------------------------- #
# fixture
# ---------------------------------------------------------------------- #
@pytest.fixture(autouse=True)
def clear_registry() -> None:
"""每条用例前清空工具注册表,避免相互污染."""
clear_tool_registry()
@pytest.fixture
def mock_run(monkeypatch: pytest.MonkeyPatch) -> list[Any]:
"""mock pyflowx.tools.run,避免实际执行任务;记录调用参数."""
calls: list[Any] = []
def _fake_run(graph: Any, **kwargs: Any) -> RunReport:
calls.append((graph, kwargs))
return RunReport(success=True)
monkeypatch.setattr(tools, "run", _fake_run)
return calls
# ---------------------------------------------------------------------- #
# ToolExitCode + _noop
# ---------------------------------------------------------------------- #
class TestToolExitCode:
"""测试退出码枚举."""
def test_values(self) -> None:
assert ToolExitCode.SUCCESS == 0
assert ToolExitCode.FAILURE == 1
assert ToolExitCode.INTERRUPTED == 130
class TestNoop:
"""测试聚合任务占位函数."""
def test_returns_none(self) -> None:
assert _noop() is None
# ---------------------------------------------------------------------- #
# @tool 装饰器 + 注册表
# ---------------------------------------------------------------------- #
class TestToolDecorator:
"""测试 @tool 装饰器与注册表."""
def test_register_single_command(self) -> None:
"""单命令工具注册到注册表."""
@tool("mytool")
def myfunc(x: int = 1) -> None:
pass
assert "mytool" in list_tools()
spec = get_tool("mytool")
assert spec.name == "mytool"
assert spec.subcommand is None
assert spec.func is myfunc
def test_register_multi_subcommand(self) -> None:
"""同一工具名多 subcommand 注册."""
@tool("pdftool", subcommand="m", help="合并")
def merge(inputs: list[Path]) -> None:
pass
@tool("pdftool", subcommand="s", help="拆分")
def split(input: Path) -> None:
pass
assert list_subcommands("pdftool") == ["m", "s"]
assert get_tool("pdftool", "m").func is merge
assert get_tool("pdftool", "s").func is split
def test_help_defaults_to_docstring(self) -> None:
"""help 为空时使用函数 docstring."""
@tool("t")
def f(x: int = 1) -> None:
"""我的文档."""
assert get_tool("t").help == "我的文档."
def test_help_explicit_overrides_docstring(self) -> None:
"""显式 help 优先于 docstring."""
@tool("t", help="显式帮助")
def f(x: int = 1) -> None:
"""文档."""
assert get_tool("t").help == "显式帮助"
def test_cmd_tuple_conversion(self) -> None:
"""cmd list 转为 tuple 存储."""
@tool("t", cmd=["echo", "hi"])
def f() -> None:
pass
spec = get_tool("t")
assert spec.cmd == ("echo", "hi")
def test_cmd_str_kept_as_str(self) -> None:
"""cmd str 保持 str."""
@tool("t", cmd="echo hi")
def f() -> None:
pass
assert get_tool("t").cmd == "echo hi"
def test_needs_tuple_conversion(self) -> None:
"""needs list 转为 tuple."""
@tool("t", subcommand="a", needs=["b", "c"])
def a() -> None:
pass
@tool("t", subcommand="b")
def b() -> None:
pass
@tool("t", subcommand="c")
def c() -> None:
pass
assert get_tool("t", "a").needs == ("b", "c")
def test_env_dict_copy(self) -> None:
"""env dict 被复制(避免外部修改)."""
env = {"KEY": "val"}
@tool("t", env=env)
def f() -> None:
pass
spec = get_tool("t")
assert spec.env == {"KEY": "val"}
env["KEY"] = "modified"
assert spec.env == {"KEY": "val"}
def test_retry_passthrough(self) -> None:
"""retry 透传 ToolSpec."""
retry = RetryPolicy(max_attempts=3)
@tool("t", retry=retry)
def f() -> None:
pass
assert get_tool("t").retry is retry
def test_timeout_passthrough(self) -> None:
"""timeout 透传."""
@tool("t", timeout=30.0)
def f() -> None:
pass
assert get_tool("t").timeout == 30.0
def test_strategy_passthrough(self) -> None:
"""strategy 透传."""
@tool("t", strategy="thread")
def f() -> None:
pass
assert get_tool("t").strategy == "thread"
def test_cwd_passthrough(self) -> None:
"""cwd 透传."""
@tool("t", cwd="/tmp")
def f() -> None:
pass
assert get_tool("t").cwd == "/tmp"
def test_allow_upstream_skip(self) -> None:
"""allow_upstream_skip 透传."""
@tool("t", allow_upstream_skip=True)
def f() -> None:
pass
assert get_tool("t").allow_upstream_skip is True
def test_hidden(self) -> None:
"""hidden 透传."""
@tool("t", subcommand="h", hidden=True)
def h() -> None:
pass
@tool("t", subcommand="v")
def v() -> None:
pass
assert get_tool("t", "h").hidden is True
assert list_subcommands("t") == ["v"]
assert list_subcommands("t", include_hidden=True) == ["h", "v"]
class TestRegistryErrors:
"""测试注册表错误分支."""
def test_duplicate_subcommand_raises(self) -> None:
"""同名同 subcommand 二次注册抛 ValueError."""
@tool("t", subcommand="a")
def a1() -> None:
pass
with pytest.raises(ValueError, match="已注册"):
@tool("t", subcommand="a")
def a2() -> None:
pass
def test_get_tool_unknown_tool(self) -> None:
"""获取未注册工具抛 KeyError."""
with pytest.raises(KeyError, match="未注册"):
get_tool("unknown")
def test_get_tool_unknown_subcommand(self) -> None:
"""获取未注册子命令抛 KeyError."""
@tool("t", subcommand="a")
def a() -> None:
pass
with pytest.raises(KeyError, match="没有子命令"):
get_tool("t", "b")
def test_list_subcommands_unknown_tool(self) -> None:
"""未注册工具返回空列表."""
assert list_subcommands("unknown") == []
def test_list_tools_sorted(self) -> None:
"""list_tools 返回排序后的工具名."""
@tool("ztool")
def z() -> None:
pass
@tool("atool")
def a() -> None:
pass
@tool("mtool")
def m() -> None:
pass
assert list_tools() == ["atool", "mtool", "ztool"]
# ---------------------------------------------------------------------- #
# _collect_with_deps
# ---------------------------------------------------------------------- #
class TestCollectWithDeps:
"""测试依赖收集 BFS."""
def test_no_deps(self) -> None:
"""无依赖 → 仅 target."""
@tool("t", subcommand="a")
def a() -> None:
pass
result = _collect_with_deps("t", "a")
assert result == ["a"]
def test_with_deps(self) -> None:
"""有依赖 → 依赖在前, target 在后."""
@tool("t", subcommand="a", needs=["b"])
def a() -> None:
pass
@tool("t", subcommand="b", needs=["c"])
def b() -> None:
pass
@tool("t", subcommand="c")
def c() -> None:
pass
result = _collect_with_deps("t", "a")
assert result == ["c", "b", "a"]
def test_unknown_tool(self) -> None:
"""未注册工具 → [target]."""
assert _collect_with_deps("unknown", "x") == ["x"]
def test_diamond_deps(self) -> None:
"""菱形依赖:每个依赖只出现一次."""
@tool("t", subcommand="a", needs=["b", "c"])
def a() -> None:
pass
@tool("t", subcommand="b", needs=["d"])
def b() -> None:
pass
@tool("t", subcommand="c", needs=["d"])
def c() -> None:
pass
@tool("t", subcommand="d")
def d() -> None:
pass
result = _collect_with_deps("t", "a")
# d 只出现一次, a 在最后
assert result[-1] == "a"
assert result.count("d") == 1
# ---------------------------------------------------------------------- #
# _has_function_logic + _is_aggregate
# ---------------------------------------------------------------------- #
class TestHasFunctionLogic:
"""测试函数体逻辑判定."""
def test_pass_only(self) -> None:
def f() -> None:
pass
assert _has_function_logic(f) is False
def test_ellipsis_only(self) -> None:
def f() -> None: ...
assert _has_function_logic(f) is False
def test_docstring_only(self) -> None:
def f() -> None:
"""文档."""
assert _has_function_logic(f) is False
def test_docstring_plus_pass(self) -> None:
def f() -> None:
"""文档."""
pass
assert _has_function_logic(f) is False
def test_has_logic(self) -> None:
def f() -> None:
print("hi")
assert _has_function_logic(f) is True
def test_has_logic_with_docstring(self) -> None:
def f() -> None:
"""文档."""
x = 1
print(x)
assert _has_function_logic(f) is True
class TestIsAggregate:
"""测试聚合任务判定."""
def test_aggregate_pass(self) -> None:
"""有 needs 无 cmd + pass → 聚合."""
@tool("t", subcommand="agg", needs=["dep"])
def agg() -> None:
pass
@tool("t", subcommand="dep")
def dep() -> None:
pass
assert _is_aggregate(get_tool("t", "agg")) is True
def test_not_aggregate_has_cmd(self) -> None:
"""有 cmd → 非聚合."""
@tool("t", subcommand="c", cmd=["echo"], needs=["dep"])
def c() -> None:
pass
@tool("t", subcommand="dep")
def dep() -> None:
pass
assert _is_aggregate(get_tool("t", "c")) is False
def test_not_aggregate_no_needs(self) -> None:
"""无 needs → 非聚合."""
@tool("t", subcommand="f")
def f() -> None:
pass
assert _is_aggregate(get_tool("t", "f")) is False
def test_not_aggregate_has_logic(self) -> None:
"""有 needs 无 cmd 但函数体有逻辑 → 非聚合."""
@tool("t", subcommand="f", needs=["dep"])
def f() -> None:
print("hi")
@tool("t", subcommand="dep")
def dep() -> None:
pass
assert _is_aggregate(get_tool("t", "f")) is False
# ---------------------------------------------------------------------- #
# _build_task_spec
# ---------------------------------------------------------------------- #
class TestBuildTaskSpec:
"""测试 ToolSpec → TaskSpec 转换."""
def test_cmd_task(self) -> None:
"""cmd 任务:用 cmd 构建 TaskSpec, cwd 从装饰器取."""
@tool("t", subcommand="c", cmd=["echo", "hi"], cwd="/tmp")
def c() -> None:
pass
spec = get_tool("t", "c")
task = _build_task_spec(spec, {})
assert task.name == "c"
assert task.cmd == ["echo", "hi"]
assert task.cwd == Path("/tmp")
def test_cmd_task_cwd_from_variables(self) -> None:
"""cmd 任务:cwd 从 variables['cwd'] 优先取."""
@tool("t", subcommand="c", cmd=["echo"], cwd="/tmp")
def c() -> None:
pass
spec = get_tool("t", "c")
task = _build_task_spec(spec, {"cwd": "/other"})
assert task.cwd == Path("/other")
def test_cmd_task_no_cwd(self) -> None:
"""cmd 任务:无 cwd → None."""
@tool("t", subcommand="c", cmd=["echo"])
def c() -> None:
pass
spec = get_tool("t", "c")
task = _build_task_spec(spec, {})
assert task.cwd is None
def test_aggregate_task(self) -> None:
"""聚合任务:fn=_noop, 有 needs."""
@tool("t", subcommand="agg", needs=["dep"], strategy="thread")
def agg() -> None:
pass
@tool("t", subcommand="dep")
def dep() -> None:
pass
spec = get_tool("t", "agg")
task = _build_task_spec(spec, {})
assert task.name == "agg"
assert task.fn is _noop
assert task.depends_on == ("dep",)
assert task.strategy == "thread"
def test_fn_task(self) -> None:
"""fn 任务:fn=func, kwargs 透传."""
@tool("t", subcommand="f")
def f(x: int = 1, y: str = "z") -> None:
pass
spec = get_tool("t", "f")
task = _build_task_spec(spec, {"x": 5, "y": "a"})
assert task.name == "f"
assert task.fn is f
assert task.kwargs == {"x": 5, "y": "a"}
def test_fn_task_partial_variables(self) -> None:
"""fn 任务:仅取签名内的变量."""
@tool("t", subcommand="f")
def f(x: int = 1) -> None:
pass
spec = get_tool("t", "f")
task = _build_task_spec(spec, {"x": 5, "extra": "ignored"})
assert task.kwargs == {"x": 5}
def test_fn_task_cwd_from_variables(self) -> None:
"""fn 任务:cwd 从 variables['cwd'] 取."""
@tool("t", subcommand="f")
def f(x: int = 1) -> None:
pass
spec = get_tool("t", "f")
task = _build_task_spec(spec, {"x": 5, "cwd": "/tmp"})
assert task.cwd == Path("/tmp")
def test_fn_task_no_cwd(self) -> None:
"""fn 任务:无 cwd → None."""
@tool("t", subcommand="f")
def f(x: int = 1) -> None:
pass
spec = get_tool("t", "f")
task = _build_task_spec(spec, {"x": 5})
assert task.cwd is None
def test_env_passthrough(self) -> None:
"""env 透传 TaskSpec."""
@tool("t", subcommand="f", env={"K": "v"})
def f() -> None:
pass
spec = get_tool("t", "f")
task = _build_task_spec(spec, {})
assert task.env == {"K": "v"}
def test_retry_default(self) -> None:
"""无 retry → 默认 RetryPolicy()."""
@tool("t", subcommand="f")
def f() -> None:
pass
spec = get_tool("t", "f")
task = _build_task_spec(spec, {})
assert isinstance(task.retry, RetryPolicy)
def test_retry_custom(self) -> None:
"""有 retry → 透传."""
retry = RetryPolicy(max_attempts=5)
@tool("t", subcommand="f", retry=retry)
def f() -> None:
pass
spec = get_tool("t", "f")
task = _build_task_spec(spec, {})
assert task.retry is retry
def test_allow_upstream_skip_passthrough(self) -> None:
"""allow_upstream_skip 透传."""
@tool("t", subcommand="f", allow_upstream_skip=True)
def f() -> None:
pass
spec = get_tool("t", "f")
task = _build_task_spec(spec, {})
assert task.allow_upstream_skip is True
def test_single_command_name_uses_tool_name(self) -> None:
"""单命令工具(subcommand=None):task name 用工具名."""
@tool("solo")
def f(x: int = 1) -> None:
pass
spec = get_tool("solo")
task = _build_task_spec(spec, {})
assert task.name == "solo"
# ---------------------------------------------------------------------- #
# _global_option_params
# ---------------------------------------------------------------------- #
class TestGlobalOptionParams:
"""测试全局选项参数构建."""
def test_count(self) -> None:
"""返回 3 个参数 (dry_run, quiet, strategy)."""
params = _global_option_params()
assert len(params) == 3
def test_names(self) -> None:
"""参数名正确."""
params = _global_option_params()
assert [p.name for p in params] == ["dry_run", "quiet", "strategy"]
def test_all_keyword_only(self) -> None:
"""所有参数为 KEYWORD_ONLY."""
import inspect
params = _global_option_params()
assert all(p.kind == inspect.Parameter.KEYWORD_ONLY for p in params)
def test_annotations(self) -> None:
"""注解正确."""
params = _global_option_params()
assert params[0].annotation is bool # dry_run
assert params[1].annotation is bool # quiet
# strategy: str | None
assert params[2].annotation is not None
# ---------------------------------------------------------------------- #
# _make_dispatcher
# ---------------------------------------------------------------------- #
class TestMakeDispatcher:
"""测试 typer 命令 dispatcher 构建."""
def test_signature_includes_user_params(self) -> None:
"""dispatcher 签名包含用户函数参数."""
@tool("t", subcommand="a")
def a(x: int = 1, y: str = "z") -> None:
pass
import inspect
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
sig = inspect.signature(dispatcher)
assert "x" in sig.parameters
assert "y" in sig.parameters
def test_signature_includes_global_options(self) -> None:
"""dispatcher 签名包含全局选项."""
@tool("t", subcommand="a")
def a(x: int = 1) -> None:
pass
import inspect
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
sig = inspect.signature(dispatcher)
assert "dry_run" in sig.parameters
assert "quiet" in sig.parameters
assert "strategy" in sig.parameters
def test_signature_filters_var_positional(self) -> None:
"""*args 被过滤."""
@tool("t", subcommand="a")
def a(*args: Any, x: int = 1) -> None:
pass
import inspect
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
sig = inspect.signature(dispatcher)
assert "args" not in sig.parameters
assert "x" in sig.parameters
def test_signature_filters_var_keyword(self) -> None:
"""**kwargs 被过滤."""
@tool("t", subcommand="a")
def a(x: int = 1, **kw: Any) -> None:
pass
import inspect
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
sig = inspect.signature(dispatcher)
assert "kw" not in sig.parameters
def test_dispatcher_name_and_doc(self) -> None:
"""dispatcher 继承函数名和 docstring."""
@tool("t", subcommand="a", help="帮助文本")
def a(x: int = 1) -> None:
pass
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
assert dispatcher.__name__ == "a"
assert dispatcher.__doc__ == "帮助文本"
# ---------------------------------------------------------------------- #
# _build_typer_app
# ---------------------------------------------------------------------- #
class TestBuildTyperApp:
"""测试 typer app 构建."""
def test_single_command_app(self) -> None:
"""单命令工具 → typer app 有 callback."""
@tool("solo")
def f(x: int = 1) -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
app = _build_typer_app("solo", _TOOL_REGISTRY["solo"])
assert app is not None
def test_multi_subcommand_app(self) -> None:
"""多 subcommand 工具 → typer app 有 commands."""
@tool("t", subcommand="a")
def a(x: int = 1) -> None:
pass
@tool("t", subcommand="b")
def b(y: str = "z") -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
app = _build_typer_app("t", _TOOL_REGISTRY["t"])
assert app is not None
def test_hidden_not_registered_as_command(self) -> None:
"""hidden subcommand 不注册为 typer 命令."""
@tool("t", subcommand="visible")
def visible() -> None:
pass
@tool("t", subcommand="hidden", hidden=True)
def hidden() -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
app = _build_typer_app("t", _TOOL_REGISTRY["t"])
# typer app 的 registered_commands 不应包含 hidden
command_names = [cmd.name for cmd in app.registered_commands]
assert "visible" in command_names
assert "hidden" not in command_names
# ---------------------------------------------------------------------- #
# _handle_list_rich
# ---------------------------------------------------------------------- #
class TestHandleListRich:
"""测试 rich 表格任务列表输出."""
def test_list_output(self, capsys: pytest.CaptureFixture[str]) -> None:
"""--list 打印 rich 任务列表."""
@tool("t", subcommand="a", needs=["b"])
def a() -> None:
pass
@tool("t", subcommand="b", cmd=["echo"])
def b() -> None:
pass
@tool("t", subcommand="h", hidden=True)
def h() -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
_handle_list_rich("t", _TOOL_REGISTRY["t"])
out = capsys.readouterr().out
assert "工具 t 的任务列表" in out
assert "a" in out
assert "b" in out
assert "h" in out # hidden 也在 list 中显示
assert "cmd" in out # b 是 cmd 任务
assert "隐藏" in out # h 是 hidden
def test_list_aggregate_type(self, capsys: pytest.CaptureFixture[str]) -> None:
"""聚合任务显示 '聚合' 类型."""
@tool("t", subcommand="agg", needs=["dep"])
def agg() -> None:
pass
@tool("t", subcommand="dep")
def dep() -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
_handle_list_rich("t", _TOOL_REGISTRY["t"])
out = capsys.readouterr().out
assert "聚合" in out
def test_list_fn_type(self, capsys: pytest.CaptureFixture[str]) -> None:
"""fn 任务显示 'fn' 类型."""
@tool("t", subcommand="f")
def f() -> None:
print("hi")
from pyflowx.tools import _TOOL_REGISTRY
_handle_list_rich("t", _TOOL_REGISTRY["t"])
out = capsys.readouterr().out
assert "fn" in out
# ---------------------------------------------------------------------- #
# run_tool
# ---------------------------------------------------------------------- #
class TestRunTool:
"""测试 run_tool 主入口."""
def test_unknown_tool(self, capsys: pytest.CaptureFixture[str]) -> None:
"""未注册工具 → 退出码 1 + stderr 提示."""
ret = run_tool("unknown", [])
assert ret == ToolExitCode.FAILURE
err = capsys.readouterr().err
assert "未注册工具" in err
def test_single_command_success(self, mock_run: list[Any]) -> None:
"""单命令工具执行成功 → 0."""
@tool("solo")
def f(x: int = 1) -> None:
pass
ret = run_tool("solo", ["--x", "5"])
assert ret == ToolExitCode.SUCCESS
assert len(mock_run) == 1
_graph, kwargs = mock_run[0]
assert kwargs["dry_run"] is False
assert kwargs["verbose"] is True
def test_multi_subcommand_success(self, mock_run: list[Any]) -> None:
"""多 subcommand 工具执行成功 → 0."""
@tool("t", subcommand="a")
def a(x: int = 1) -> None:
pass
@tool("t", subcommand="b")
def b(y: str = "z") -> None:
pass
ret = run_tool("t", ["a", "--x", "5"])
assert ret == ToolExitCode.SUCCESS
def test_dry_run(self, mock_run: list[Any]) -> None:
"""--dry-run 透传."""
@tool("t")
def f() -> None:
pass
run_tool("t", ["--dry-run"])
_, kwargs = mock_run[0]
assert kwargs["dry_run"] is True
def test_quiet(self, mock_run: list[Any]) -> None:
"""--quiet → verbose=False."""
@tool("t")
def f() -> None:
pass
run_tool("t", ["--quiet"])
_, kwargs = mock_run[0]
assert kwargs["verbose"] is False
def test_strategy_override(self, mock_run: list[Any]) -> None:
"""--strategy CLI 选项覆盖 spec.strategy."""
@tool("t", strategy="thread")
def f() -> None:
pass
run_tool("t", ["--strategy", "sequential"])
_, kwargs = mock_run[0]
assert kwargs["strategy"] == "sequential"
def test_strategy_from_spec(self, mock_run: list[Any]) -> None:
"""无 --strategy 时用 spec.strategy."""
@tool("t", strategy="thread")
def f() -> None:
pass
run_tool("t", [])
_, kwargs = mock_run[0]
assert kwargs["strategy"] == "thread"
def test_strategy_default(self, mock_run: list[Any]) -> None:
"""无 --strategy 无 spec.strategy → 'dependency'."""
@tool("t")
def f() -> None:
pass
run_tool("t", [])
_, kwargs = mock_run[0]
assert kwargs["strategy"] == "dependency"
def test_list_flag(self, capsys: pytest.CaptureFixture[str]) -> None:
"""--list → 打印 rich 任务列表 + 退出 0."""
@tool("t", subcommand="a")
def a() -> None:
pass
ret = run_tool("t", ["--list"])
assert ret == ToolExitCode.SUCCESS
out = capsys.readouterr().out
assert "工具 t 的任务列表" in out
def test_list_flag_with_subcommand(self, capsys: pytest.CaptureFixture[str]) -> None:
"""--list 与子命令同时出现 → 仍列出任务."""
@tool("t", subcommand="a")
def a() -> None:
pass
ret = run_tool("t", ["a", "--list"])
assert ret == ToolExitCode.SUCCESS
out = capsys.readouterr().out
assert "工具 t 的任务列表" in out
def test_multi_no_subcommand_prints_help(self, capsys: pytest.CaptureFixture[str]) -> None:
"""多 subcommand 工具无 subcommand → typer 打印 help."""
@tool("t", subcommand="a")
def a() -> None:
pass
@tool("t", subcommand="b")
def b() -> None:
pass
ret = run_tool("t", [])
# typer no_args_is_help=True → SystemExit(0)
assert ret == ToolExitCode.SUCCESS
def test_task_failed_error(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
"""TaskFailedError(继承 PyFlowXError) → 退出码 1 + stderr."""
from pyflowx.errors import TaskFailedError
def _fail(graph: Any, **kw: Any) -> Any:
raise TaskFailedError("t", RuntimeError("boom"), 1)
monkeypatch.setattr(tools, "run", _fail)
@tool("t")
def f() -> None:
pass
ret = run_tool("t", [])
assert ret == ToolExitCode.FAILURE
err = capsys.readouterr().err
assert "boom" in err or "failed" in err
def test_pyflowx_error(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
"""PyFlowXError → 退出码 1 + stderr."""
from pyflowx.errors import PyFlowXError
def _fail(graph: Any, **kw: Any) -> Any:
raise PyFlowXError("graph error")
monkeypatch.setattr(tools, "run", _fail)
@tool("t")
def f() -> None:
pass
ret = run_tool("t", [])
assert ret == ToolExitCode.FAILURE
err = capsys.readouterr().err
assert "graph error" in err
def test_keyboard_interrupt(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
"""KeyboardInterrupt → 退出码 130."""
def _fail(graph: Any, **kw: Any) -> Any:
raise KeyboardInterrupt
monkeypatch.setattr(tools, "run", _fail)
@tool("t")
def f() -> None:
pass
ret = run_tool("t", [])
assert ret == ToolExitCode.INTERRUPTED
err = capsys.readouterr().err
assert "已中断" in err
def test_unknown_option_returns_failure(self) -> None:
"""未知选项 → ClickException → 退出码 1."""
@tool("t")
def f() -> None:
pass
ret = run_tool("t", ["--nonexistent-option"])
assert ret == ToolExitCode.FAILURE
def test_run_with_deps(self, mock_run: list[Any]) -> None:
"""有依赖时构建子图(多个 TaskSpec)."""
@tool("t", subcommand="a", needs=["b"])
def a() -> None:
pass
@tool("t", subcommand="b")
def b() -> None:
pass
ret = run_tool("t", ["a"])
assert ret == ToolExitCode.SUCCESS
graph, _ = mock_run[0]
# 图中应包含 a 和 b 两个任务
assert len(graph) == 2
assert set(graph.specs.keys()) == {"a", "b"}
def test_cmd_task_no_function_execution(self, mock_run: list[Any]) -> None:
"""cmd 任务:函数体不执行,TaskSpec 用 cmd."""
@tool("t", subcommand="c", cmd=["echo", "hi"])
def c() -> None:
raise RuntimeError("不应被执行")
ret = run_tool("t", ["c"])
assert ret == ToolExitCode.SUCCESS
# ---------------------------------------------------------------------- #
# 集成:工具描述 + 完整流程
# ---------------------------------------------------------------------- #
class TestIntegration:
"""集成测试:完整工具流程."""
def test_full_pymake_like_flow(self, mock_run: list[Any]) -> None:
"""模拟 pymake 流程:b + bc 聚合 ba + hidden 内部 job."""
@tool("pymake", subcommand="b", cmd=["uv", "build"], cwd=".")
def b(cwd: Path = Path()) -> None:
pass
@tool("pymake", subcommand="bc", cmd=["maturin", "build"], cwd=".")
def bc(cwd: Path = Path()) -> None:
pass
@tool("pymake", subcommand="ba", needs=["b", "bc"], strategy="thread")
def ba(cwd: Path = Path()) -> None:
pass
ret = run_tool("pymake", ["ba", "--cwd", "/tmp"])
assert ret == ToolExitCode.SUCCESS
def test_full_pdftool_like_flow(self, mock_run: list[Any]) -> None:
"""模拟 pdftool:m 合并 PDF."""
@tool("pdftool", subcommand="m", help="合并 PDF")
def merge(
input_paths: list[Path],
output_path: Path = Path("merged.pdf"),
password: str = "",
) -> None:
"""合并 PDF 文件."""
pass
ret = run_tool("pdftool", ["m", "a.pdf", "b.pdf", "--output-path", "out.pdf"])
assert ret == ToolExitCode.SUCCESS
def test_list_tools_after_multiple_registrations(self) -> None:
"""多次注册后 list_tools 排序正确."""
@tool("ztool")
def z() -> None:
pass
@tool("atool")
def a() -> None:
pass
@tool("mtool", subcommand="x")
def mx() -> None:
pass
@tool("mtool", subcommand="y")
def my() -> None:
pass
assert list_tools() == ["atool", "mtool", "ztool"]
assert list_subcommands("mtool") == ["x", "y"]
def test_clear_registry(self) -> None:
"""clear_tool_registry 清空注册表."""
@tool("t")
def f() -> None:
pass
assert "t" in list_tools()
clear_tool_registry()
assert list_tools() == []
Reference in New Issue View Git Blame Copy Permalink