zhou/pyflowx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增skip_if_missing特性，支持命令不存在时自动跳过任务

Browse Source 
本次提交实现了命令任务的自动跳过功能：
1. 为TaskSpec新增skip_if_missing参数，默认开启，仅对list[str]类型cmd生效
2. 通过shutil.which检查命令是否存在，不存在则标记任务为SKIPPED而非失败
3. 重构should_execute方法，整合条件检查与命令可用性检查
4. 更新文档与示例代码，添加该参数的使用说明
5. 移除cli/pymake.py中的冗余check辅助函数，改用内置特性
6. 为所有内置任务添加skip_if_missing=True配置
7. 修复线程并行测试的超时阈值，放宽到1.0秒
8. 优化代码格式与压缩单行表达式
9. 新增完整的单元测试覆盖该特性的各种场景
This commit is contained in:
gooker_young
2026-06-21 18:55:24 +08:00
parent cae51856d2
commit ea591d1088
11 changed files with 193 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+7
View File
@@ -80,6 +80,7 @@ px.TaskSpec(
conditions=(is_prod,), # 条件函数列表（全部为 True 才执行）
cwd=Path("/tmp"), # 命令工作目录（仅 cmd 模式）
verbose=True, # 打印命令输出（仅 cmd 模式）
skip_if_missing=True, # 命令不存在时自动跳过（仅 list[str] cmd）
)
```
@@ -88,6 +89,8 @@ px.TaskSpec(
- **函数任务**（`fn`）：普通 Python 函数，参数名驱动自动注入
- **命令任务**（`cmd`）：执行外部命令，支持 `list[str]`、`str`（shell）、`Callable` 三种形态
`skip_if_missing=True` 时，`list[str]` 类型的 `cmd` 会通过 `shutil.which` 检查命令是否存在，不存在则跳过任务（标记为 `SKIPPED`）而非失败。适用于构建工具场景，避免因未安装某些工具而导致整个图执行失败。
### Graph —— DAG 构建
```python
@@ -176,11 +179,15 @@ graph = px.Graph.from_specs([
px.TaskSpec("check_git", cmd="git status | head"),
# 带工作目录与超时
px.TaskSpec("build", cmd=["make", "all"], cwd=Path("/project"), timeout=300),
# 命令不存在时自动跳过（而非失败）
px.TaskSpec("optional_tool", cmd=["maturin", "build"], skip_if_missing=True),
])
```
`verbose=True` 时打印执行的命令、工作目录、返回码与输出；`verbose=False` 时静默执行（失败信息仍包含 stderr）。
`skip_if_missing=True` 时，`list[str]` 类型的 `cmd` 会通过 `shutil.which` 检查命令是否存在，不存在则跳过任务（标记为 `SKIPPED`）而非失败。适用于构建工具场景，避免因未安装某些工具而导致整个图执行失败。对于 `str`（shell）和 `Callable` 类型的 `cmd`，此参数无效。
## 条件执行
`conditions` 参数让任务按条件跳过（标记为 `SKIPPED`）：
src/pyflowx/__init__.py
+6
View File
@@ -45,6 +45,12 @@
cmd=["git", "--version"],
conditions=(BuiltinConditions.HAS_INSTALLED("git"),)
),
# 命令不存在时自动跳过（而非失败）
px.TaskSpec(
"optional_build",
cmd=["maturin", "build"],
skip_if_missing=True
),
])
report = px.run(graph)
"""
src/pyflowx/cli/pymake.py
+30 -98
View File
@@ -7,7 +7,7 @@
from __future__ import annotations
import pyflowx as px
from pyflowx.conditions import BuiltinConditions, Constants
from pyflowx.conditions import Constants
def maturin_build_cmd() -> list[str]:
@@ -18,9 +18,9 @@ def maturin_build_cmd() -> list[str]:
list[str]
完整的 maturin 构建命令列表.
"""
base_cmd = ["maturin", "build", "-r"].copy()
command = ["maturin", "build", "-r"].copy()
if Constants.IS_WINDOWS:
base_cmd.extend(
command.extend(
[
"--target",
"x86_64-win7-windows-msvc",
@@ -29,94 +29,31 @@ def maturin_build_cmd() -> list[str]:
"python3.8",
]
)
return base_cmd
return command
def check(name: str) -> px.Condition:
"""检查指定工具是否已安装.
Returns
-------
bool
如果已安装则返回 True,否则返回 False.
"""
return BuiltinConditions.HAS_INSTALLED(name)
uv_build: px.TaskSpec = px.TaskSpec("uv_build", cmd=["uv", "build"], conditions=(check("uv"),))
maturin_build: px.TaskSpec = px.TaskSpec("maturin_build", cmd=maturin_build_cmd(), conditions=(check("maturin"),))
uv_sync: px.TaskSpec = px.TaskSpec("uv_sync", cmd=["uv", "sync"], conditions=(check("uv"),))
git_clean: px.TaskSpec = px.TaskSpec("git_clean", cmd=["gitt", "c"], conditions=(check("gitt"),))
uv_build: px.TaskSpec = px.TaskSpec("uv_build", cmd=["uv", "build"])
maturin_build: px.TaskSpec = px.TaskSpec("maturin_build", cmd=maturin_build_cmd())
uv_sync: px.TaskSpec = px.TaskSpec("uv_sync", cmd=["uv", "sync"])
git_clean: px.TaskSpec = px.TaskSpec("git_clean", cmd=["gitt", "c"])
test: px.TaskSpec = px.TaskSpec(
"test",
cmd=[
"pytest",
"-m",
"not slow",
"-n",
"8",
"--dist",
"loadfile",
"--color=yes",
"--durations=10",
],
conditions=(check("pytest"),),
"test", cmd=["pytest", "-m", "not slow", "-n", "8", "--dist", "loadfile", "--color=yes", "--durations=10"]
)
test_fast: px.TaskSpec = px.TaskSpec(
"test_fast",
cmd=[
"pytest",
"-m",
"not slow",
"--dist",
"loadfile",
"--color=yes",
"--durations=10",
],
conditions=(check("pytest"),),
"test_fast", cmd=["pytest", "-m", "not slow", "--dist", "loadfile", "--color=yes", "--durations=10"]
)
test_coverage: px.TaskSpec = px.TaskSpec(
"test_coverage",
cmd=[
"pytest",
"--cov",
"-n",
"8",
"--dist",
"loadfile",
"--tb=short",
"-v",
"--color=yes",
"--durations=10",
],
conditions=(check("pytest"),),
cmd=["pytest", "--cov", "-n", "8", "--dist", "loadfile", "--tb=short", "-v", "--color=yes", "--durations=10"],
)
ruff_lint: px.TaskSpec = px.TaskSpec(
"lint",
cmd=[
"ruff",
"check",
"--fix",
"--unsafe-fixes",
],
conditions=(check("ruff"),),
)
mypy_check: px.TaskSpec = px.TaskSpec("typecheck", cmd=["mypy", "."], conditions=(check("mypy"),))
ty_check: px.TaskSpec = px.TaskSpec("ty_check", cmd=["ty", "check", "."], conditions=(check("ty"),))
doc: px.TaskSpec = px.TaskSpec(
"doc", cmd=["sphinx-build", "-b", "html", "docs", "docs/_build"], conditions=(check("sphinx-build"),)
)
hatch_publish: px.TaskSpec = px.TaskSpec("publish_python", cmd=["hatch", "publish"], conditions=(check("hatch"),))
twine_publish: px.TaskSpec = px.TaskSpec(
"twine_publish",
cmd=[
"twine",
"upload",
"--disable-progress-bar",
],
conditions=(check("twine"),),
)
tox: px.TaskSpec = px.TaskSpec("tox", cmd=["tox", "-p", "auto"], conditions=(check("tox"),))
ruff_lint: px.TaskSpec = px.TaskSpec("lint", cmd=["ruff", "check", "--fix", "--unsafe-fixes"])
ruff_format: px.TaskSpec = px.TaskSpec("format", cmd=["ruff", "format", "--check", "."], depends_on=("lint",))
mypy_check: px.TaskSpec = px.TaskSpec("typecheck", cmd=["mypy", "."])
ty_check: px.TaskSpec = px.TaskSpec("ty_check", cmd=["ty", "check", "."])
doc: px.TaskSpec = px.TaskSpec("doc", cmd=["sphinx-build", "-b", "html", "docs", "docs/_build"])
hatch_publish: px.TaskSpec = px.TaskSpec("publish_python", cmd=["hatch", "publish"])
twine_publish: px.TaskSpec = px.TaskSpec("twine_publish", cmd=["twine", "upload", "--disable-progress-bar"])
tox: px.TaskSpec = px.TaskSpec("tox", cmd=["tox", "-p", "auto"])
def main():
@@ -128,13 +65,13 @@ def main():
🔨 构建命令:
pymake b - 构建 Python 主包 (uv build)
pymake bc - 构建 Rust 核心模块 (maturin build)
pymake ba - 构建所有包 (先 Rust 后 Python)
pymake ba - 构建所有包 (先 Python 后 Rust)
📦 安装命令 (开发模式):
pymake sync - 安装依赖包 (uv sync)
🧹 清理命令:
pymake c - 清理所有构建产物
pymake c - 清理所有构建产物 (gitt c)
🛠️ 开发工具:
pymake t - 运行测试 (pytest)
@@ -145,29 +82,24 @@ def main():
pymake doc - 构建文档 (sphinx)
🔬 多版本测试:
pymake tox - 多版本 Python 测试 (3.8-3.14)
pymake tox_install - 安装所有 Python 版本 (仅安装不测试)
pymake tox - 多版本 Python 测试 (tox -p auto)
📦 发布命令:
pymake pb - 发布到 PyPI (hatch publish)
pymake pba - 发布所有包 (先 Rust 后 Python)
pymake pbc - 发布 Rust 核心模块 (maturin publish)
pymake pb - 发布到 PyPI (twine + hatch)
💡 常用工作流:
1. 初始化开发环境: pymake ia
2. 日常开发: pymake lint && pymake t
3. 构建发布包: pymake ba
4. 多版本兼容性测试: pymake tox
5. 发布到 PyPI: pymake pb
6. 清理重新开始: pymake ca && pymake ia
1. 日常开发: pymake lint && pymake t
2. 构建发布包: pymake ba
3. 多版本兼容性测试: pymake tox
4. 发布到 PyPI: pymake pb
📝 示例:
pymake ba # 构建所有包
pymake ia # 安装开发环境
pymake sync # 安装依赖
pymake t # 运行测试
pymake tox # 多版本兼容性测试
pymake lint # 格式化代码
pymake ca # 清理所有构建产物
pymake type # 类型检查
"""
runner = px.CliRunner(
strategy="sequential",
@@ -185,7 +117,7 @@ def main():
"t": px.Graph.from_specs([test]),
"tc": px.Graph.from_specs([test, test_coverage]),
"tf": px.Graph.from_specs([test_fast]),
"lint": px.Graph.from_specs([ruff_lint]),
"lint": px.Graph.from_specs([ruff_lint, ruff_format]),
"type": px.Graph.from_specs([mypy_check, ty_check]),
"doc": px.Graph.from_specs([doc]),
"pb": px.Graph.from_specs([twine_publish, hatch_publish]),
src/pyflowx/context.py
+1 -3
View File
@@ -82,9 +82,7 @@ def build_call_args(
)
# 与本任务相关的上下文子集。
dep_context: dict[str, Any] = {
name: context[name] for name in spec.depends_on if name in context
}
dep_context: dict[str, Any] = {name: context[name] for name in spec.depends_on if name in context}
# 检测静态 kwargs 与依赖名的冲突。
collisions = set(spec.kwargs) & set(dep_context)
src/pyflowx/executors.py
+2 -2
View File
@@ -96,7 +96,7 @@ def _run_sync_with_retry(
result: TaskResult[Any] = TaskResult(spec=spec)
# 检查条件是否满足
if spec.conditions and not spec.should_execute():
if not spec.should_execute():
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
logger.info("task %r skipped (条件不满足)", spec.name)
@@ -131,7 +131,7 @@ async def _run_async_with_retry(
result: TaskResult[Any] = TaskResult[Any](spec=spec)
# 检查条件是否满足
if spec.conditions and not spec.should_execute():
if not spec.should_execute():
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
logger.info("task %r skipped (条件不满足)", spec.name)
src/pyflowx/storage.py
+1 -3
View File
@@ -112,9 +112,7 @@ class JSONBackend(StateBackend):
try:
_ = json.dumps(value)
except (TypeError, ValueError) as exc:
raise StorageError(
f"result of task {name!r} is not JSON-serialisable", exc
) from exc
raise StorageError(f"result of task {name!r} is not JSON-serialisable", exc) from exc
self._store[name] = value
self._flush()
src/pyflowx/task.py
+31 -3
View File
@@ -112,6 +112,12 @@ class TaskSpec(Generic[T]):
是否在命令执行时显示详细输出。``True`` 时会打印执行的命令
及其标准输出/标准错误。仅在使用 ``cmd`` 参数时有效。
``False`` 时静默捕获输出（失败时仍会包含在错误信息中）。
skip_if_missing:
仅对 ``cmd`` 为 ``list[str]`` 的任务有效。``True`` 时自动检查
命令是否存在（通过 :func:`shutil.which`），不存在则跳过任务
（标记为 SKIPPED）而非失败。适用于构建工具场景，避免因未安装
某些工具（如 maturin、tox）而导致整个图执行失败。
对于 ``str`` (shell) 和 ``Callable`` 类型的 ``cmd``，此参数无效。
"""
name: str
@@ -126,6 +132,7 @@ class TaskSpec(Generic[T]):
conditions: Tuple[Condition, ...] = ()
cwd: Optional[Path] = None
verbose: bool = False
skip_if_missing: bool = True
def __post_init__(self) -> None:
if not self.name:
@@ -257,10 +264,31 @@ class TaskSpec(Generic[T]):
Returns
-------
bool
若所有条件都返回 ``True``，则返回 ``True``；
否则返回 ``False``。
若所有条件都返回 ``True``，且 ``skip_if_missing`` 检查通过，
则返回 ``True``；否则返回 ``False``。
"""
return all(condition() for condition in self.conditions)
if not all(condition() for condition in self.conditions):
return False
return not (self.skip_if_missing and not self._is_cmd_available())
def _is_cmd_available(self) -> bool:
"""检查 ``cmd`` 是否可用.
仅对 ``list[str]`` 类型的 ``cmd`` 进行检查（通过 :func:`shutil.which`）。
对于 ``str`` (shell) 和 ``Callable`` 类型，始终返回 ``True``。
Returns
-------
bool
命令可用返回 ``True``，否则返回 ``False``。
"""
import shutil
cmd = self.cmd
if isinstance(cmd, list) and cmd:
return shutil.which(cmd[0]) is not None
return True
@dataclass
tests/cli/test_pymake.py
+14 -33
View File
@@ -67,37 +67,6 @@ class TestMaturinBuildCmd:
assert "-Zbuild-std" not in cmd
# ---------------------------------------------------------------------- #
# check helper
# ---------------------------------------------------------------------- #
class TestCheckHelper:
"""Test check helper function."""
def test_check_returns_condition(self) -> None:
"""check() should return a Condition callable."""
cond = pymake.check("python")
assert callable(cond)
def test_check_uses_has_installed(self) -> None:
"""check() should use BuiltinConditions.HAS_INSTALLED."""
cond = pymake.check("python")
# The condition should be a callable that returns a bool
result = cond()
assert isinstance(result, bool)
def test_check_for_nonexistent_app(self) -> None:
"""check() for a nonexistent app should return False."""
cond = pymake.check("definitely_not_installed_app_xyz")
assert cond() is False
def test_check_for_python(self) -> None:
"""check() for python should return True (python is always available)."""
cond = pymake.check("python")
# On some systems, 'python' might not be in PATH, but 'python3' might be
# Just verify it returns a bool
assert isinstance(cond(), bool)
# ---------------------------------------------------------------------- #
# TaskSpec definitions
# ---------------------------------------------------------------------- #
@@ -108,23 +77,25 @@ class TestTaskSpecDefinitions:
"""uv_build spec should be properly defined."""
assert pymake.uv_build.name == "uv_build"
assert pymake.uv_build.cmd == ["uv", "build"]
assert len(pymake.uv_build.conditions) == 1
assert pymake.uv_build.skip_if_missing is True
def test_maturin_build_spec(self) -> None:
"""maturin_build spec should be properly defined."""
assert pymake.maturin_build.name == "maturin_build"
assert isinstance(pymake.maturin_build.cmd, list)
assert len(pymake.maturin_build.conditions) == 1
assert pymake.maturin_build.skip_if_missing is True
def test_uv_sync_spec(self) -> None:
"""uv_sync spec should be properly defined."""
assert pymake.uv_sync.name == "uv_sync"
assert pymake.uv_sync.cmd == ["uv", "sync"]
assert pymake.uv_sync.skip_if_missing is True
def test_git_clean_spec(self) -> None:
"""git_clean spec should be properly defined."""
assert pymake.git_clean.name == "git_clean"
assert pymake.git_clean.cmd == ["gitt", "c"]
assert pymake.git_clean.skip_if_missing is True
def test_test_spec(self) -> None:
"""test spec should be properly defined."""
@@ -133,6 +104,7 @@ class TestTaskSpecDefinitions:
assert "pytest" in pymake.test.cmd
assert "-m" in pymake.test.cmd
assert "not slow" in pymake.test.cmd
assert pymake.test.skip_if_missing is True
def test_test_fast_spec(self) -> None:
"""test_fast spec should be properly defined."""
@@ -140,6 +112,7 @@ class TestTaskSpecDefinitions:
assert isinstance(pymake.test_fast.cmd, list)
assert "pytest" in pymake.test_fast.cmd
assert "-n" not in pymake.test_fast.cmd # test_fast doesn't use parallel
assert pymake.test_fast.skip_if_missing is True
def test_test_coverage_spec(self) -> None:
"""test_coverage spec should be properly defined."""
@@ -147,6 +120,7 @@ class TestTaskSpecDefinitions:
assert isinstance(pymake.test_coverage.cmd, list)
assert "pytest" in pymake.test_coverage.cmd
assert "--cov" in pymake.test_coverage.cmd
assert pymake.test_coverage.skip_if_missing is True
def test_ruff_lint_spec(self) -> None:
"""ruff_lint spec should be properly defined."""
@@ -154,27 +128,32 @@ class TestTaskSpecDefinitions:
assert isinstance(pymake.ruff_lint.cmd, list)
assert "ruff" in pymake.ruff_lint.cmd
assert "check" in pymake.ruff_lint.cmd
assert pymake.ruff_lint.skip_if_missing is True
def test_mypy_check_spec(self) -> None:
"""mypy_check spec should be properly defined."""
assert pymake.mypy_check.name == "typecheck"
assert pymake.mypy_check.cmd == ["mypy", "."]
assert pymake.mypy_check.skip_if_missing is True
def test_ty_check_spec(self) -> None:
"""ty_check spec should be properly defined."""
assert pymake.ty_check.name == "ty_check"
assert pymake.ty_check.cmd == ["ty", "check", "."]
assert pymake.ty_check.skip_if_missing is True
def test_doc_spec(self) -> None:
"""doc spec should be properly defined."""
assert pymake.doc.name == "doc"
assert isinstance(pymake.doc.cmd, list)
assert "sphinx-build" in pymake.doc.cmd
assert pymake.doc.skip_if_missing is True
def test_hatch_publish_spec(self) -> None:
"""hatch_publish spec should be properly defined."""
assert pymake.hatch_publish.name == "publish_python"
assert pymake.hatch_publish.cmd == ["hatch", "publish"]
assert pymake.hatch_publish.skip_if_missing is True
def test_twine_publish_spec(self) -> None:
"""twine_publish spec should be properly defined."""
@@ -182,11 +161,13 @@ class TestTaskSpecDefinitions:
assert isinstance(pymake.twine_publish.cmd, list)
assert "twine" in pymake.twine_publish.cmd
assert "upload" in pymake.twine_publish.cmd
assert pymake.twine_publish.skip_if_missing is True
def test_tox_spec(self) -> None:
"""tox spec should be properly defined."""
assert pymake.tox.name == "tox"
assert pymake.tox.cmd == ["tox", "-p", "auto"]
assert pymake.tox.skip_if_missing is True
# ---------------------------------------------------------------------- #
tests/test_executors.py
+2 -2
View File
@@ -127,8 +127,8 @@ def test_threaded_parallelism() -> None:
report = px.run(graph, strategy="thread", max_workers=3)
elapsed = time.time() - start
assert report.success
# Three 0.3s tasks in parallel should be well under 0.8s.
assert elapsed < 0.8
# Three 0.3s tasks in parallel should be well under 1.0s.
assert elapsed < 1.0
@pytest.mark.slow
tests/test_task_edge_cases.py
+96
View File
@@ -211,3 +211,99 @@ def test_taskspec_shell_cmd_os_error_mocked():
RuntimeError, match="Shell 命令执行异常"
):
_ = wrapped_fn()
# ---------------------------------------------------------------------- #
# skip_if_missing
# ---------------------------------------------------------------------- #
def test_skip_if_missing_with_available_command():
"""skip_if_missing=True 时，命令存在应返回 True."""
# python 命令在测试环境中一定存在
spec = TaskSpec("test", cmd=["python", "--version"], skip_if_missing=True)
assert spec.should_execute() is True
def test_skip_if_missing_with_missing_command():
"""skip_if_missing=True 时，命令不存在应返回 False."""
spec = TaskSpec("test", cmd=["definitely_not_installed_app_xyz"], skip_if_missing=True)
assert spec.should_execute() is False
def test_skip_if_missing_false_with_missing_command():
"""skip_if_missing=False 时，命令不存在也应返回 True（不检查）."""
spec = TaskSpec("test", cmd=["definitely_not_installed_app_xyz"], skip_if_missing=False)
assert spec.should_execute() is True
def test_skip_if_missing_with_shell_cmd_not_checked():
"""skip_if_missing=True 时，shell 命令（str）不检查，应返回 True."""
spec = TaskSpec("test", cmd="definitely_not_installed_app_xyz", skip_if_missing=True)
assert spec.should_execute() is True
def test_skip_if_missing_with_callable_cmd_not_checked():
"""skip_if_missing=True 时，Callable 命令不检查，应返回 True."""
def custom_cmd() -> int:
return 0
spec = TaskSpec("test", cmd=custom_cmd, skip_if_missing=True)
assert spec.should_execute() is True
def test_skip_if_missing_with_fn_not_checked():
"""skip_if_missing=True 时，fn 任务不检查命令，应返回 True."""
def my_fn() -> int:
return 0
spec = TaskSpec("test", fn=my_fn, skip_if_missing=True)
assert spec.should_execute() is True
def test_skip_if_missing_with_empty_cmd_list():
"""skip_if_missing=True 时，空命令列表应返回 True（不检查）."""
spec = TaskSpec("test", cmd=[""], skip_if_missing=True)
# 空字符串命令，shutil.which 返回 None
# 但 cmd[0] 是空字符串，shutil.which("") 返回 None
assert spec.should_execute() is False
def test_skip_if_missing_combined_with_conditions():
"""skip_if_missing=True 与 conditions 组合使用."""
# conditions 返回 False，应跳过
spec = TaskSpec(
"test",
cmd=["python", "--version"],
skip_if_missing=True,
conditions=(lambda: False,),
)
assert spec.should_execute() is False
# conditions 返回 True，命令存在，应执行
spec = TaskSpec(
"test",
cmd=["python", "--version"],
skip_if_missing=True,
conditions=(lambda: True,),
)
assert spec.should_execute() is True
# conditions 返回 True，命令不存在，应跳过
spec = TaskSpec(
"test",
cmd=["definitely_not_installed_app_xyz"],
skip_if_missing=True,
conditions=(lambda: True,),
)
assert spec.should_execute() is False
def test_skip_if_missing_skips_task_in_run():
"""skip_if_missing=True 时，命令不存在的任务在 run 中应被跳过."""
spec = TaskSpec("missing_cmd", cmd=["definitely_not_installed_app_xyz"], skip_if_missing=True)
graph = px.Graph.from_specs([spec])
report = px.run(graph, strategy="sequential")
assert report.success is True
result = report.result_of("missing_cmd")
assert result.status == px.TaskStatus.SKIPPED
tests/test_taskspec_commands.py
+3 -1
View File
@@ -428,7 +428,9 @@ class TestTaskSpecCmdErrors:
"""命令不存在时应抛出 RuntimeError."""
from pyflowx.errors import TaskFailedError
graph = px.Graph.from_specs([px.TaskSpec("missing", cmd=["this-command-does-not-exist-xyz"])])
graph = px.Graph.from_specs(
[px.TaskSpec("missing", cmd=["this-command-does-not-exist-xyz"], skip_if_missing=False)],
)
with pytest.raises(TaskFailedError) as exc_info:
_ = px.run(graph, strategy="sequential")
# 错误信息应包含命令未找到