zhou
179e5b3811
1. 将Strategy枚举改为Literal类型，移除normalize_strategy函数 2. 内联策略验证逻辑到run函数中 3. 使用dataclasses.field重构CliRunner的初始化方式 4. 修复测试用例中的函数名和调用方式不匹配问题 5. 调整部分测试用例的构造语法，适配新的API 6. 修正pymake模块中的函数重命名和条件变量命名问题 7. 为部分耗时测试添加@pytest.mark.slow标记
545 lines
17 KiB
Python
545 lines
17 KiB
Python
"""测试 TaskSpec 的命令和条件执行功能."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import pyflowx as px
|
|
from pyflowx.conditions import (
|
|
IS_LINUX,
|
|
IS_MACOS,
|
|
IS_WINDOWS,
|
|
BuiltinConditions,
|
|
)
|
|
|
|
# 跨平台的 echo 命令
|
|
if sys.platform == "win32":
|
|
ECHO_CMD = ["cmd", "/c", "echo"]
|
|
else:
|
|
ECHO_CMD = ["echo"]
|
|
|
|
|
|
def test_taskspec_with_cmd_list():
|
|
"""测试使用命令列表的 TaskSpec."""
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec("echo_test", cmd=[*ECHO_CMD, "hello"]),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert "echo_test" in report.results
|
|
assert report.results["echo_test"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
def test_taskspec_with_cmd_string():
|
|
"""测试使用 shell 命令字符串的 TaskSpec."""
|
|
if sys.platform == "win32":
|
|
shell_cmd = 'cmd /c "echo hello from shell"'
|
|
else:
|
|
shell_cmd = "echo 'hello from shell'"
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec("shell_test", cmd=shell_cmd),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert "shell_test" in report.results
|
|
assert report.results["shell_test"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
def test_taskspec_with_conditions_skip():
|
|
"""测试条件不满足时任务被跳过."""
|
|
|
|
# 创建一个永远不会满足的条件
|
|
def never_true():
|
|
return False
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"should_skip",
|
|
cmd=[*ECHO_CMD, "this should not run"],
|
|
conditions=(never_true,),
|
|
),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert "should_skip" in report.results
|
|
assert report.results["should_skip"].status == px.TaskStatus.SKIPPED
|
|
|
|
|
|
def test_taskspec_with_conditions_execute():
|
|
"""测试条件满足时任务正常执行."""
|
|
|
|
# 创建一个总是满足的条件
|
|
def always_true():
|
|
return True
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"should_run",
|
|
cmd=[*ECHO_CMD, "this should run"],
|
|
conditions=(always_true,),
|
|
),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert "should_run" in report.results
|
|
assert report.results["should_run"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
def test_platform_conditions():
|
|
"""测试平台条件."""
|
|
if sys.platform == "win32":
|
|
win_cmd = ["cmd", "/c", "echo", "Windows"]
|
|
posix_cmd = ["echo", "POSIX"]
|
|
else:
|
|
win_cmd = ["echo", "Windows"]
|
|
posix_cmd = ["echo", "POSIX"]
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"win_task",
|
|
cmd=win_cmd,
|
|
conditions=(IS_WINDOWS,),
|
|
),
|
|
px.TaskSpec(
|
|
"linux_task",
|
|
cmd=posix_cmd,
|
|
conditions=(IS_LINUX,),
|
|
),
|
|
px.TaskSpec(
|
|
"macos_task",
|
|
cmd=posix_cmd,
|
|
conditions=(IS_MACOS,),
|
|
),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
|
|
# 检查只有当前平台的任务执行了
|
|
if sys.platform == "win32":
|
|
assert report.results["win_task"].status == px.TaskStatus.SUCCESS
|
|
assert report.results["linux_task"].status == px.TaskStatus.SKIPPED
|
|
assert report.results["macos_task"].status == px.TaskStatus.SKIPPED
|
|
elif sys.platform == "linux":
|
|
assert report.results["win_task"].status == px.TaskStatus.SKIPPED
|
|
assert report.results["linux_task"].status == px.TaskStatus.SUCCESS
|
|
assert report.results["macos_task"].status == px.TaskStatus.SKIPPED
|
|
elif sys.platform == "darwin":
|
|
assert report.results["win_task"].status == px.TaskStatus.SKIPPED
|
|
assert report.results["linux_task"].status == px.TaskStatus.SKIPPED
|
|
assert report.results["macos_task"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
def test_app_installed_conditions():
|
|
"""测试应用安装条件."""
|
|
# 测试 python 应该总是安装的
|
|
if sys.platform == "win32":
|
|
python_cmd = ["python", "--version"]
|
|
else:
|
|
python_cmd = ["python3", "--version"]
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"python_check",
|
|
cmd=python_cmd,
|
|
conditions=(BuiltinConditions.HAS_APP_INSTALLED("python"),),
|
|
),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert "python_check" in report.results
|
|
# python 应该总是安装的
|
|
assert report.results["python_check"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
def test_combined_conditions():
|
|
"""测试组合条件."""
|
|
# AND 条件
|
|
and_condition = BuiltinConditions.AND(
|
|
lambda: True,
|
|
lambda: True,
|
|
)
|
|
|
|
# OR 条件
|
|
or_condition = BuiltinConditions.OR(
|
|
lambda: True,
|
|
lambda: False,
|
|
)
|
|
|
|
# NOT 条件
|
|
not_condition = BuiltinConditions.NOT(lambda: False)
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"and_test",
|
|
cmd=[*ECHO_CMD, "AND"],
|
|
conditions=(and_condition,),
|
|
),
|
|
px.TaskSpec(
|
|
"or_test",
|
|
cmd=[*ECHO_CMD, "OR"],
|
|
conditions=(or_condition,),
|
|
),
|
|
px.TaskSpec(
|
|
"not_test",
|
|
cmd=[*ECHO_CMD, "NOT"],
|
|
conditions=(not_condition,),
|
|
),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert report.results["and_test"].status == px.TaskStatus.SUCCESS
|
|
assert report.results["or_test"].status == px.TaskStatus.SUCCESS
|
|
assert report.results["not_test"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
def test_taskspec_with_cwd():
|
|
"""测试工作目录设置."""
|
|
if sys.platform == "win32":
|
|
ls_cmd = ["cmd", "/c", "dir"]
|
|
else:
|
|
ls_cmd = ["ls", "-la"]
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"list_current",
|
|
cmd=ls_cmd,
|
|
cwd=Path.cwd(),
|
|
),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert "list_current" in report.results
|
|
assert report.results["list_current"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
@pytest.mark.slow
|
|
def test_taskspec_with_timeout():
|
|
"""测试超时设置."""
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
# 短时间任务应该成功
|
|
px.TaskSpec(
|
|
"short_task",
|
|
cmd=["python", "-c", "import time; time.sleep(0.1)"],
|
|
timeout=1.0,
|
|
),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert "short_task" in report.results
|
|
assert report.results["short_task"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
def test_taskspec_dependency_with_conditions():
|
|
"""测试依赖和条件的组合."""
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"first",
|
|
cmd=[*ECHO_CMD, "first"],
|
|
conditions=(lambda: True,),
|
|
),
|
|
px.TaskSpec(
|
|
"second",
|
|
cmd=[*ECHO_CMD, "second"],
|
|
depends_on=("first",),
|
|
conditions=(lambda: True,),
|
|
),
|
|
px.TaskSpec(
|
|
"third",
|
|
cmd=[*ECHO_CMD, "third"],
|
|
depends_on=("second",),
|
|
),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert report.results["first"].status == px.TaskStatus.SUCCESS
|
|
assert report.results["second"].status == px.TaskStatus.SUCCESS
|
|
assert report.results["third"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
def test_taskspec_mixed_fn_and_cmd():
|
|
"""测试混合使用 fn 和 cmd."""
|
|
|
|
def my_function():
|
|
return "result from function"
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec("fn_task", fn=my_function),
|
|
px.TaskSpec("cmd_task", cmd=[*ECHO_CMD, "from command"]),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert report.results["fn_task"].status == px.TaskStatus.SUCCESS
|
|
assert report.results["fn_task"].value == "result from function"
|
|
assert report.results["cmd_task"].status == px.TaskStatus.SUCCESS
|
|
|
|
|
|
def test_taskspec_cmd_overrides_fn():
|
|
"""测试 cmd 参数优先于 fn 参数."""
|
|
|
|
def my_function():
|
|
return "should not run"
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"cmd_priority",
|
|
fn=my_function,
|
|
cmd=[*ECHO_CMD, "cmd takes priority"],
|
|
),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert report.results["cmd_priority"].status == px.TaskStatus.SUCCESS
|
|
# cmd 应该被执行，而不是 fn
|
|
assert report.results["cmd_priority"].value is None
|
|
|
|
|
|
def test_taskspec_callable_cmd():
|
|
"""测试 cmd 参数使用可调用对象."""
|
|
|
|
def my_callable():
|
|
return "callable result"
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec("callable_cmd", cmd=my_callable),
|
|
]
|
|
)
|
|
|
|
report = px.run(graph, strategy="sequential")
|
|
assert report.success
|
|
assert report.results["callable_cmd"].status == px.TaskStatus.SUCCESS
|
|
assert report.results["callable_cmd"].value == "callable result"
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# verbose 模式测试
|
|
# ---------------------------------------------------------------------- #
|
|
class TestTaskSpecVerbose:
|
|
"""测试 TaskSpec 的 verbose 字段."""
|
|
|
|
def test_verbose_default_is_false(self) -> None:
|
|
"""verbose 默认应为 False."""
|
|
spec: px.TaskSpec[object] = px.TaskSpec("a", cmd=[*ECHO_CMD, "hi"])
|
|
assert spec.verbose is False
|
|
|
|
def test_verbose_true_prints_command(
|
|
self, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""verbose=True 时应打印执行的命令."""
|
|
graph = px.Graph.from_specs(
|
|
[px.TaskSpec("echo", cmd=[*ECHO_CMD, "verbose-output"], verbose=True)]
|
|
)
|
|
px.run(graph, strategy="sequential")
|
|
captured = capsys.readouterr()
|
|
assert "执行命令" in captured.out
|
|
assert "返回码" in captured.out
|
|
|
|
def test_verbose_false_silent(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""verbose=False 时不应打印命令信息."""
|
|
graph = px.Graph.from_specs(
|
|
[px.TaskSpec("echo", cmd=[*ECHO_CMD, "silent"], verbose=False)]
|
|
)
|
|
px.run(graph, strategy="sequential")
|
|
captured = capsys.readouterr()
|
|
assert "执行命令" not in captured.out
|
|
assert "返回码" not in captured.out
|
|
|
|
def test_verbose_true_shell_cmd(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""verbose=True 时 shell 命令也应打印执行信息."""
|
|
if sys.platform == "win32":
|
|
shell_cmd = 'cmd /c "echo shell-verbose"'
|
|
else:
|
|
shell_cmd = "echo 'shell-verbose'"
|
|
|
|
graph = px.Graph.from_specs([px.TaskSpec("shell", cmd=shell_cmd, verbose=True)])
|
|
px.run(graph, strategy="sequential")
|
|
captured = capsys.readouterr()
|
|
assert "执行 Shell" in captured.out
|
|
|
|
def test_verbose_prints_cwd(self, capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""verbose=True 且设置了 cwd 时应打印工作目录."""
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
graph = px.Graph.from_specs(
|
|
[px.TaskSpec("ls", cmd=ECHO_CMD, cwd=Path(tmpdir), verbose=True)]
|
|
)
|
|
px.run(graph, strategy="sequential")
|
|
captured = capsys.readouterr()
|
|
assert "工作目录" in captured.out
|
|
|
|
def test_verbose_failure_includes_returncode(
|
|
self, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
"""verbose=True 时失败也应打印返回码."""
|
|
from pyflowx.errors import TaskFailedError
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"fail",
|
|
cmd=["python", "-c", "import sys; sys.exit(1)"],
|
|
verbose=True,
|
|
)
|
|
]
|
|
)
|
|
with pytest.raises(TaskFailedError):
|
|
px.run(graph, strategy="sequential")
|
|
captured = capsys.readouterr()
|
|
assert "返回码" in captured.out
|
|
|
|
|
|
# ---------------------------------------------------------------------- #
|
|
# _wrap_cmd 错误路径测试
|
|
# ---------------------------------------------------------------------- #
|
|
class TestTaskSpecCmdErrors:
|
|
"""测试 _wrap_cmd 的错误处理路径."""
|
|
|
|
def test_cmd_list_file_not_found(self) -> None:
|
|
"""命令不存在时应抛出 RuntimeError."""
|
|
from pyflowx.errors import TaskFailedError
|
|
|
|
graph = px.Graph.from_specs(
|
|
[px.TaskSpec("missing", cmd=["this-command-does-not-exist-xyz"])]
|
|
)
|
|
with pytest.raises(TaskFailedError) as exc_info:
|
|
px.run(graph, strategy="sequential")
|
|
# 错误信息应包含命令未找到
|
|
assert (
|
|
"命令未找到" in str(exc_info.value.cause)
|
|
or "not found" in str(exc_info.value.cause).lower()
|
|
)
|
|
|
|
def test_cmd_list_failure_includes_stderr(self) -> None:
|
|
"""命令失败时错误信息应包含 stderr."""
|
|
from pyflowx.errors import TaskFailedError
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"fail",
|
|
cmd=[
|
|
"python",
|
|
"-c",
|
|
"import sys; sys.stderr.write('error-msg'); sys.exit(1)",
|
|
],
|
|
)
|
|
]
|
|
)
|
|
with pytest.raises(TaskFailedError) as exc_info:
|
|
px.run(graph, strategy="sequential")
|
|
# 非 verbose 模式下, stderr 应包含在错误信息中
|
|
assert "error-msg" in str(exc_info.value.cause)
|
|
|
|
def test_cmd_string_file_not_found(self) -> None:
|
|
"""shell 命令不存在时应抛出 RuntimeError."""
|
|
from pyflowx.errors import TaskFailedError
|
|
|
|
graph = px.Graph.from_specs(
|
|
[px.TaskSpec("missing", cmd="this-command-does-not-exist-xyz-123")]
|
|
)
|
|
with pytest.raises(TaskFailedError):
|
|
px.run(graph, strategy="sequential")
|
|
|
|
def test_cmd_string_failure(self) -> None:
|
|
"""shell 命令失败时应抛出 RuntimeError."""
|
|
from pyflowx.errors import TaskFailedError
|
|
|
|
graph = px.Graph.from_specs(
|
|
[px.TaskSpec("fail", cmd='python -c "import sys; sys.exit(1)"')]
|
|
)
|
|
with pytest.raises(TaskFailedError) as exc_info:
|
|
_ = px.run(graph, strategy="sequential")
|
|
assert "Shell 命令执行失败" in str(exc_info.value.cause)
|
|
|
|
@pytest.mark.slow
|
|
def test_cmd_timeout_raises(self) -> None:
|
|
"""命令超时应抛出 RuntimeError."""
|
|
from pyflowx.errors import TaskFailedError
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"slow",
|
|
cmd=["python", "-c", "import time; time.sleep(5)"],
|
|
timeout=0.1,
|
|
)
|
|
]
|
|
)
|
|
with pytest.raises(TaskFailedError) as exc_info:
|
|
_ = px.run(graph, strategy="sequential")
|
|
assert "超时" in str(exc_info.value.cause)
|
|
|
|
@pytest.mark.slow
|
|
def test_cmd_string_timeout_raises(self) -> None:
|
|
"""shell 命令超时应抛出 RuntimeError."""
|
|
from pyflowx.errors import TaskFailedError
|
|
|
|
graph = px.Graph.from_specs(
|
|
[
|
|
px.TaskSpec(
|
|
"slow", cmd='python -c "import time; time.sleep(5)"', timeout=0.1
|
|
)
|
|
]
|
|
)
|
|
with pytest.raises(TaskFailedError) as exc_info:
|
|
_ = px.run(graph, strategy="sequential")
|
|
assert "超时" in str(exc_info.value.cause)
|
|
|
|
def test_unsupported_cmd_type_raises(self) -> None:
|
|
"""不支持的 cmd 类型应在执行时抛出 TypeError."""
|
|
from pyflowx.errors import TaskFailedError
|
|
|
|
graph = px.Graph.from_specs(
|
|
[px.TaskSpec("bad", cmd=123)] # type: ignore[arg-type]
|
|
)
|
|
with pytest.raises((TypeError, TaskFailedError)):
|
|
_ = px.run(graph, strategy="sequential")
|
|
|
|
def test_no_fn_no_cmd_raises(self) -> None:
|
|
"""没有 fn 和 cmd 时应抛出 ValueError."""
|
|
with pytest.raises(ValueError, match="必须提供 fn 或 cmd"):
|
|
px.TaskSpec("empty")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|