zhou
6da42ec5ff
1. 新增tools.py模块实现@px.tool装饰器与工具注册表 2. 将所有configs下的YAML工具迁移为ops/xxx.py模块形式 3. 重构cli路由逻辑，优先加载Python工具实现回退YAML 4. 删除所有YAML配置文件与旧的yaml_loader相关代码 5. 调整__init__.py导出API，移除YAML相关依赖
1311 lines
37 KiB
Python
1311 lines
37 KiB
Python
"""tests/test_tools.py - @px.tool 装饰器核心模块测试.
|
|
|
|
覆盖 tools.py 全部分支:注册/重复检测/CLI 生成各类型/单命令执行/
|
|
cmd/fn/聚合/DAG/hidden/全局选项/退出码三态.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
from typing import Any, Literal, Optional, Union
|
|
|
|
import pytest
|
|
|
|
from pyflowx import tools
|
|
from pyflowx.executors import RunReport
|
|
from pyflowx.task import RetryPolicy
|
|
from pyflowx.tools import (
|
|
ToolExitCode,
|
|
_add_func_args_to_parser,
|
|
_add_global_options,
|
|
_build_parser,
|
|
_build_task_spec,
|
|
_collect_with_deps,
|
|
_extract_variables,
|
|
_handle_list,
|
|
_has_function_logic,
|
|
_is_aggregate,
|
|
_map_param_type,
|
|
_noop,
|
|
_snake_to_kebab,
|
|
_unwrap_optional,
|
|
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"]
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _snake_to_kebab
|
|
# ---------------------------------------------------------------------- #
|
|
class TestSnakeToKebab:
|
|
"""测试 snake_case → kebab-case 转换."""
|
|
|
|
def test_simple(self) -> None:
|
|
assert _snake_to_kebab("output") == "output"
|
|
|
|
def test_snake(self) -> None:
|
|
assert _snake_to_kebab("output_dir") == "output-dir"
|
|
|
|
def test_multi_word(self) -> None:
|
|
assert _snake_to_kebab("python_mirror") == "python-mirror"
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _unwrap_optional
|
|
# ---------------------------------------------------------------------- #
|
|
class TestUnwrapOptional:
|
|
"""测试 Optional 解包."""
|
|
|
|
def test_non_optional(self) -> None:
|
|
inner, is_opt = _unwrap_optional(int)
|
|
assert inner is int
|
|
assert is_opt is False
|
|
|
|
def test_optional(self) -> None:
|
|
inner, is_opt = _unwrap_optional(Optional[int])
|
|
assert inner is int
|
|
assert is_opt is True
|
|
|
|
def test_union_none(self) -> None:
|
|
inner, is_opt = _unwrap_optional(Union[int, None])
|
|
assert inner is int
|
|
assert is_opt is True
|
|
|
|
def test_str(self) -> None:
|
|
inner, is_opt = _unwrap_optional(str)
|
|
assert inner is str
|
|
assert is_opt is False
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _map_param_type
|
|
# ---------------------------------------------------------------------- #
|
|
class TestMapParamType:
|
|
"""测试类型注解 → argparse kwargs 映射."""
|
|
|
|
def test_int(self) -> None:
|
|
assert _map_param_type(int) == {"type": int}
|
|
|
|
def test_float(self) -> None:
|
|
assert _map_param_type(float) == {"type": float}
|
|
|
|
def test_str(self) -> None:
|
|
assert _map_param_type(str) == {"type": str}
|
|
|
|
def test_path(self) -> None:
|
|
assert _map_param_type(Path) == {"type": Path}
|
|
|
|
def test_list_of_path(self) -> None:
|
|
kwargs = _map_param_type(list[Path])
|
|
assert kwargs["type"] is Path
|
|
assert kwargs["nargs"] == "+"
|
|
|
|
def test_list_no_type_arg(self) -> None:
|
|
"""list 无类型参数 → nargs='+', 无 type."""
|
|
kwargs = _map_param_type(list)
|
|
# list 的 origin 是 list,但 get_args(list) 为空
|
|
assert kwargs.get("nargs") == "+" or kwargs == {}
|
|
|
|
def test_literal(self) -> None:
|
|
kwargs = _map_param_type(Literal["a", "b"]) # type: ignore[valid-type]
|
|
assert kwargs["choices"] == ["a", "b"]
|
|
|
|
def test_unknown_type(self) -> None:
|
|
"""未知类型返回空 dict."""
|
|
assert _map_param_type(complex) == {}
|
|
|
|
def test_optional_int(self) -> None:
|
|
"""Optional[int] 解包后映射 int."""
|
|
kwargs = _map_param_type(Optional[int])
|
|
assert kwargs == {"type": int}
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _add_func_args_to_parser
|
|
# ---------------------------------------------------------------------- #
|
|
class TestAddFuncArgsToParser:
|
|
"""测试函数签名 → argparse 参数."""
|
|
|
|
def _parse(self, func: Any, argv: list[str]) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser()
|
|
_add_func_args_to_parser(parser, func)
|
|
return parser.parse_args(argv)
|
|
|
|
def test_positional(self) -> None:
|
|
def f(name: str) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["hello"])
|
|
assert ns.name == "hello"
|
|
|
|
def test_option_with_default(self) -> None:
|
|
def f(output: str = "out") -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--output", "x"])
|
|
assert ns.output == "x"
|
|
|
|
def test_option_default_value(self) -> None:
|
|
def f(output: str = "out") -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, [])
|
|
assert ns.output == "out"
|
|
|
|
def test_int_type(self) -> None:
|
|
def f(level: int = 5) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--level", "10"])
|
|
assert ns.level == 10
|
|
|
|
def test_float_type(self) -> None:
|
|
def f(rate: float = 1.0) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--rate", "2.5"])
|
|
assert ns.rate == 2.5
|
|
|
|
def test_path_type(self) -> None:
|
|
def f(p: Path = Path()) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--p", "/tmp"])
|
|
assert ns.p == Path("/tmp")
|
|
|
|
def test_list_positional_nargs_plus(self) -> None:
|
|
def f(inputs: list[Path]) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["a.pdf", "b.pdf"])
|
|
assert ns.inputs == [Path("a.pdf"), Path("b.pdf")]
|
|
|
|
def test_literal_choices(self) -> None:
|
|
def f(mode: Literal["a", "b"] = "a") -> None: # type: ignore[valid-type]
|
|
pass
|
|
|
|
ns = self._parse(f, ["--mode", "b"])
|
|
assert ns.mode == "b"
|
|
|
|
def test_bool_default_false_store_true(self) -> None:
|
|
def f(dry_run: bool = False) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--dry-run"])
|
|
assert ns.dry_run is True
|
|
|
|
def test_bool_default_false_absent(self) -> None:
|
|
def f(dry_run: bool = False) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, [])
|
|
assert ns.dry_run is False
|
|
|
|
def test_bool_default_true_store_false(self) -> None:
|
|
def f(verbose: bool = True) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--no-verbose"])
|
|
assert ns.verbose is False
|
|
|
|
def test_bool_default_true_absent(self) -> None:
|
|
def f(verbose: bool = True) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, [])
|
|
assert ns.verbose is True
|
|
|
|
def test_bool_no_default_store_true(self) -> None:
|
|
"""bool 无默认值 → store_true, 默认 False."""
|
|
|
|
def f(flag: bool) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--flag"])
|
|
assert ns.flag is True
|
|
|
|
def test_skip_self(self) -> None:
|
|
"""self 参数被跳过."""
|
|
|
|
class C:
|
|
def m(self, x: int = 1) -> None:
|
|
pass
|
|
|
|
ns = self._parse(C().m, ["--x", "5"])
|
|
assert ns.x == 5
|
|
|
|
def test_skip_var_positional(self) -> None:
|
|
"""*args 被跳过."""
|
|
|
|
def f(*args: Any, x: int = 1) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--x", "5"])
|
|
assert ns.x == 5
|
|
|
|
def test_skip_var_keyword(self) -> None:
|
|
"""**kwargs 被跳过."""
|
|
|
|
def f(x: int = 1, **kw: Any) -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--x", "5"])
|
|
assert ns.x == 5
|
|
|
|
def test_kebab_case_option(self) -> None:
|
|
"""snake_case 参数名 → kebab-case 选项."""
|
|
|
|
def f(output_dir: str = "out") -> None:
|
|
pass
|
|
|
|
ns = self._parse(f, ["--output-dir", "x"])
|
|
assert ns.output_dir == "x"
|
|
|
|
def test_no_annotation_defaults_to_str(self) -> None:
|
|
"""无类型注解 → str."""
|
|
|
|
def f(x="default") -> None: # type: ignore[no-untyped-def]
|
|
pass
|
|
|
|
ns = self._parse(f, ["--x", "val"])
|
|
assert ns.x == "val"
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _add_global_options
|
|
# ---------------------------------------------------------------------- #
|
|
class TestAddGlobalOptions:
|
|
"""测试全局选项注入."""
|
|
|
|
def test_global_options_on_simple_parser(self) -> None:
|
|
parser = argparse.ArgumentParser()
|
|
_add_global_options(parser)
|
|
ns = parser.parse_args(["--dry-run", "--quiet", "--strategy", "thread", "--list"])
|
|
assert ns.dry_run is True
|
|
assert ns.verbose is False
|
|
assert ns.strategy == "thread"
|
|
assert ns.list_jobs is True
|
|
|
|
def test_global_options_on_subparsers(self) -> None:
|
|
"""subparser 也获得全局选项."""
|
|
|
|
def f(x: int = 1) -> None:
|
|
pass
|
|
|
|
parser = argparse.ArgumentParser()
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
sub = subparsers.add_parser("cmd")
|
|
_add_func_args_to_parser(sub, f)
|
|
_add_global_options(parser)
|
|
|
|
ns = parser.parse_args(["cmd", "--x", "5", "--dry-run", "--quiet"])
|
|
assert ns.x == 5
|
|
assert ns.dry_run is True
|
|
assert ns.verbose is False
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _build_parser
|
|
# ---------------------------------------------------------------------- #
|
|
class TestBuildParser:
|
|
"""测试 parser 构建."""
|
|
|
|
def test_single_command_parser(self) -> None:
|
|
"""单命令工具(仅 subcommand=None)→ 简单 parser."""
|
|
|
|
@tool("solo")
|
|
def f(x: int = 1) -> None:
|
|
pass
|
|
|
|
from pyflowx.tools import _TOOL_REGISTRY
|
|
|
|
parser = _build_parser("solo", _TOOL_REGISTRY["solo"])
|
|
ns = parser.parse_args(["--x", "5"])
|
|
assert ns.x == 5
|
|
|
|
def test_multi_subcommand_parser(self) -> None:
|
|
"""多 subcommand → 带 subparsers."""
|
|
|
|
@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
|
|
|
|
parser = _build_parser("t", _TOOL_REGISTRY["t"])
|
|
ns = parser.parse_args(["a", "--x", "5"])
|
|
assert ns.command == "a"
|
|
assert ns.x == 5
|
|
|
|
def test_parser_description_from_description(self) -> None:
|
|
"""description 优先于 help."""
|
|
|
|
@tool("t", description="工具描述")
|
|
def f() -> None:
|
|
pass
|
|
|
|
from pyflowx.tools import _TOOL_REGISTRY
|
|
|
|
parser = _build_parser("t", _TOOL_REGISTRY["t"])
|
|
assert parser.description == "工具描述"
|
|
|
|
def test_parser_description_from_help(self) -> None:
|
|
"""无 description 时用 help."""
|
|
|
|
@tool("t", help="帮助文本")
|
|
def f() -> None:
|
|
pass
|
|
|
|
from pyflowx.tools import _TOOL_REGISTRY
|
|
|
|
parser = _build_parser("t", _TOOL_REGISTRY["t"])
|
|
assert parser.description == "帮助文本"
|
|
|
|
def test_parser_description_from_name(self) -> None:
|
|
"""无 description/help 时用工具名."""
|
|
|
|
@tool("t")
|
|
def f() -> None:
|
|
pass
|
|
|
|
from pyflowx.tools import _TOOL_REGISTRY
|
|
|
|
parser = _build_parser("t", _TOOL_REGISTRY["t"])
|
|
assert parser.description == "t"
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _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"
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _extract_variables
|
|
# ---------------------------------------------------------------------- #
|
|
class TestExtractVariables:
|
|
"""测试 argparse namespace → 函数参数提取."""
|
|
|
|
def test_extract_matching_params(self) -> None:
|
|
def f(x: int = 1, y: str = "z") -> None:
|
|
pass
|
|
|
|
ns = argparse.Namespace(x=5, y="a", extra="ignored")
|
|
variables = _extract_variables(ns, f)
|
|
assert variables == {"x": 5, "y": "a"}
|
|
|
|
def test_extract_missing_attr(self) -> None:
|
|
"""namespace 缺少属性 → 不包含在结果中."""
|
|
|
|
def f(x: int = 1, y: str = "z") -> None:
|
|
pass
|
|
|
|
ns = argparse.Namespace(x=5)
|
|
variables = _extract_variables(ns, f)
|
|
assert variables == {"x": 5}
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _handle_list
|
|
# ---------------------------------------------------------------------- #
|
|
class TestHandleList:
|
|
"""测试 --list 输出."""
|
|
|
|
def test_list_output(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""--list 打印任务列表."""
|
|
|
|
@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
|
|
|
|
ret = _handle_list("t", _TOOL_REGISTRY["t"])
|
|
assert ret == 0
|
|
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_no_deps(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""无依赖显示 (无依赖)."""
|
|
|
|
@tool("t", subcommand="a")
|
|
def a() -> None:
|
|
pass
|
|
|
|
from pyflowx.tools import _TOOL_REGISTRY
|
|
|
|
_handle_list("t", _TOOL_REGISTRY["t"])
|
|
out = capsys.readouterr().out
|
|
assert "(无依赖)" 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], capsys: pytest.CaptureFixture[str]) -> 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 透传到 run."""
|
|
|
|
# 注意: run_tool 用 target_spec.strategy or "dependency"
|
|
# 单命令无 strategy → "dependency"
|
|
@tool("t", strategy="thread")
|
|
def f() -> None:
|
|
pass
|
|
|
|
run_tool("t", [])
|
|
_, kwargs = mock_run[0]
|
|
# strategy 来自 target_spec.strategy
|
|
# run_tool 调用 run(graph, strategy=target_spec.strategy or "dependency", ...)
|
|
# mock_run 捕获 kwargs
|
|
assert kwargs["strategy"] == "thread" or "strategy" in kwargs
|
|
|
|
def test_list_flag(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""--list → 打印任务列表 + 退出 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_multi_no_subcommand_prints_help(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""多 subcommand 工具无 subcommand → 打印 help + 0."""
|
|
|
|
@tool("t", subcommand="a")
|
|
def a() -> None:
|
|
pass
|
|
|
|
@tool("t", subcommand="b")
|
|
def b() -> None:
|
|
pass
|
|
|
|
ret = run_tool("t", [])
|
|
assert ret == ToolExitCode.SUCCESS
|
|
out = capsys.readouterr().out
|
|
# 应打印 help( argparse 的 print_help 输出包含 usage)
|
|
assert "usage" in out.lower() or "pf t" in out
|
|
|
|
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_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
|
|
|
|
# 验证图中包含 3 个任务
|
|
_graph, _ = mock_run[0]
|
|
# cwd 应从 CLI 透传(但 _build_task_spec 对 cmd 任务从 variables['cwd'] 取)
|
|
# variables 包含 cwd=/tmp
|
|
# b 和 bc 的 cwd 应为 /tmp
|
|
# ba 是聚合,无 cwd
|
|
|
|
def test_full_pdftool_like_flow(self, mock_run: list[Any]) -> None:
|
|
"""模拟 pdftool:m 合并 PDF."""
|
|
|
|
@tool("pdftool", subcommand="m", help="合并 PDF")
|
|
def merge(
|
|
inputs: list[Path],
|
|
output: Path = Path("merged.pdf"),
|
|
password: str = "",
|
|
) -> None:
|
|
"""合并 PDF 文件."""
|
|
pass
|
|
|
|
ret = run_tool("pdftool", ["m", "a.pdf", "b.pdf", "--output", "out.pdf"])
|
|
assert ret == ToolExitCode.SUCCESS
|
|
|
|
# 验证 kwargs 透传
|
|
_graph, _ = mock_run[0]
|
|
# fn 任务的 kwargs 应包含 inputs/output/password
|
|
|
|
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() == []
|