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

feat(cli): add pxp performance profiler command

Browse Source 
1. 新增pxp CLI工具用于分析PyFlowX脚本生成性能报告
2. 新增ProfileReport.to_html方法生成自包含HTML报告
3. 新增完整的profiler功能测试用例
4. 更新pyproject.toml添加pxp入口点
5. 版本升级至0.2.12
This commit is contained in:
gooker_young
2026-06-28 20:30:17 +08:00
parent 3d6d769685
commit ce31f60441
5 changed files with 1074 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+1
View File
@@ -38,6 +38,7 @@ packtool = "pyflowx.cli.packtool:main"
pdftool = "pyflowx.cli.pdftool:main"
piptool = "pyflowx.cli.piptool:main"
pymake = "pyflowx.cli.pymake:main"
pxp = "pyflowx.cli.profiler:main"
reseticon = "pyflowx.cli.reseticoncache:main"
scrcap = "pyflowx.cli.screenshot:main"
sglang = "pyflowx.cli.llm.sglang:main"
src/pyflowx/cli/profiler.py
+272
View File
@@ -0,0 +1,272 @@
"""pxp —— PyFlowX 性能分析器.
分析包含 ``px`` 调用的 Python 脚本，生成工作流执行性能剖面报告。
工作原理
--------
1. 注入 hook：monkey-patch ``pyflowx.run`` / ``pyflowx.executors.run`` /
``pyflowx.runner.run``，捕获最后一次执行的 ``Graph`` 与 ``RunReport``。
2. 执行目标脚本：用 ``runpy.run_path`` 以 ``__main__`` 身份执行，
捕获 ``SystemExit``（脚本可能调 ``sys.exit``）。
3. 生成报告：从捕获的 report + graph 构建 :class:`ProfileReport`，
默认输出 HTML 并自动打开浏览器。
使用方式
--------
# 分析 pymake.py，生成 HTML 报告并打开浏览器
pxp pymake.py
# 传递参数给被分析脚本（用 -- 分隔）
pxp pymake.py -- t
# 指定输出文件
pxp pymake.py -o report.html
# 不打开浏览器
pxp pymake.py --no-browser
# 输出纯文本报告
pxp pymake.py -E text
"""
from __future__ import annotations
__all__ = ["main"]
import argparse
import runpy
import sys
import webbrowser
from pathlib import Path
from typing import Any
from .. import executors as _executors
from .. import runner as _runner
from ..profiling import ProfileReport
from ..report import RunReport
def _build_parser() -> argparse.ArgumentParser:
"""构建参数解析器。"""
parser = argparse.ArgumentParser(
prog="pxp",
description="PyFlowX 性能分析器：分析包含 px 调用的脚本，生成性能剖面报告。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例:\n"
" pxp pymake.py # 分析并打开 HTML 报告\n"
" pxp pymake.py -- t # 传递参数 t 给脚本\n"
" pxp pymake.py -E text # 输出纯文本报告\n"
" pxp pymake.py -o out.html # 指定输出文件\n"
),
)
_ = parser.add_argument(
"--export",
"-E",
choices=["html", "text"],
default="html",
help="导出格式（默认: html）",
)
_ = parser.add_argument(
"--no-browser",
action="store_true",
help="不自动打开浏览器（仅 HTML 格式有效）",
)
_ = parser.add_argument(
"-o",
"--output",
help="输出文件路径（默认: <script>_profile.html）",
)
return parser
def _capture_px_run() -> dict[str, Any]:
"""注入 hook 捕获 px.run() 调用。
返回一个字典，``run()`` 执行后填充 ``graph`` 与 ``report``。
同时返回还原函数用于 finally 块。
Note
-----
需同时 patch 三处引用：
* ``pyflowx.executors.run`` —— 实际实现
* ``pyflowx.runner.run`` —— ``CliRunner`` 直接 import 的引用
* ``pyflowx.run`` —— 顶层包导出的引用（用户脚本常用 ``px.run()``）
另外 patch ``RunReport.__init__`` 以捕获 ``run()`` 内部创建的 report 实例。
这对于 ``run()`` 抛出 ``TaskFailedError`` 的场景至关重要：此时 ``run()``
不会正常返回 report，但 report 对象已在内部创建并填充了已执行任务的结果。
通过 ``capture_enabled`` 标志确保只在 ``patched_run`` 调用期间捕获。
"""
captured: dict[str, Any] = {}
original_exec_run = _executors.run
original_runner_run = _runner.run
# 惰性获取顶层 pyflowx.run 引用（避免循环导入）
import pyflowx as px_mod
original_px_run = px_mod.run
original_report_init = RunReport.__init__
capture_enabled = [False]
def patched_report_init(self: RunReport, *args: Any, **kwargs: Any) -> None:
original_report_init(self, *args, **kwargs)
if capture_enabled[0]:
captured["report"] = self
RunReport.__init__ = patched_report_init # type: ignore[assignment]
def patched_run(graph: Any, *args: Any, **kwargs: Any) -> RunReport:
captured["graph"] = graph
capture_enabled[0] = True
try:
report = original_exec_run(graph, *args, **kwargs)
# 正常返回时确保 captured["report"] 是返回的 report
captured["report"] = report
return report
finally:
capture_enabled[0] = False
# patch 所有引用 run 的入口
_executors.run = patched_run # type: ignore[assignment]
_runner.run = patched_run # type: ignore[assignment]
px_mod.run = patched_run # type: ignore[assignment]
def _restore() -> None:
_executors.run = original_exec_run # type: ignore[assignment]
_runner.run = original_runner_run # type: ignore[assignment]
px_mod.run = original_px_run # type: ignore[assignment]
RunReport.__init__ = original_report_init # type: ignore[assignment]
captured["_restore"] = _restore
return captured
def _run_target_script(script: Path, script_args: list[str]) -> dict[str, Any]:
"""执行目标脚本。
将脚本所在目录加入 ``sys.path``，设置 ``sys.argv``，然后用
``runpy.run_path`` 以 ``__main__`` 身份执行。捕获 ``SystemExit``。
Returns
-------
dict[str, Any]
脚本模块的全局变量字典（含 ``main`` 等定义）。
"""
sys.argv = [str(script), *script_args]
script_dir = str(script.parent.resolve())
if script_dir not in sys.path:
sys.path.insert(0, script_dir)
return runpy.run_path(str(script), run_name="__main__")
def _try_call_main(module_globals: dict[str, Any]) -> None:
"""若模块定义了 ``main`` 可调用对象，调用它。
用于脚本无 ``if __name__ == "__main__"`` 块的场景（如通过 entry points
注册的 CLI 工具脚本）。``main`` 通常调用 ``CliRunner.run_cli()``，
后者读取 ``sys.argv[1:]`` 执行对应命令。
"""
main_fn = module_globals.get("main")
if callable(main_fn):
main_fn()
def _output_report(
profile: ProfileReport,
export: str,
output: str | None,
script_stem: str,
no_browser: bool,
) -> None:
"""输出性能报告。"""
if export == "text":
print(profile.describe())
return
# HTML 格式
html = profile.to_html()
if output:
out_path = Path(output)
else:
out_path = Path.cwd() / f"{script_stem}_profile.html"
out_path.write_text(html, encoding="utf-8")
print(f"HTML 报告已生成: {out_path}")
if not no_browser:
try:
webbrowser.open(f"file://{out_path.resolve()}")
except Exception as e:
print(f"警告：无法打开浏览器: {e}", file=sys.stderr)
def main() -> None:
"""pxp CLI 入口。"""
parser = _build_parser()
pxp_args, remaining = parser.parse_known_args()
if not remaining:
parser.print_help()
sys.exit(2)
script_str = remaining[0]
script_args = remaining[1:]
script_path = Path(script_str).resolve()
if not script_path.is_file():
print(f"错误：脚本不存在: {script_path}", file=sys.stderr)
sys.exit(2)
# 注入 hook
captured = _capture_px_run()
# 执行目标脚本
print(f"正在分析: {script_path}")
if script_args:
print(f"脚本参数: {script_args}")
print("-" * 60)
module_globals: dict[str, Any] = {}
try:
module_globals = _run_target_script(script_path, script_args)
except SystemExit:
# 脚本调用了 sys.exit，正常情况
pass
except Exception as e:
print(f"警告：脚本执行抛出异常: {e}", file=sys.stderr)
# 若脚本执行未捕获到 run()，尝试调用模块的 main() 函数
# （适用于无 ``if __name__ == "__main__"`` 块的 CLI 脚本）
if captured.get("report") is None and module_globals:
try:
_try_call_main(module_globals)
except SystemExit:
pass
except Exception as e:
print(f"警告：调用 main() 抛出异常: {e}", file=sys.stderr)
# 还原 hook
restore = captured.pop("_restore", None)
if restore is not None:
restore()
# 检查是否捕获到 run() 调用
report = captured.get("report")
graph = captured.get("graph")
if report is None or graph is None:
print("错误：未捕获到 px.run() 调用，无法生成性能报告", file=sys.stderr)
print("请确保脚本通过 px.run() 或 CliRunner 执行任务流图。", file=sys.stderr)
sys.exit(1)
# 生成报告
profile = ProfileReport.from_report(report, graph)
_output_report(
profile,
export=pxp_args.export,
output=pxp_args.output,
script_stem=script_path.stem,
no_browser=pxp_args.no_browser,
)
if __name__ == "__main__":
main()
src/pyflowx/profiling.py
+255 -1
View File
@@ -384,8 +384,15 @@ class ProfileReport:
"bottlenecks": [t.to_dict() for t in self.top_bottlenecks(5)],
}
def to_html(self) -> str:
"""生成自包含的 HTML 报告（含 CSS，无外部依赖）。
报告含：图级指标卡片、关键路径、时间线甘特图、Top 瓶颈表格、
全部任务表格。适合直接用浏览器打开查看。
"""
return _render_html(self)
def describe(self) -> str:
"""人类可读的多行性能报告。"""
lines: list[str] = []
lines.append("=" * 70)
lines.append("PyFlowX 性能剖面报告")
@@ -449,3 +456,250 @@ class ProfileReport:
f"avg_par={self.avg_parallelism:.2f}, "
f"peak_par={self.peak_parallelism})"
)
# ---------------------------------------------------------------------- #
# HTML 渲染（私有，零依赖）
# ---------------------------------------------------------------------- #
_HTML_TEMPLATE = """<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>PyFlowX 性能剖面报告</title>
<style>
:root {{
--bg: #f5f5f7;
--card: #ffffff;
--border: #d2d2d7;
--text: #1d1d1f;
--muted: #6e6e73;
--accent: #0071e3;
--success: #34c759;
--warning: #ff9f0a;
--danger: #ff3b30;
--critical: #af52de;
}}
* {{ box-sizing: border-box; }}
body {{
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
margin: 0;
padding: 24px;
background: var(--bg);
color: var(--text);
line-height: 1.5;
}}
h1 {{ margin: 0 0 8px; font-size: 28px; }}
h2 {{ margin: 32px 0 12px; font-size: 20px; border-bottom: 1px solid var(--border); padding-bottom: 6px; }}
.subtitle {{ color: var(--muted); margin: 0 0 24px; font-size: 14px; }}
.cards {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 12px; margin-bottom: 8px; }}
.card {{
background: var(--card);
border: 1px solid var(--border);
border-radius: 10px;
padding: 16px;
}}
.card .label {{ font-size: 12px; color: var(--muted); margin-bottom: 4px; text-transform: uppercase; letter-spacing: 0.5px; }}
.card .value {{ font-size: 22px; font-weight: 600; }}
.card .unit {{ font-size: 13px; color: var(--muted); margin-left: 2px; }}
.critical-path {{
background: var(--card);
border: 1px solid var(--border);
border-left: 4px solid var(--critical);
border-radius: 10px;
padding: 16px;
margin-bottom: 8px;
}}
.critical-path .label {{ font-size: 12px; color: var(--muted); margin-bottom: 8px; text-transform: uppercase; letter-spacing: 0.5px; }}
.critical-path .chain {{ font-family: ui-monospace, "SF Mono", Menlo, monospace; font-size: 13px; word-break: break-all; }}
.critical-path .arrow {{ color: var(--critical); margin: 0 6px; font-weight: 600; }}
/* 甘特图 */
.gantt {{
background: var(--card);
border: 1px solid var(--border);
border-radius: 10px;
padding: 16px;
overflow-x: auto;
}}
.gantt-row {{ display: flex; align-items: center; margin-bottom: 6px; min-width: 600px; }}
.gantt-label {{ width: 200px; flex-shrink: 0; font-size: 13px; font-family: ui-monospace, monospace; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }}
.gantt-track {{ flex: 1; height: 22px; background: #f0f0f3; border-radius: 4px; position: relative; }}
.gantt-bar {{ position: absolute; height: 100%; border-radius: 4px; min-width: 2px; }}
.gantt-bar.success {{ background: var(--success); }}
.gantt-bar.failed {{ background: var(--danger); }}
.gantt-bar.skipped {{ background: var(--muted); }}
.gantt-bar.critical {{ box-shadow: 0 0 0 2px var(--critical) inset; }}
.gantt-bar:hover {{ opacity: 0.85; }}
.gantt-tooltip {{ position: absolute; bottom: 100%; left: 50%; transform: translateX(-50%); background: #1d1d1f; color: #fff; padding: 4px 8px; border-radius: 4px; font-size: 11px; white-space: nowrap; opacity: 0; pointer-events: none; transition: opacity 0.15s; }}
.gantt-bar:hover .gantt-tooltip {{ opacity: 1; }}
/* 表格 */
table {{ width: 100%; border-collapse: collapse; background: var(--card); border-radius: 10px; overflow: hidden; border: 1px solid var(--border); }}
th, td {{ padding: 10px 12px; text-align: left; font-size: 13px; }}
th {{ background: #fafafa; font-weight: 600; color: var(--muted); text-transform: uppercase; font-size: 11px; letter-spacing: 0.5px; }}
tbody tr {{ border-top: 1px solid var(--border); }}
tbody tr:hover {{ background: #fafafa; }}
td.num {{ font-family: ui-monospace, monospace; text-align: right; }}
.badge {{ display: inline-block; padding: 2px 8px; border-radius: 10px; font-size: 11px; font-weight: 500; }}
.badge.success {{ background: rgba(52,199,89,0.15); color: var(--success); }}
.badge.failed {{ background: rgba(255,59,48,0.15); color: var(--danger); }}
.badge.skipped {{ background: rgba(110,110,115,0.15); color: var(--muted); }}
.star {{ color: var(--critical); font-weight: 700; }}
.footer {{ margin-top: 32px; color: var(--muted); font-size: 12px; text-align: center; }}
</style>
</head>
<body>
<h1>PyFlowX 性能剖面报告</h1>
<p class="subtitle">由 <code>pxp</code> 生成 · {generated_at}</p>
<h2>图级指标</h2>
<div class="cards">
<div class="card"><div class="label">总耗时</div><div class="value">{total_duration:.3f}<span class="unit">s</span></div></div>
<div class="card"><div class="label">关键路径耗时</div><div class="value">{critical_duration:.3f}<span class="unit">s</span></div></div>
<div class="card"><div class="label">平均并行度</div><div class="value">{avg_par:.2f}</div></div>
<div class="card"><div class="label">峰值并行度</div><div class="value">{peak_par}</div></div>
<div class="card"><div class="label">并行度效率</div><div class="value">{efficiency:.1f}<span class="unit">%</span></div></div>
<div class="card"><div class="label">任务总数</div><div class="value">{task_count}</div></div>
</div>
<h2>关键路径</h2>
<div class="critical-path">
<div class="label">最长依赖路径（串行瓶颈）</div>
<div class="chain">{critical_chain}</div>
</div>
<h2>任务时间线</h2>
<div class="gantt">
{gantt_rows}
</div>
<h2>Top 瓶颈任务</h2>
<table>
<thead><tr><th>任务</th><th class="num">耗时</th><th class="num">等待</th><th class="num">尝试</th><th>关键路径</th><th>状态</th></tr></thead>
<tbody>
{bottleneck_rows}
</tbody>
</table>
<h2>全部任务</h2>
<table>
<thead><tr><th>任务</th><th class="num">耗时</th><th class="num">等待</th><th class="num">尝试</th><th>关键路径</th><th>状态</th><th>依赖</th></tr></thead>
<tbody>
{all_task_rows}
</tbody>
</table>
<div class="footer">由 PyFlowX · pxp 生成</div>
</body>
</html>"""
def _status_badge(status: TaskStatus) -> str:
"""生成状态徽章 HTML。"""
cls = status.value
return f'<span class="badge {cls}">{cls}</span>'
def _format_critical_chain(path: tuple[str, ...]) -> str:
"""格式化关键路径为 HTML 链。"""
if not path:
return '<em style="color:var(--muted)">(无)</em>'
arrow = '<span class="arrow">→</span>'
return arrow.join(f"<strong>{name}</strong>" for name in path)
def _render_gantt(profile: ProfileReport) -> str:
"""渲染甘特图行 HTML。
每个任务一行：标签 + 时间条。时间条位置基于 wait_time + 依赖关系
重建相对开始时间（相对最早任务起点），归一化到 0-100% 宽度。
SKIPPED 任务不显示（无时间戳）。
"""
visible = [t for t in profile.tasks if t.status != TaskStatus.SKIPPED and t.duration > 0]
if not visible:
return '<div style="color:var(--muted);padding:12px;">(无时间线数据)</div>'
# 重建相对开始时间：start[name] = max(end[dep]) + wait_time
# profile.tasks 已是拓扑序，可直接按序计算
start: dict[str, float] = {}
end: dict[str, float] = {}
for t in profile.tasks:
if t.status == TaskStatus.SKIPPED:
continue
dep_end = 0.0
for dep in t.deps:
dep_end = max(dep_end, end.get(dep, 0.0))
s = dep_end + t.wait_time
start[t.name] = s
end[t.name] = s + t.duration
# 归一化：以最早开始时间为 0，最晚结束为 100%
min_start = min(start.get(t.name, 0.0) for t in visible)
max_end = max(end.get(t.name, 0.0) for t in visible)
span = max_end - min_start
if span <= 0:
span = 1.0
rows: list[str] = []
for t in visible:
s = start.get(t.name, 0.0) - min_start
left_pct = (s / span) * 100
width_pct = (t.duration / span) * 100
cls = t.status.value
critical_cls = " critical" if t.is_on_critical_path else ""
tooltip = f"{t.name}: {t.duration:.3f}s @ +{s:.3f}s ({t.status.value})"
rows.append(
f' <div class="gantt-row">'
f'<div class="gantt-label" title="{t.name}">{t.name}</div>'
f'<div class="gantt-track">'
f'<div class="gantt-bar {cls}{critical_cls}" style="left:{left_pct:.2f}%;width:{width_pct:.2f}%">'
f'<span class="gantt-tooltip">{tooltip}</span>'
f"</div></div></div>"
)
return "\n".join(rows)
def _render_task_row(t: TaskProfile, show_deps: bool = False) -> str:
"""渲染任务表格行 HTML。"""
star = '<span class="star">★</span>' if t.is_on_critical_path else ""
deps = ", ".join(t.deps) if show_deps and t.deps else ""
deps_cell = f"<td>{deps}</td>" if show_deps else ""
return (
f" <tr>"
f"<td><code>{t.name}</code></td>"
f'<td class="num">{t.duration:.3f}s</td>'
f'<td class="num">{t.wait_time:.3f}s</td>'
f'<td class="num">{t.attempts}</td>'
f"<td>{star}</td>"
f"<td>{_status_badge(t.status)}</td>"
f"{deps_cell}"
f"</tr>"
)
def _render_html(profile: ProfileReport) -> str:
"""渲染完整 HTML 报告。"""
from datetime import datetime as _dt
bottlenecks = profile.top_bottlenecks(5)
bottleneck_rows = (
"\n".join(_render_task_row(t) for t in bottlenecks)
or ' <tr><td colspan="6" style="color:var(--muted);">(无)</td></tr>'
)
all_task_rows = (
"\n".join(_render_task_row(t, show_deps=True) for t in profile.tasks)
or ' <tr><td colspan="7" style="color:var(--muted);">(无)</td></tr>'
)
return _HTML_TEMPLATE.format(
generated_at=_dt.now().strftime("%Y-%m-%d %H:%M:%S"),
total_duration=profile.total_duration,
critical_duration=profile.critical_path_duration,
avg_par=profile.avg_parallelism,
peak_par=profile.peak_parallelism,
efficiency=profile.parallelism_efficiency * 100,
task_count=len(profile.tasks),
critical_chain=_format_critical_chain(profile.critical_path),
gantt_rows=_render_gantt(profile),
bottleneck_rows=bottleneck_rows,
all_task_rows=all_task_rows,
)
tests/cli/test_profiler.py
+545
View File
@@ -0,0 +1,545 @@
"""pxp 性能分析器测试.
覆盖策略：
* HTML 渲染：to_html() 输出结构正确，含关键章节。
* pxp CLI：参数解析、脚本执行、报告生成、浏览器调用、错误处理。
* hook 注入：捕获 px.run() 调用，还原原始函数。
"""
from __future__ import annotations
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
import pytest
import pyflowx as px
from pyflowx.cli import profiler
from pyflowx.profiling import ProfileReport
from pyflowx.report import RunReport
from pyflowx.task import TaskResult, TaskSpec, TaskStatus
def _fn() -> int:
return 1
def _spec(name: str, deps: tuple[str, ...] = ()) -> TaskSpec[Any]:
return TaskSpec[Any](name, _fn, depends_on=deps)
def _result(
name: str,
start: datetime,
duration: float,
*,
status: TaskStatus = TaskStatus.SUCCESS,
attempts: int = 1,
) -> TaskResult[Any]:
"""构造带时间戳的 TaskResult."""
end = start + timedelta(seconds=duration) if duration > 0 else start
return TaskResult[Any](
spec=_spec(name),
status=status,
value=None,
attempts=attempts,
started_at=start if duration > 0 or status != TaskStatus.SKIPPED else None,
finished_at=end if duration > 0 or status != TaskStatus.SKIPPED else None,
)
def _build_simple_profile() -> ProfileReport:
"""构造一个简单的 ProfileReport 用于测试 HTML 输出."""
start = datetime(2024, 1, 1, 0, 0, 0)
report = px.RunReport()
report.results["a"] = _result("a", start, 1.0)
report.results["b"] = _result("b", start + timedelta(seconds=1), 2.0)
graph = px.Graph.from_specs([
_spec("a"),
_spec("b", deps=("a",)),
])
return ProfileReport.from_report(report, graph)
class TestToHtml:
"""测试 ProfileReport.to_html()."""
def test_to_html_contains_key_sections(self) -> None:
"""HTML 应包含所有关键章节标题。"""
profile = _build_simple_profile()
html = profile.to_html()
assert "<!DOCTYPE html>" in html
assert "PyFlowX 性能剖面报告" in html
assert "图级指标" in html
assert "关键路径" in html
assert "任务时间线" in html
assert "Top 瓶颈任务" in html
assert "全部任务" in html
def test_to_html_contains_metrics(self) -> None:
"""HTML 应包含图级指标数值。"""
profile = _build_simple_profile()
html = profile.to_html()
# 总耗时 3.0s (a=1 + b=2)
assert "3.000" in html
# 任务名
assert "a" in html
assert "b" in html
def test_to_html_contains_critical_path(self) -> None:
"""HTML 应包含关键路径任务链。"""
profile = _build_simple_profile()
html = profile.to_html()
# 关键路径是 a -> b
assert "<strong>a</strong>" in html
assert "<strong>b</strong>" in html
def test_to_html_contains_gantt_bars(self) -> None:
"""HTML 应包含甘特图条。"""
profile = _build_simple_profile()
html = profile.to_html()
assert "gantt-row" in html
assert "gantt-bar" in html
# 每个非 SKIPPED 任务一个条
assert html.count("gantt-bar") >= 2
def test_to_html_empty_profile(self) -> None:
"""空报告的 HTML 应不崩溃。"""
report = px.RunReport()
graph = px.Graph()
profile = ProfileReport.from_report(report, graph)
html = profile.to_html()
assert "PyFlowX 性能剖面报告" in html
assert "(无)" in html
def test_to_html_with_failed_task(self) -> None:
"""含 FAILED 任务的 HTML 应包含失败状态徽章。"""
start = datetime(2024, 1, 1, 0, 0, 0)
report = px.RunReport()
report.results["a"] = _result("a", start, 1.0, status=TaskStatus.FAILED)
graph = px.Graph.from_specs([_spec("a")])
profile = ProfileReport.from_report(report, graph)
html = profile.to_html()
assert "failed" in html
assert "badge" in html
def test_to_html_with_skipped_task(self) -> None:
"""含 SKIPPED 任务的 HTML 不应在甘特图中显示该任务。"""
start = datetime(2024, 1, 1, 0, 0, 0)
report = px.RunReport()
report.results["a"] = _result("a", start, 1.0)
report.results["b"] = TaskResult[Any](
spec=_spec("b"),
status=TaskStatus.SKIPPED,
reason="skip",
)
graph = px.Graph.from_specs([_spec("a"), _spec("b")])
profile = ProfileReport.from_report(report, graph)
html = profile.to_html()
# SKIPPED 任务的徽章应出现
assert "skipped" in html
def test_to_html_self_contained(self) -> None:
"""HTML 应自包含（无外部依赖）。"""
profile = _build_simple_profile()
html = profile.to_html()
# 不引用外部资源
assert "<link" not in html
assert "<script src" not in html
class TestProfilerArgumentParsing:
"""测试 pxp CLI 参数解析。"""
def test_default_export_is_html(self) -> None:
"""默认导出格式为 html。"""
parser = profiler._build_parser()
args, remaining = parser.parse_known_args(["pymake.py"])
assert args.export == "html"
assert args.no_browser is False
assert args.output is None
assert remaining == ["pymake.py"]
def test_export_text(self) -> None:
"""-E text 应设置导出格式为 text。"""
parser = profiler._build_parser()
args, _ = parser.parse_known_args(["-E", "text", "pymake.py"])
assert args.export == "text"
def test_no_browser_flag(self) -> None:
"""--no-browser 应设置标志。"""
parser = profiler._build_parser()
args, _ = parser.parse_known_args(["--no-browser", "pymake.py"])
assert args.no_browser is True
def test_output_option(self) -> None:
"""-o 应设置输出路径。"""
parser = profiler._build_parser()
args, _ = parser.parse_known_args(["-o", "report.html", "pymake.py"])
assert args.output == "report.html"
def test_script_args_separated(self) -> None:
"""脚本参数应通过 remaining 分离。"""
parser = profiler._build_parser()
_, remaining = parser.parse_known_args(["pymake.py", "t", "--quiet"])
assert remaining == ["pymake.py", "t", "--quiet"]
def test_no_args_prints_help(
self,
capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""无参数应打印帮助并以退出码 2 退出。"""
monkeypatch.setattr(sys, "argv", ["pxp"])
with pytest.raises(SystemExit) as exc_info:
profiler.main()
assert exc_info.value.code == 2
captured = capsys.readouterr()
assert "usage" in captured.out.lower() or "usage" in captured.err.lower()
class TestCapturePxRun:
"""测试 _capture_px_run hook 注入。"""
def test_capture_captures_run_call(self) -> None:
"""hook 应捕获 px.run() 调用的 graph 和 report。"""
captured = profiler._capture_px_run()
try:
graph = px.Graph.from_specs([px.TaskSpec("a", lambda: 1)])
px.run(graph, strategy="sequential")
assert "graph" in captured
assert "report" in captured
assert captured["graph"] is graph
finally:
captured["_restore"]()
def test_capture_restores_original(self) -> None:
"""还原后 px.run 和 RunReport.__init__ 应恢复为原函数。"""
original_run = px.run
original_init = RunReport.__init__
captured = profiler._capture_px_run()
# 注入期间 px.run 和 RunReport.__init__ 已被替换
assert px.run is not original_run
assert RunReport.__init__ is not original_init
captured["_restore"]()
# 还原后恢复
assert px.run is original_run
assert RunReport.__init__ is original_init
def test_capture_via_runner_run(self) -> None:
"""hook 应捕获通过 CliRunner 执行的 run() 调用。"""
from pyflowx import runner as runner_mod
captured = profiler._capture_px_run()
try:
# 验证 runner.run 也被 patch（指向 patched_run）
assert runner_mod.run is px.executors.run
graph = px.Graph.from_specs([px.TaskSpec("a", lambda: 1)])
runner_mod.run(graph, strategy="sequential")
assert "report" in captured
finally:
captured["_restore"]()
def test_capture_captures_report_on_failure(self) -> None:
"""run() 抛出 TaskFailedError 时仍应捕获 report 实例。"""
from pyflowx.executors import TaskFailedError
def failing() -> None:
raise RuntimeError("boom")
graph = px.Graph.from_specs([px.TaskSpec("a", failing)])
captured = profiler._capture_px_run()
try:
with pytest.raises(TaskFailedError):
px.run(graph, strategy="sequential")
# 即使 run() 抛异常，report 也应被捕获（含已执行任务的结果）
assert "report" in captured
assert "graph" in captured
assert captured["graph"] is graph
finally:
captured["_restore"]()
class TestRunTargetScript:
"""测试 _run_target_script。"""
def test_run_simple_script(self, tmp_path: Path) -> None:
"""应能执行简单脚本并返回模块字典。"""
script = tmp_path / "simple.py"
script.write_text("x = 42\n", encoding="utf-8")
result = profiler._run_target_script(script, [])
assert result["x"] == 42
def test_run_script_with_sys_exit(self, tmp_path: Path) -> None:
"""脚本调用 sys.exit 应抛 SystemExit。"""
script = tmp_path / "exit.py"
script.write_text("import sys; sys.exit(0)\n", encoding="utf-8")
with pytest.raises(SystemExit):
profiler._run_target_script(script, [])
def test_run_script_sets_argv(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""应正确设置 sys.argv。"""
script = tmp_path / "argv.py"
script.write_text(
"import sys\nassert sys.argv[0] == __file__\nassert sys.argv[1:] == ['arg1', 'arg2']\n",
encoding="utf-8",
)
profiler._run_target_script(script, ["arg1", "arg2"])
def test_run_script_adds_dir_to_path(self, tmp_path: Path) -> None:
"""脚本所在目录应加入 sys.path。"""
script = tmp_path / "pathcheck.py"
script.write_text(
"import sys, os\nassert os.path.dirname(__file__) in sys.path\n",
encoding="utf-8",
)
profiler._run_target_script(script, [])
class TestOutputReport:
"""测试 _output_report。"""
def test_output_text_format(
self,
capsys: pytest.CaptureFixture[str],
) -> None:
"""text 格式应打印 describe() 到 stdout。"""
profile = _build_simple_profile()
profiler._output_report(profile, export="text", output=None, script_stem="test", no_browser=True)
captured = capsys.readouterr()
assert "PyFlowX 性能剖面报告" in captured.out
assert "图级指标" in captured.out
def test_output_html_default_filename(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""HTML 默认输出到 <script>_profile.html。"""
monkeypatch.chdir(tmp_path)
profile = _build_simple_profile()
profiler._output_report(profile, export="html", output=None, script_stem="mymake", no_browser=True)
out_file = tmp_path / "mymake_profile.html"
assert out_file.exists()
content = out_file.read_text(encoding="utf-8")
assert "PyFlowX 性能剖面报告" in content
def test_output_html_custom_path(self, tmp_path: Path) -> None:
"""HTML 应写入指定路径。"""
out_file = tmp_path / "custom.html"
profile = _build_simple_profile()
profiler._output_report(profile, export="html", output=str(out_file), script_stem="test", no_browser=True)
assert out_file.exists()
assert "PyFlowX" in out_file.read_text(encoding="utf-8")
def test_output_html_opens_browser(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""no_browser=False 应调用 webbrowser.open。"""
monkeypatch.chdir(tmp_path)
opened: list[str] = []
monkeypatch.setattr(profiler.webbrowser, "open", opened.append)
profile = _build_simple_profile()
profiler._output_report(profile, export="html", output=None, script_stem="test", no_browser=False)
assert len(opened) == 1
assert opened[0].startswith("file://")
def test_output_html_no_browser_flag(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""no_browser=True 不应调用 webbrowser.open。"""
monkeypatch.chdir(tmp_path)
opened: list[str] = []
monkeypatch.setattr(profiler.webbrowser, "open", opened.append)
profile = _build_simple_profile()
profiler._output_report(profile, export="html", output=None, script_stem="test", no_browser=True)
assert len(opened) == 0
class TestProfilerMainIntegration:
"""main() 集成测试。"""
def test_main_analyses_script_with_px_run(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""main() 应分析含 px.run() 的脚本并生成 HTML。"""
script = tmp_path / "mytool.py"
script.write_text(
"import pyflowx as px\n"
"graph = px.Graph.from_specs([\n"
" px.TaskSpec('a', lambda: 1),\n"
" px.TaskSpec('b', lambda: 2, depends_on=('a',)),\n"
"])\n"
"px.run(graph, strategy='sequential')\n",
encoding="utf-8",
)
out_file = tmp_path / "report.html"
monkeypatch.setattr(sys, "argv", ["pxp", "--no-browser", "-o", str(out_file), str(script)])
profiler.main()
assert out_file.exists()
content = out_file.read_text(encoding="utf-8")
assert "PyFlowX 性能剖面报告" in content
assert "任务时间线" in content
def test_main_analyses_script_with_clirunner(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""main() 应分析含 CliRunner 的脚本。"""
script = tmp_path / "clirunner_tool.py"
script.write_text(
"import pyflowx as px\n"
"runner = px.CliRunner(\n"
" aliases={'t': px.TaskSpec('t', lambda: 1)},\n"
")\n"
"runner.run_cli(['t'])\n",
encoding="utf-8",
)
out_file = tmp_path / "report.html"
monkeypatch.setattr(sys, "argv", ["pxp", "--no-browser", "-o", str(out_file), str(script)])
profiler.main()
assert out_file.exists()
content = out_file.read_text(encoding="utf-8")
assert "PyFlowX 性能剖面报告" in content
def test_main_text_export(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""main() -E text 应输出文本到 stdout。"""
script = tmp_path / "simple.py"
script.write_text(
"import pyflowx as px\n"
"graph = px.Graph.from_specs([px.TaskSpec('a', lambda: 1)])\n"
"px.run(graph, strategy='sequential')\n",
encoding="utf-8",
)
monkeypatch.setattr(sys, "argv", ["pxp", "-E", "text", "--no-browser", str(script)])
profiler.main()
captured = capsys.readouterr()
assert "PyFlowX 性能剖面报告" in captured.out
def test_main_script_not_exist(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""脚本不存在应以退出码 2 退出。"""
monkeypatch.setattr(sys, "argv", ["pxp", "--no-browser", str(tmp_path / "nonexistent.py")])
with pytest.raises(SystemExit) as exc_info:
profiler.main()
assert exc_info.value.code == 2
def test_main_no_px_run_captured(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""脚本未调用 px.run() 应以退出码 1 退出。"""
script = tmp_path / "no_run.py"
script.write_text("print('just printing')\n", encoding="utf-8")
monkeypatch.setattr(sys, "argv", ["pxp", "--no-browser", str(script)])
with pytest.raises(SystemExit) as exc_info:
profiler.main()
assert exc_info.value.code == 1
def test_main_passes_script_args(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""应将脚本参数传递给目标脚本。"""
script = tmp_path / "argcheck.py"
script.write_text(
"import sys\n"
"assert sys.argv[1:] == ['myarg'], f'got {sys.argv[1:]}'\n"
"import pyflowx as px\n"
"px.run(px.Graph.from_specs([px.TaskSpec('a', lambda: 1)]), strategy='sequential')\n",
encoding="utf-8",
)
out_file = tmp_path / "report.html"
monkeypatch.setattr(sys, "argv", ["pxp", "--no-browser", "-o", str(out_file), str(script), "myarg"])
profiler.main() # 不抛异常即成功
def test_main_handles_script_exception(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""脚本抛异常时应捕获并继续生成报告（如果有 report）。"""
script = tmp_path / "raise.py"
script.write_text(
"import pyflowx as px\n"
"px.run(px.Graph.from_specs([px.TaskSpec('a', lambda: 1)]), strategy='sequential')\n"
"raise RuntimeError('after run')\n",
encoding="utf-8",
)
out_file = tmp_path / "report.html"
monkeypatch.setattr(sys, "argv", ["pxp", "--no-browser", "-o", str(out_file), str(script)])
profiler.main() # 不抛异常即成功
assert out_file.exists()
def test_main_auto_calls_main_when_no_main_block(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""脚本无 __main__ 块但定义了 main() 时应自动调用。"""
script = tmp_path / "no_main_block.py"
script.write_text(
"import pyflowx as px\n"
"def main():\n"
" px.run(px.Graph.from_specs([px.TaskSpec('a', lambda: 1)]), strategy='sequential')\n"
"# 无 if __name__ == '__main__' 块\n",
encoding="utf-8",
)
out_file = tmp_path / "report.html"
monkeypatch.setattr(sys, "argv", ["pxp", "--no-browser", "-o", str(out_file), str(script)])
profiler.main()
assert out_file.exists()
def test_main_auto_calls_main_with_clirunner(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""脚本无 __main__ 块但定义了调用 CliRunner 的 main() 时应自动调用。"""
script = tmp_path / "cli_tool.py"
script.write_text(
"import pyflowx as px\n"
"def main():\n"
" runner = px.CliRunner(\n"
" aliases={'t': px.TaskSpec('t', lambda: 1)},\n"
" )\n"
" runner.run_cli(['t'])\n",
encoding="utf-8",
)
out_file = tmp_path / "report.html"
monkeypatch.setattr(sys, "argv", ["pxp", "--no-browser", "-o", str(out_file), str(script), "t"])
profiler.main()
assert out_file.exists()
content = out_file.read_text(encoding="utf-8")
assert "PyFlowX 性能剖面报告" in content
def test_main_no_main_function_exits_with_1(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""脚本无 main() 且未调用 px.run() 应以退出码 1 退出。"""
script = tmp_path / "no_main.py"
script.write_text("x = 1\n", encoding="utf-8")
monkeypatch.setattr(sys, "argv", ["pxp", "--no-browser", str(script)])
with pytest.raises(SystemExit) as exc_info:
profiler.main()
assert exc_info.value.code == 1
class TestTryCallMain:
"""测试 _try_call_main。"""
def test_calls_main_when_present(self) -> None:
"""模块字典含 main 可调用对象时应调用它。"""
called: list[bool] = []
def fake_main() -> None:
called.append(True)
profiler._try_call_main({"main": fake_main})
assert called == [True]
def test_no_main_does_nothing(self) -> None:
"""模块字典不含 main 时不应报错。"""
profiler._try_call_main({}) # 不抛异常即成功
def test_non_callable_main_does_nothing(self) -> None:
"""main 不是可调用对象时不应报错。"""
profiler._try_call_main({"main": "not a function"}) # 不抛异常即成功
uv.lock
Generated
+1 -1
View File
@@ -5603,7 +5603,7 @@ pycountry = [
[[package]]
name = "pyflowx"
version = "0.2.11"
version = "0.2.12"
source = { editable = "." }
dependencies = [
{ name = "graphlib-backport", marker = "python_full_version < '3.9'" },