zhou
8d626f606e
CI / Lint, Typecheck & Test (push) Has been cancelled
pf graph 支持 ASCII 分层树/Mermaid/list/deps 四种输出格式； yaml_loader._safe_load 包装 YAMLError 为 ValueError 统一错误处理。
283 lines
8.5 KiB
Python
283 lines
8.5 KiB
Python
"""Tests for ``pf graph`` and ``pf yamlrun --progress`` CLI commands."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from pyflowx.cli.pf import PfApp
|
||
|
||
|
||
def _write_yaml(tmp_path: Path, content: str) -> Path:
|
||
"""写入 YAML 文件并返回路径。"""
|
||
yaml_file = tmp_path / "pipeline.yaml"
|
||
yaml_file.write_text(content, encoding="utf-8")
|
||
return yaml_file
|
||
|
||
|
||
_SIMPLE_YAML = """
|
||
jobs:
|
||
extract:
|
||
cmd: ["echo", "extract"]
|
||
tags: ["ingest"]
|
||
transform:
|
||
cmd: ["echo", "transform"]
|
||
needs: ["extract"]
|
||
tags: ["process"]
|
||
report:
|
||
cmd: ["echo", "report"]
|
||
needs: ["transform"]
|
||
tags: ["report"]
|
||
"""
|
||
|
||
_SOFT_DEP_YAML = """
|
||
jobs:
|
||
a:
|
||
cmd: ["echo", "a"]
|
||
b:
|
||
cmd: ["echo", "b"]
|
||
c:
|
||
cmd: ["echo", "c"]
|
||
needs: ["a"]
|
||
soft-needs: ["b"]
|
||
"""
|
||
|
||
|
||
class TestPfGraphCommand:
|
||
"""``pf graph <yaml>`` 命令测试。"""
|
||
|
||
def test_graph_ascii_default(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""默认 ASCII 渲染应输出分层视图与依赖表。"""
|
||
yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
|
||
app = PfApp(["graph", str(yaml_file)])
|
||
rc = app.run()
|
||
assert rc == 0
|
||
out = capsys.readouterr().out
|
||
assert "Graph" in out
|
||
assert "Layer 1" in out
|
||
assert "Layer 2" in out
|
||
assert "extract" in out
|
||
assert "transform" in out
|
||
assert "report" in out
|
||
# 依赖关系表也应被渲染
|
||
assert "硬依赖" in out
|
||
assert "标签" in out
|
||
|
||
def test_graph_mermaid_format(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""--format mermaid 应输出 Mermaid 语法。"""
|
||
yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
|
||
app = PfApp(["graph", str(yaml_file), "--format", "mermaid"])
|
||
rc = app.run()
|
||
assert rc == 0
|
||
out = capsys.readouterr().out
|
||
assert "graph TD" in out
|
||
assert 'extract["extract"]' in out
|
||
assert "extract --> transform" in out
|
||
assert "transform --> report" in out
|
||
|
||
def test_graph_mermaid_orientation_lr(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""--orientation LR 应改变 Mermaid 方向。"""
|
||
yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
|
||
app = PfApp(["graph", str(yaml_file), "--format", "mermaid", "--orientation", "LR"])
|
||
rc = app.run()
|
||
assert rc == 0
|
||
out = capsys.readouterr().out
|
||
assert "graph LR" in out
|
||
|
||
def test_graph_mermaid_invalid_orientation(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""非法 orientation 应被 to_mermaid 拒绝（ValueError），CLI 返回非零。"""
|
||
yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
|
||
app = PfApp(["graph", str(yaml_file), "--format", "mermaid", "--orientation", "XX"])
|
||
rc = app.run()
|
||
assert rc != 0
|
||
err = capsys.readouterr().err
|
||
assert "Invalid orientation" in err or "Traceback" in err
|
||
|
||
def test_graph_list_format(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""--format list 应仅列出任务名。"""
|
||
yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
|
||
app = PfApp(["graph", str(yaml_file), "--format", "list"])
|
||
rc = app.run()
|
||
assert rc == 0
|
||
out = capsys.readouterr().out
|
||
# 任务名按声明顺序输出
|
||
lines = [line.strip() for line in out.splitlines() if line.strip()]
|
||
assert "extract" in lines
|
||
assert "transform" in lines
|
||
assert "report" in lines
|
||
# 不应包含 ASCII 装饰
|
||
assert "Layer" not in out
|
||
assert "Graph" not in out
|
||
|
||
def test_graph_deps_format(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""--format deps 应输出依赖关系表。"""
|
||
yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
|
||
app = PfApp(["graph", str(yaml_file), "--format", "deps"])
|
||
rc = app.run()
|
||
assert rc == 0
|
||
out = capsys.readouterr().out
|
||
assert "任务依赖关系" in out
|
||
assert "硬依赖" in out
|
||
assert "软依赖" in out
|
||
assert "标签" in out
|
||
assert "extract" in out
|
||
assert "ingest" in out
|
||
|
||
def test_graph_soft_deps_displayed(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""软依赖应在依赖表中显示。"""
|
||
yaml_file = _write_yaml(tmp_path, _SOFT_DEP_YAML)
|
||
app = PfApp(["graph", str(yaml_file), "--format", "deps"])
|
||
rc = app.run()
|
||
assert rc == 0
|
||
out = capsys.readouterr().out
|
||
# c 的软依赖 b 应被显示
|
||
assert "b" in out
|
||
|
||
def test_graph_file_not_found(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""不存在的 YAML 文件应返回非零退出码与错误信息。"""
|
||
missing = tmp_path / "nonexistent.yaml"
|
||
app = PfApp(["graph", str(missing)])
|
||
rc = app.run()
|
||
assert rc != 0
|
||
err = capsys.readouterr().err
|
||
assert "加载 YAML 失败" in err
|
||
|
||
def test_graph_invalid_yaml(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""无效 YAML 应返回非零退出码与错误信息。"""
|
||
yaml_file = _write_yaml(tmp_path, "not: valid: yaml: [")
|
||
app = PfApp(["graph", str(yaml_file)])
|
||
rc = app.run()
|
||
assert rc != 0
|
||
err = capsys.readouterr().err
|
||
assert "加载 YAML 失败" in err
|
||
|
||
def test_graph_empty_yaml(
|
||
self,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
"""空 jobs（``jobs: {}``）应被 yaml_loader 拒绝并返回错误。"""
|
||
yaml_file = _write_yaml(tmp_path, "jobs: {}\n")
|
||
app = PfApp(["graph", str(yaml_file)])
|
||
rc = app.run()
|
||
assert rc != 0
|
||
err = capsys.readouterr().err
|
||
assert "加载 YAML 失败" in err
|
||
|
||
def test_graph_invalid_format_choice(
|
||
self,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""--format 非法值应触发 argparse SystemExit(2)。"""
|
||
yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
|
||
app = PfApp(["graph", str(yaml_file), "--format", "invalid"])
|
||
with pytest.raises(SystemExit) as exc_info:
|
||
app.run()
|
||
assert exc_info.value.code == 2
|
||
|
||
|
||
class TestPfYamlrunProgressFlag:
|
||
"""``pf yamlrun --progress`` 选项测试。"""
|
||
|
||
def test_yamlrun_progress_flag_executes_successfully(
|
||
self,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""``--progress`` 标志应能成功执行简单图。"""
|
||
yaml_file = _write_yaml(
|
||
tmp_path,
|
||
"""
|
||
jobs:
|
||
hello:
|
||
cmd: ["echo", "hello"]
|
||
""",
|
||
)
|
||
app = PfApp(["yamlrun", str(yaml_file), "--progress", "--quiet"])
|
||
rc = app.run()
|
||
assert rc == 0
|
||
|
||
def test_yamlrun_progress_with_only_flag(
|
||
self,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
"""``--progress`` 与 ``--only`` 应能组合使用。"""
|
||
yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
|
||
app = PfApp(["yamlrun", str(yaml_file), "--progress", "--quiet", "--only", "transform"])
|
||
rc = app.run()
|
||
assert rc == 0
|
||
|
||
def test_yamlrun_progress_suppressed_by_quiet(
|
||
self,
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""``--quiet`` 与 ``--progress`` 同时存在时，progress 应被禁用。"""
|
||
yaml_file = _write_yaml(
|
||
tmp_path,
|
||
"""
|
||
jobs:
|
||
hello:
|
||
cmd: ["echo", "hello"]
|
||
""",
|
||
)
|
||
|
||
captured_progress: dict[str, bool] = {}
|
||
|
||
def fake_run(*_args: object, **kwargs: object) -> object:
|
||
captured_progress["progress"] = bool(kwargs.get("progress"))
|
||
from pyflowx import RunReport
|
||
|
||
report = RunReport()
|
||
report.success = True
|
||
return report
|
||
|
||
# 拦截 pyflowx.run 调用以检查 progress 参数
|
||
import pyflowx
|
||
|
||
monkeypatch.setattr(pyflowx, "run", fake_run)
|
||
# pf.py 中通过 `from pyflowx import Graph, run` 局部导入，
|
||
# 所以 monkeypatch 替换 pyflowx.run 后，下次执行 _run_yaml 时会重新导入
|
||
|
||
app = PfApp(["yamlrun", str(yaml_file), "--progress", "--quiet"])
|
||
rc = app.run()
|
||
assert rc == 0
|
||
# quiet 优先级高于 progress
|
||
assert captured_progress["progress"] is False
|