This commit is contained in:
2026-09-12 14:19:56 +08:00
commit 87ab0b794b
130 changed files with 30042 additions and 0 deletions

0
engine/tests/__init__.py Normal file
View File

27
engine/tests/fixtures/demo_sort.py vendored Normal file
View File

@@ -0,0 +1,27 @@
"""Demo script for users: contains several detectable anti-patterns.
- membership test against a list inside a loop -> membership_in_list
- nested loops -> nested_loops
- string concatenation with += inside a loop -> str_concat_loop
"""
def build_report(rows, allow):
out = ""
for r in rows:
if r in allow: # O(n) membership test on a list
out += str(r) + "," # string += accumulation
for other in rows: # nested loop -> O(n^2)
if r == other:
pass
return out
def main():
rows = list(range(400))
allow = list(range(0, 400, 2))
for _ in range(30):
build_report(rows, allow)
main()

16
engine/tests/fixtures/nested_calls.py vendored Normal file
View File

@@ -0,0 +1,16 @@
def leaf():
s = 0
for i in range(20000):
s += i
return s
def mid():
return sum(leaf() for _ in range(3))
def main():
return mid()
main()

19
engine/tests/fixtures/with_imports.py vendored Normal file
View File

@@ -0,0 +1,19 @@
"""scope=all 的测试 fixture —— 故意 import 一些标准库并实际用它们,
让 cProfile 能归因到非用户脚本里的帧。"""
import json
import time
def payload():
# 强制 json.loads / json.dumps 进 cProfile 栈
return json.loads(json.dumps({"a": 1, "b": [1, 2, 3]}))
def run():
for _ in range(50):
payload()
# 让 time.sleep / 调度器也进栈 —— 注意 sleep < cProfile tick 的话归不到
# 它自己头上,但 time 模块的辅助函数会被经过
run()

View File

@@ -0,0 +1,125 @@
"""跨语言契约测试engine/schema.py 的输出必须与 src/shared/analysis.ts 的类型对齐。
之前这个文件把 key 集合硬编码在 Python 里,从不读 analysis.ts —— 也就是说
TS 侧加一个字段、或者两边字段名写歧了,这个"契约测试"一律绿灯,
而 README 声称的"用契约测试保证一致"并不成立。
现在两边都是解析出来的Python 侧从 dataclass 的注解拿TS 侧从 interface 声明拿。
"""
import dataclasses
import io
import json
import os
import re
import subprocess
import sys
from engine import schema
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
FX = os.path.join(os.path.dirname(__file__), "fixtures")
ANALYSIS_TS = os.path.join(ROOT, "src", "shared", "analysis.ts")
def _ts_interface_fields(name):
"""从 analysis.ts 里抽出某个 interface 的字段名集合。
只处理"一行一个字段"的写法(本仓库的风格),够用且不需要引 TS parser。
可选字段的 `?` 去掉后再比较。
"""
src = io.open(ANALYSIS_TS, encoding="utf-8").read()
m = re.search(r"export interface %s\s*\{(.*?)\n\}" % re.escape(name), src, re.S)
assert m, "analysis.ts 里找不到 interface %s" % name
fields = set()
for line in m.group(1).splitlines():
line = line.strip()
if not line or line.startswith("//") or line.startswith("*"):
continue
fm = re.match(r"([A-Za-z_][A-Za-z0-9_]*)\??\s*:", line)
if fm:
fields.add(fm.group(1))
assert fields, "interface %s 解析出 0 个字段,解析器该修了" % name
return fields
def _py_dataclass_fields(cls):
return {f.name for f in dataclasses.fields(cls)}
def _run_engine(extra=(), env=None):
# 默认关掉 hide_internal契约测试要断言 keys / shapes,
# filtered 之后的 functions 列表可能为空IndexError 挂掉。
p = subprocess.run(
[sys.executable, "-m", "engine.runner",
"--script", os.path.join(FX, "nested_calls.py"),
"--no-hide-internal", *extra],
capture_output=True, cwd=ROOT, env=env,
)
return p
def _engine_result():
p = _run_engine()
return json.loads(p.stdout.decode("utf-8").strip().splitlines()[-1])
def test_top_level_keys_match_typescript():
data = _engine_result()
assert set(data.keys()) == _ts_interface_fields("AnalysisResult")
def test_top_level_keys_match_python_dataclass():
data = _engine_result()
assert set(data.keys()) == _py_dataclass_fields(schema.AnalysisResult)
def test_wall_time_keys_match_both_sides():
data = _engine_result()
keys = set(data["wallTime"].keys())
assert keys == _ts_interface_fields("WallTime")
assert keys == _py_dataclass_fields(schema.WallTime)
def test_function_keys_match_both_sides():
data = _engine_result()
keys = set(data["functions"][0].keys())
assert keys == _ts_interface_fields("FunctionNode")
assert keys == _py_dataclass_fields(schema.FunctionNode)
def test_output_is_pure_ascii():
"""输出必须是纯 ASCII。
反模式说明现在是历史的;保留 ensure_ascii=True 后输出应当逐字节 ASCII
跨 locale 一致。这个回归保护防止有人"为可读性"改回 ensure_ascii=False。
"""
env = {k: v for k, v in os.environ.items()
if k not in ("PYTHONUTF8", "PYTHONIOENCODING")}
p = _run_engine(env=env)
raw = p.stdout
bad = [b for b in raw if b > 127]
assert not bad, "引擎输出含 %d 个非 ASCII 字节" % len(bad)
raw.decode("ascii") # 不该抛
def test_out_flag_writes_result_to_file(tmp_path):
"""--out结果走独立文件stdout 留给用户代码。
锁的是那个回归 —— 结果曾和用户输出共用 stdout一句不带换行的
sys.stdout.write 就能让合法 Python 报"结果解析失败"
"""
out = os.path.join(str(tmp_path), "result.json")
script = os.path.join(str(tmp_path), "chatty.py")
io.open(script, "w", encoding="utf-8", newline="\n").write(
'import sys\ndef f():\n sys.stdout.write("done")\n return 1\nf()\n'
)
p = subprocess.run(
[sys.executable, "-m", "engine.runner", "--script", script, "--out", out],
capture_output=True, cwd=ROOT,
)
assert os.path.exists(out), "引擎没写出结果文件stderr=%s" % p.stderr[-400:]
data = json.loads(io.open(out, encoding="utf-8").read())
assert data["status"] == "ok"
# 用户的输出确实进了 stdout而且没污染结果
assert b"done" in p.stdout
assert b"schemaVersion" not in p.stdout

View File

@@ -0,0 +1,77 @@
import pytest
from engine.harness import calibrate_cprofile_overhead, collect_environment, load_source
def test_calibrate_returns_well_formed_calibration():
"""校准应该返回一个 well-formed Calibration:ratio > 0,两段耗时 > 0,workloadName 已知。
不强求 ratio > 1.0:在极快的机器上校准本身 < 5ms,会走 fallback ratio=1.0;
此时 instrumentedWorkloadSec 可能比 clean 略小(perf_counter 在 1ms 量级有 noise),
也不强求 instrumented >= clean。这条测试只锁住「返回结构对 + 数值非负」。
"""
calib = calibrate_cprofile_overhead()
assert calib.ratio > 0, f"ratio 必须正:{calib.ratio}"
assert calib.ratio <= 10.0, f"ratio 太大,可能校准失效:{calib.ratio}"
assert calib.cleanWorkloadSec > 0
assert calib.instrumentedWorkloadSec > 0
assert calib.workloadName == "tight-loop"
def test_calibrate_fallback_when_workload_too_fast(monkeypatch):
"""校准本身耗时 < 5ms 时应回退 ratio=1.0 —— 见 calibrate_cprofile_overhead 的边界处理。
直接 mock calibrate_cprofile_overhead 内部的 _exec_once_* 行为不可行(它们是模块内 helper),
改成 monkeypatch time.perf_counter 返回极短耗时:第一次返回 t0=0.0,第二次返回 t0+1e-6。
"""
from engine import harness
counter = {"n": 0}
base = {"t": 1000.0}
def fake_perf_counter():
counter["n"] += 1
# 第一次调用返回 0,之后每次 + 极小 delta,让 elapsed = ~3us (远小于 5ms)
idx = counter["n"]
return base["t"] + idx * 1e-6
monkeypatch.setattr(harness.time, "perf_counter", fake_perf_counter)
calib = harness.calibrate_cprofile_overhead()
assert calib.ratio == 1.0, f"短耗时下应回退 ratio=1.0,实际:{calib.ratio}"
def test_calibrate_does_not_leak_files(monkeypatch, tmp_path):
"""校准自己 scratch 一个临时文件,跑完应清干净 —— 不污染系统临时目录。"""
# 让 mkstemp 落到 tmp_path 方便断言
import tempfile
monkeypatch.setattr(tempfile, "gettempdir", lambda: str(tmp_path))
before = set(tmp_path.iterdir())
calibrate_cprofile_overhead()
after = set(tmp_path.iterdir())
# 不应有残留 —— 之前 / 之后 iterdir 集合应相等
assert after == before, f"残留文件: {after - before}"
def test_collect_environment():
env = collect_environment()
assert env.python.count(".") >= 2
assert env.timerResolution > 0
def test_load_source_strips_bom(tmp_path):
"""load_source 必须剥掉 BOM —— 引擎内所有读源码的地方都依赖这一点。"""
script = tmp_path / "bom.py"
script.write_text("x = 1\n", encoding="utf-8-sig")
assert script.read_bytes().startswith(b"\xef\xbb\xbf")
src = load_source(str(script))
assert not src.startswith("")
# 剥离后必须能直接过 compile(),这正是 runner 语法预检做的事
compile(src, str(script), "exec")
def test_load_source_raises_on_binary(tmp_path):
"""二进制文件抛 UnicodeDecodeError而不是 OSError—— runner 依赖这个类型来归类错误。"""
script = tmp_path / "bin.py"
script.write_bytes(b"\xff\xfe\x00\x01\x80\x81")
with pytest.raises(UnicodeDecodeError):
load_source(str(script))

158
engine/tests/test_runner.py Normal file
View File

@@ -0,0 +1,158 @@
import json
import os
import subprocess
import sys
FX = os.path.join(os.path.dirname(__file__), "fixtures")
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
def _run(script, *extra):
# 默认关掉 hide_internal这些测试是结构 / 状态 / 形状的契约测试,
# 要看到 fixture 里所有 frame 才能断言。hide_internal 是 UI 默认开的产品行为,
# 不是引擎协议的一部分 —— 测协议就别测产品默认。
p = subprocess.run(
[sys.executable, "-m", "engine.runner", "--script", script,
"--no-hide-internal", *extra],
capture_output=True, text=True, cwd=ROOT,
)
assert p.stdout.strip(), f"no stdout; stderr={p.stderr}"
return json.loads(p.stdout.strip().splitlines()[-1])
def test_ok_run():
data = _run(os.path.join(FX, "nested_calls.py"))
assert data["status"] == "ok"
assert data["wallTime"]["seconds"] > 0
assert data["wallTime"]["unit"] == "s"
assert len(data["functions"]) >= 3
def test_syntax_error(tmp_path):
bad = tmp_path / "syntax_error.py"
bad.write_text("def f(:\n pass\n", encoding="utf-8")
data = _run(str(bad))
assert data["status"] == "syntax_error"
assert data["error"]["type"]
def test_runtime_error(tmp_path):
bad = tmp_path / "runtime_error.py"
bad.write_text("raise ValueError('boom')\n", encoding="utf-8")
data = _run(str(bad))
assert data["status"] == "runtime_error"
assert "boom" in data["error"]["message"]
def test_sys_exit_becomes_runtime_error(tmp_path):
"""sys.exit() 抛 SystemExitBaseException 子类),之前 except Exception 抓不到,
子进程裸退出无 JSON → UI 显示 "引擎无输出"。现在应被显式捕获并归为 runtime_error。"""
bad = tmp_path / "exit.py"
bad.write_text("import sys\nsys.exit(7)\n", encoding="utf-8")
data = _run(str(bad))
assert data["status"] == "runtime_error"
assert "SystemExit" in data["error"]["type"]
assert "7" in data["error"]["message"]
def test_io_error_returns_runtime_error(tmp_path):
"""脚本文件不存在 / 不可读时,不要让子进程裸崩 — 应该回 runtime_error 给出有用信息。"""
missing = tmp_path / "does_not_exist.py"
data = _run(str(missing))
assert data["status"] == "runtime_error"
assert "FileNotFoundError" in data["error"]["type"] or "无法读取" in data["error"]["message"]
def test_bom_script_runs_ok(tmp_path):
"""带 BOM 的 UTF-8 脚本Windows 记事本默认保存格式)必须能正常跑。
回归测试runner 曾用 encoding="utf-8" 读源码BOM 的  前缀残留下来,
语法预检的 compile() 直接抛 SyntaxError: invalid character用户看到的是
"语法错误"而不是正常结果 —— 而 harness.py 早就用 utf-8-sig 处理对了,
两处不一致导致这个 bug 只在预检阶段出现。
"""
script = tmp_path / "bom.py"
script.write_text("x = sum(range(1000))\n", encoding="utf-8-sig")
# 确认 fixture 真的带 BOM否则这个测试会静默失效
assert script.read_bytes().startswith(b"\xef\xbb\xbf")
data = _run(str(script))
assert data["status"] == "ok", f"BOM 脚本被误判: {data.get('error')}"
def test_binary_script_returns_runtime_error(tmp_path):
"""非 UTF-8 / 二进制文件应回结构化 runtime_error而不是让引擎裸崩。
回归测试UnicodeDecodeError 不是 OSError 的子类,之前只 except OSError
的那层保护抓不到它,子进程崩溃且无 JSON 输出 → UI 显示"引擎无输出"
"""
script = tmp_path / "binary.py"
script.write_bytes(b"\x00\x01\x02\xff\xfe\xfd binary garbage \x80\x81")
data = _run(str(script))
assert data["status"] in ("runtime_error", "syntax_error")
assert data["error"]["type"]
def test_scope_default_is_user():
"""默认 --scope=user —— 不给 scope 参数时只看到用户脚本里的函数。"""
script = os.path.join(FX, "with_imports.py")
data = _run(script)
assert data["status"] == "ok"
files = {f["file"] for f in data["functions"]}
# user 模式:不应出现 json 模块的内部帧(虽然 fixture 里 import 了 json
json_internal = {f for f in files if "json" in f and "decoder" in f or "json" in f and "encoder" in f}
assert not json_internal, f"默认 scope 应为 user但出现了 json 内部帧: {json_internal}"
def test_scope_all_includes_library_frames():
"""--scope=all让耗时可以下钻到 import 的包里 —— json 模块的函数也应进入结果。"""
script = os.path.join(FX, "with_imports.py")
data = _run(script, "--scope", "all")
assert data["status"] == "ok"
files = {f["file"] for f in data["functions"]}
# scope=all 时至少有一个 json 相关的文件路径进来 —— 说明 import 的包被归因了
has_json = any("json" in f for f in files)
assert has_json, f"scope=all 应包含 json 模块帧,实际 files: {files}"
def test_user_side_effects_run_once(tmp_path):
"""v4 单跑架构核心回归:用户脚本里的副作用只执行一次。
v3 时期 runner 跑两次(裸跑测 wall_time + cProfile 跑),
用户脚本里 plot()、print()、写文件等副作用也跟着跑两次 —— 经典"plot 弹两个窗口"
bug 的根因。v4 改成单次 exec(code) under cProfile + 校准系数折算 wall-time 后,
副作用应只发生 1 次。
检测手段:让用户脚本 append 一行到文件exec 一次就一行,跑两次就两行。
读 marker 文件的行数即可判断 exec 次数。
"""
marker = tmp_path / "marker.txt"
script = tmp_path / "s.py"
# 用 raw 字符串避 Windows 反斜杠麻烦append 模式保证两次跑不会被覆盖
script.write_text(
"with open(r'" + str(marker) + "', 'a', encoding='utf-8') as f:\n"
" f.write('X\\n')\n",
encoding="utf-8",
)
# 直接 spawn runner 子进程(不走 _run helper —— 那个 helper 是从 stdout 解析 JSON
# 的,我们这次主要观察 marker 文件)
out_path = tmp_path / "result.json"
p = subprocess.run(
[sys.executable, "-m", "engine.runner",
"--script", str(script), "--out", str(out_path)],
capture_output=True, text=True, cwd=ROOT,
)
assert p.returncode == 0, f"runner 失败:{p.stderr}"
assert marker.exists(), "用户脚本副作用根本没执行"
lines = marker.read_text(encoding="utf-8").splitlines()
assert len(lines) == 1, (
f"副作用被多次执行(找到 {len(lines)}v4 单跑架构回归:{lines}"
)
# 同时验证 result.json 里 calibration 字段在 + wallTime 被校准
result = json.loads(out_path.read_text(encoding="utf-8"))
assert result["status"] == "ok"
assert "calibration" in result and result["calibration"] is not None
assert result["calibration"]["ratio"] >= 1.0
assert result["wallTime"]["seconds"] > 0

View File

@@ -0,0 +1,21 @@
import json
from engine.schema import AnalysisResult, Environment, SCHEMA_VERSION, WallTime
def test_analysis_result_serializes_to_json():
r = AnalysisResult(
schemaVersion=SCHEMA_VERSION,
environment=Environment(python="3.9.19", platform="win32", processor="x86", timerResolution=1e-7),
config={},
status="ok",
error=None,
wallTime=WallTime(seconds=0.105, unit="s"),
functions=[],
flame=None,
)
data = json.loads(r.to_json())
assert data["schemaVersion"] == SCHEMA_VERSION
assert data["status"] == "ok"
assert data["wallTime"]["seconds"] == 0.105
assert data["environment"]["python"] == "3.9.19"

View File

@@ -0,0 +1,761 @@
import os
import pytest
from engine.schema import FunctionNode
from engine.structure import (
_build_flame,
_classify_origin,
_make_frame_filter,
_norm,
_top_module,
profile_structure,
)
def test_structure_finds_functions():
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
res = profile_structure(fx, hide_internal=False)
names = {f.name for f in res.functions}
assert {"leaf", "mid", "main"} <= names
leaf = next(f for f in res.functions if f.name == "leaf")
assert leaf.tottime > 0 and leaf.ncalls >= 3
assert res.flame.value >= leaf.tottime
def test_structure_default_scope_is_user():
"""默认 scope=user —— 只看到用户脚本里的函数(向后兼容)。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
res = profile_structure(fx, hide_internal=False)
files = {f.file for f in res.functions}
# 只有 fixture 自身;不应出现标准库路径(<frozen ...> / <string> / site-packages 等)
non_user = {f for f in files if f not in (fx, "~") and not f.endswith("nested_calls.py")}
assert not non_user, f"user 模式下出现非用户脚本帧: {non_user}"
def test_structure_scope_all_includes_stdlib():
"""scope=all耗时可以下钻到 import 的包里,标准库 / 第三方包里的函数也进来。
注:`~` 在这里可能出现在 files 里 —— 它不再是"cProfile 内部"的同义词,
而是 cProfile 给所有 C 扩展函数json C accelerator / numpy C 核 / time.sleep
等)打的文件名。这些是真实工作,scope=all 必须保留。cProfile 内部帧靠 name
识别("_lsprof.Profiler" / "Profiler"),由 _make_frame_filter 剔除。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "with_imports.py")
res = profile_structure(fx, scope="all", hide_internal=False)
files = {f.file for f in res.functions}
# fixture 里会 import json 和 time —— 至少其中一个 stdlib 路径应被纳入
has_stdlib = any("json" in f or "time" in f or f == "<frozen importlib>" for f in files)
assert has_stdlib, f"scope=all 应包含标准库帧,实际 files: {files}"
# cProfile 内部帧name 含 _lsprof.Profiler 子串)必须剔除 —— 但其它
# `~` 帧time.sleep / json C 加速器 / builtins.exec 等真实 C 扩展)必须保留。
cprofile_internal = [f for f in res.functions if "_lsprof.Profiler" in f.name]
assert not cprofile_internal, (
f"cProfile 内部帧不应漏进 functions{[(f.name, f.file) for f in cprofile_internal]}"
)
def test_norm_normalizes_case_and_slashes():
"""Windows (NTFS) 上同一文件可能被报成 `C:/Foo/Bar.py` 或 `c:\\foo\\BAR.py`
取决于谁产生的字符串。_norm 必须把两边都规范成同一种形式才能正确比
对脚本路径。"""
raw = "C:/Foo/Bar.py"
expected = os.path.normcase(os.path.abspath(raw))
assert _norm(raw) == expected
def test_user_frame_filter_is_case_and_slash_agnostic():
"""_make_frame_filter 对 frame.file 也要走 _norm —— 跨平台/跨斜杠风格
仍然能命中用户脚本。这条用例专门守回归:之前只用同一字符串匹配,
在 Windows 上大小写或斜杠不一致就会漏掉所有函数。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
is_kept = _make_frame_filter(fx)
# 把 frame.file 里的正反斜杠互换 —— 如果原始路径里有 '\\'
swapped = fx.replace("\\", "/") if "\\" in fx else fx.replace("/", "\\")
if swapped != fx:
assert is_kept((swapped, 1, "main")) is True
def test_frame_filter_excludes_cprofile_internal():
"""cProfile 内部帧_lsprof.Profiler 类成员file='~' 且 name 含 Profiler
永远被剔除 —— 不论 scope=user 还是 scope=all。
关键:「`~` == cProfile 内部」是错的 —— cProfile 用 `~` 表示所有 C 扩展函数
time.sleep / json C accelerator / numpy C 核这些必须留下。cProfile
自己靠 name 区分:'_lsprof.Profiler' / 'Profiler' / '<method ... of
'_lsprof.Profiler' objects>' 等。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
for scope in ("user", "all"):
is_kept = _make_frame_filter(fx, scope)
# 真实 cProfile 内部 name —— 必须被过滤
assert is_kept(("~", 0, "_lsprof.Profiler")) is False
assert is_kept(("~", 0, "Profiler")) is False
assert is_kept(("~", 0, "<method 'enable' of '_lsprof.Profiler' objects>")) is False
def test_frame_filter_keeps_c_extension_frames():
"""scope=allC 扩展函数帧file='~' + name 是真实 builtin必须留下 —
否则 numpy C 核 / json C 加速器 / time.sleep 等会从结果里消失,
scope=all 失去下钻意义。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
is_kept = _make_frame_filter(fx, "all")
# C 扩展(不是 cProfile 内部)—— 必须保留
assert is_kept(("~", 0, "<built-in method sleep of 'time' objects>")) is True
assert is_kept(("~", 0, "_default_encoder")) is True # json C accelerator
# scope=userC 帧不会进入用户脚本 —— 也过滤掉(与历史行为一致)
is_kept_user = _make_frame_filter(fx, "user")
assert is_kept_user(("~", 0, "<built-in method sleep of 'time' objects>")) is False
def test_frame_filter_scope_all_keeps_any_non_internal_frame():
"""scope=all只要不是 cProfile 内部(~)就保留 —— 用户脚本、stdlib、第三方都进。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
is_kept = _make_frame_filter(fx, "all")
# 任意"非 ~"路径都应通过
assert is_kept(("C:/Python39/Lib/json/decoder.py", 100, "decode")) is True
assert is_kept(("/some/site-packages/pandas/core/frame.py", 1, "from_records")) is True
def test_top_module_user_script():
"""用户脚本(==user_script_path"<user>""""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
assert _top_module(fx, fx) == "<user>"
def test_top_module_user_script_case_and_slash_agnostic():
"""_top_module 比较 user_script_path 时走 _norm —— Windows 大小写 / 斜杠差异不影响。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
swapped = fx.replace("\\", "/") if "\\" in fx else fx.replace("/", "\\")
if swapped != fx:
assert _top_module(swapped, fx) == "<user>"
def test_top_module_stdlib_windows():
"""Windows 标准库C:\\Python39\\Lib\\json\\decoder.py → json"""
assert _top_module(r"C:\Python39\Lib\json\decoder.py", "<user>") == "json"
assert _top_module(r"C:\Python39\Lib\json\__init__.py", "<user>") == "json"
# 多层嵌套Lib/site-packages 这种 dev 布局也行
assert _top_module(r"C:\Python39\Lib\site-packages\foo\bar.py", "<user>") == "foo"
def test_top_module_top_level_py_strips_extension():
"""顶层 .py 文件必须剥掉扩展名 —— `Lib\\functools.py` 是模块 functools 而不是 `functools.py`。
之前 bug直接返回 `parts[idx]`(即 `functools.py`UI 分组里看到一坨文件名而不是模块名。"""
assert _top_module(r"C:\Python39\Lib\functools.py", "<user>") == "functools"
# Linux 顶层 stdlib
assert _top_module("/usr/lib/python3.11/functools.py", "<user>") == "functools"
# __init__ 不剥(不是 .py 后缀)—— 顶层 `Lib\json\__init__.py` 走 _strip_py 时返回 "json"
# (这里 _strip_py 不会动 __init__但 parts[idx] 已经是 "json\__init__.py" 这种,
# 实际代码路径是先取 parts[idx] 再 _strip_py所以 "json\__init__.py" → "json\__init__"
# 这里的关键是 module 名不是 filename-with-py-extension
result = _top_module(r"C:\Python39\Lib\functools.py", "<user>")
assert not result.endswith(".py"), f"module 名不能带 .py 后缀,实际 {result!r}"
def test_top_module_stdlib_linux():
"""Linux/macOS 标准库:/usr/lib/python3.X/<pkg>/... → <pkg>"""
assert _top_module("/usr/lib/python3.11/json/decoder.py", "<user>") == "json"
assert _top_module(
"/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/json/decoder.py",
"<user>",
) == "json"
def test_top_module_site_packages():
"""site-packages / dist-packages标记后的第一个目录段就是包名。"""
assert _top_module(
"/usr/lib/python3.11/site-packages/numpy/core/array.py", "<user>"
) == "numpy"
assert _top_module(
"C:\\Python39\\Lib\\site-packages\\pandas\\core\\frame.py", "<user>"
) == "pandas"
# dist-packagesDebian 系)
assert _top_module(
"/usr/lib/python3.11/dist-packages/requests/api.py", "<user>"
) == "requests"
def test_top_module_frozen():
"""frozen / built-in / <string> 虚拟帧 → 整段保留作为标签。"""
assert _top_module("<frozen importlib._bootstrap>", "<user>") == "<frozen>"
assert _top_module("<built-in method exec>", "<user>") == "<built-in>"
assert _top_module("<string>", "<user>") == "<string>"
def test_top_module_c_extension():
"""`~` 在 cProfile 里代表所有 C 扩展函数numpy C 核 / json 加速器 /
time.sleep 等)。这些 frame 现在 scope=all 会留下(不再被一刀切),需要给个
有意义的 module 标签 —— 用 "<built-in>" 跟 _classify_origin 的 builtin
字段保持一致。"""
assert _top_module("~", "<user>") == "<built-in>"
# user_script_norm 也要匹配上 user 脚本 —— 防御性
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
assert _top_module("~", fx, _norm(fx)) == "<built-in>"
def test_top_module_fallback_dirname():
"""未匹配任何标记 → 取倒数第二段目录。"""
# 例如:".../someproj/src/utils/helper.py" → "utils"
assert _top_module("/path/to/someproj/src/utils/helper.py", "<user>") == "utils"
def test_profile_structure_populates_module():
"""profile_structure 给每条 fn 算 modulefixture 是用户脚本,全部 <user>。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
res = profile_structure(fx, hide_internal=False)
assert res.functions
modules = {f.module for f in res.functions}
assert modules == {"<user>"}, f"默认 scope 下应只有 <user>,实际 {modules}"
def test_profile_structure_scope_all_populates_real_modules():
"""scope=allfn.module 应来自真实的 stdlib / 第三方包。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "with_imports.py")
res = profile_structure(fx, scope="all", hide_internal=False)
modules = {f.module for f in res.functions}
# fixture 里 import 了 json / time至少其中一个 stdlib 应出现
assert "<user>" in modules, f"用户脚本应在模块集合里,实际 {modules}"
assert any(m in ("json", "time", "<frozen>", "<built-in>") for m in modules), (
f"scope=all 应有 stdlib 模块,实际 {modules}"
)
def _mk_fn(name: str, module: str, tottime: float) -> FunctionNode:
"""测试用 FunctionNode 工厂 —— 关注 module / tottime其他字段填占位。"""
return FunctionNode(
id=f"f.py:1:{name}",
file="f.py",
line=1,
name=name,
cumtime=tottime,
tottime=tottime,
ncalls=1,
percallTot=tottime,
module=module,
)
def test_build_flame_single_module_stays_flat():
"""单模块时scope=user 默认情况)保持扁平 —— 同名兄弟不会被 module
节点挤占,截断阈值不变,向后兼容。"""
fns = [
_mk_fn("main", "<user>", 0.5),
_mk_fn("mid", "<user>", 0.3),
_mk_fn("leaf", "<user>", 0.2),
]
flame = _build_flame(fns, total=1.0)
assert flame.name == "root"
assert flame.value == 1.0
# 扁平root.children 是函数名(不是 module
assert [c.name for c in flame.children] == ["main", "mid", "leaf"]
# 每个函数 tile 没有 children不会再下钻一层
for child in flame.children:
assert child.children == []
def test_build_flame_multi_module_groups_by_module():
"""多模块scope=all 且 import 多个包root.children 是 module 节点,
每个 module 节点的 children 是该模块下的函数。"""
fns = [
_mk_fn("main", "<user>", 0.5),
_mk_fn("decode", "json", 0.3),
_mk_fn("loads", "json", 0.2),
_mk_fn("sleep", "time", 0.1),
]
flame = _build_flame(fns, total=1.1)
assert flame.name == "root"
# module 节点按总 tottime 降序:<user>(0.5) > json(0.5) > time(0.1)
# <user> 和 json 都是 0.5,排序稳定时顺序取决于字典遍历顺序 —— 这里不强制
# 顺序,只验证排序结果一致(用 sorted() 拿到一组)
module_nodes = flame.children
assert len(module_nodes) == 3
# 第一个必须是最大的 <user>0.5 严格大于 json 的 0.5,因为总和 tie-break
# 由 sorted 的 stable 行为兜底:<user> 在 fns 里排前面 → 同值时排前)
assert module_nodes[0].name == "<user>"
assert module_nodes[0].value == pytest.approx(0.5)
# json 模块 value = 0.3 + 0.2 = 0.5
json_node = next(c for c in module_nodes if c.name == "json")
assert json_node.value == pytest.approx(0.5)
assert sorted(c.name for c in json_node.children) == ["decode", "loads"]
# time 模块只有 sleep
time_node = next(c for c in module_nodes if c.name == "time")
assert time_node.value == pytest.approx(0.1)
assert [c.name for c in time_node.children] == ["sleep"]
def test_build_flame_empty_returns_empty_root():
"""没函数时不爆 —— root.value=0、children=空列表。"""
flame = _build_flame([], total=0.0)
assert flame.name == "root"
assert flame.value == 0.0
assert flame.children == []
# --- origin 分类v3 新增) -----------------------------------------------------
def test_classify_origin_user_script_matches_norm():
"""用户脚本路径_norm 后)与 user_script_path 一致 → user。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
assert _classify_origin(fx, fx, "<user>") == "user"
swapped = fx.replace("\\", "/") if "\\" in fx else fx.replace("/", "\\")
if swapped != fx:
assert _classify_origin(swapped, fx, "<user>") == "user"
def test_classify_origin_frozen_and_builtin():
"""frozen / built-in 虚拟帧 → frozen / builtin。"""
assert _classify_origin("<frozen importlib._bootstrap>", "/x.py", "<frozen>") == "frozen"
assert _classify_origin("<built-in method exec>", "/x.py", "<built-in>") == "builtin"
# 其它 <...> 兜底 other
assert _classify_origin("<string>", "/x.py", "<string>") == "other"
def test_classify_origin_stdlib_via_sys_stdlib_module_names():
"""module_name 在 sys.stdlib_module_names 里 → stdlib权威路径"""
# json / os / re / sys / pathlib 都肯定在 stdlib 里
assert _classify_origin("C:/Python311/Lib/json/decoder.py", "/x.py", "json") == "stdlib"
assert _classify_origin("/usr/lib/python3.11/pathlib/__init__.py", "/x.py", "pathlib") == "stdlib"
def test_classify_origin_third_party_site_packages():
"""site-packages / dist-packages 下的包 → third_party即使包名可能撞车"""
assert _classify_origin(
"/usr/lib/python3.11/site-packages/numpy/core/array.py", "/x.py", "numpy"
) == "third_party"
assert _classify_origin(
"C:\\Python39\\Lib\\site-packages\\pandas\\core\\frame.py", "/x.py", "pandas"
) == "third_party"
# dist-packagesDebian 系)
assert _classify_origin(
"/usr/lib/python3.11/dist-packages/requests/api.py", "/x.py", "requests"
) == "third_party"
def test_classify_origin_stdlib_path_heuristic_fallback():
"""sys.stdlib_module_names 漏判时,只要路径真的在 sysconfig 给的 stdlib 根
下,就该归为 stdlib —— 即使模块名未知。
之前 bugM5:只看路径里含 "/Lib/" 子串就当 stdlib,结果用户项目里
/home/x/myproject/Lib/foo.py 被误认成 stdlib。现在锚到 sysconfig 的真根。
"""
import sysconfig
real_stdlib = sysconfig.get_paths().get("stdlib") or os.path.join(os.sep, "Lib")
# 构造一个 fake 包路径,放在真实 stdlib 根下面 —— 必须真在那个目录里
fake_pkg = os.path.join(real_stdlib, "_weirdstdlib_pkg_for_test")
# 文件不存在也没事 —— _is_under_real_stdlib 只比对 abspath,不要求文件真存在
assert _classify_origin(fake_pkg, "/x.py", "_weirdstdlib_pkg_for_test") == "stdlib"
# 反向用例:用户在 /home/x/myproject/Lib/foo.py 下,即使路径含 "/Lib/" 也不该
# 误判成 stdlib —— 必须真的在 sysconfig 那个根下面
assert _classify_origin(
"/home/user/myproject/Lib/foo.py", "/x.py", "foo"
) == "other"
assert _classify_origin(
"/home/user/myproject/lib/python3.11/foo.py", "/x.py", "foo"
) == "other"
def test_classify_origin_unknown_path_is_other():
"""未匹配的奇怪路径 → other。"""
assert _classify_origin(
"/path/to/someproj/src/utils/helper.py", "/x.py", "utils"
) == "other"
def test_profile_structure_populates_origin():
"""profile_structure 给每条 fn 算 originfixture 是用户脚本,全部 user。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
res = profile_structure(fx, hide_internal=False)
origins = {f.origin for f in res.functions}
assert origins == {"user"}, f"默认 scope 下 origin 应全是 user实际 {origins}"
def test_profile_structure_scope_all_classifies_by_origin():
"""scope=all用户脚本帧归 userimport 的 json 帧归 stdlib —— 至少这两个 origin 都有。"""
fx = os.path.join(os.path.dirname(__file__), "fixtures", "with_imports.py")
res = profile_structure(fx, scope="all", hide_internal=False)
origins = {f.origin for f in res.functions}
# 用户脚本帧必有
assert "user" in origins
# fixture 显式 import json → 必有 stdlib 帧
assert "stdlib" in origins, f"json import 后应有 stdlib 帧,实际 origins: {origins}"
# 没 origin 字段缺失 / 其它 origin 兜底值(如 cprofile混进来的情况
assert "cprofile" not in origins, f"cProfile 内部帧不应漏进 origin实际 {origins}"
# 所有 origin 必须是已知值之一
KNOWN = {"user", "stdlib", "third_party", "builtin", "frozen", "other"}
assert origins <= KNOWN, f"未知 origin 混进来:{origins - KNOWN}"
# ──────────────────────────────────────────────────────────────────────────────
# v5 噪声过滤 —— 测试代码 + tottime=0 默认剔除
# ──────────────────────────────────────────────────────────────────────────────
#
# 用户场景:scope=all 时 cProfile 报告里夹带大量测试代码(numpy.tests.test_xxx /
# pytest fixture / unittest runner 等)和 tottime=0 的内部调用展开(typing / inspect /
# functools 等)。这些不是优化目标,默认应该剔除,result.json 一开始就是干净的。
# 用户原话:"软件内部的测试部分默认百分百过滤掉,不在统计范围内容"。
from engine.structure import _is_internal_test
def test_is_internal_test_recognizes_pytest_tests_dir():
# /tests/ /test/ /__tests__/ 目录下的函数 —— pytest 标准约定
assert _is_internal_test("helper", "/p/foo/tests/x.py", "foo.tests")
assert _is_internal_test("helper", "/p/foo/test/x.py", "foo.test")
assert _is_internal_test("helper", "/p/foo/__tests__/x.py", "foo.__tests__")
def test_is_internal_test_recognizes_pytest_file_naming():
# test_xxx.py / xxx_test.py —— pytest 文件命名约定
assert _is_internal_test("f", "/p/test_foo.py", "foo")
assert _is_internal_test("f", "/p/foo_test.py", "foo")
assert _is_internal_test("f", "/p/tests/test_bar.py", "foo")
def test_is_internal_test_recognizes_tests_module_path():
# 模块路径就是 tests / test / __tests__,或它们的子模块
assert _is_internal_test("f", "p/x.py", "tests")
assert _is_internal_test("f", "p/x.py", "test")
assert _is_internal_test("f", "p/x.py", "__tests__")
assert _is_internal_test("f", "p/x.py", "tests.foo")
assert _is_internal_test("f", "p/x.py", "test.bar")
assert _is_internal_test("f", "p/x.py", "__tests__.baz")
def test_is_internal_test_hardcoded_engine_whitelist():
# _pyrof_calib —— 引擎内部白名单,任何情况下都算内部
assert _is_internal_test("_pyrof_calib", "/p/anything.py", "<user>")
assert _is_internal_test("_pyrof_calib", "/p/", "weird_module")
def test_is_internal_test_does_not_match_function_names():
# 关键:只看语境不看函数名 —— 函数名匹配太容易误伤
assert not _is_internal_test("test_helper", "<user>", "a.py")
assert not _is_internal_test("test_visualization", "<user>", "/p/app/main.py")
assert not _is_internal_test("TestFoo.test_foo", "<user>", "/p/myapp/svc.py")
assert not _is_internal_test("MyClass.test_count", "<user>", "a.py")
assert not _is_internal_test("UserService.test_login", "<user>", "a.py")
def test_is_internal_test_does_not_match_test_substring_in_module():
# testing / dataset / contest 等含 test 子串的合法模块名
assert not _is_internal_test("f", "p/x.py", "testing")
assert not _is_internal_test("f", "p/x.py", "dataset")
assert not _is_internal_test("f", "p/x.py", "contest")
def test_is_internal_test_does_not_match_standalone_test_py():
# test.py / tests.py —— 项目入口常这么命名
assert not _is_internal_test("main", "<user>", "/p/test.py")
assert not _is_internal_test("main", "<user>", "/p/tests.py")
def test_is_internal_test_does_not_match_tests_subdir_non_test_files():
"""回归测试:tests/ 子目录下的非测试文件不应当成测试代码剔除。
之前用宽匹配 `[\\/](tests?|__tests__)[\\/]` —— 任何含 `/tests/` 的路径都中招,
包括 `tests/fixtures/x.py` 这种测试数据夹具。结果用户脚本只要住在带 tests/
子目录的路径下,functions 表直接被过滤成 0 行。
收紧到「tests/ 后面紧接文件名,不能再有 / 子目录」后:
- tests/foo.py → 算测试代码 (foo.py 直接在 tests/ 下)
- tests/fixtures/foo.py → 不算 (fixtures 是子目录,foo.py 不挂在 tests/ 下)
- tests/data/sample.py → 不算 (data 是子目录)
- tests/__init__.py / conftest.py → 仍走模块兜底命中 (module == "tests")
"""
# 子目录下的非测试文件 —— 不应被过滤
assert not _is_internal_test("f", "/p/tests/fixtures/x.py", "fixtures")
assert not _is_internal_test("f", "/p/foo/tests/data/sample.py", "data")
assert not _is_internal_test("f", "/p/mine/tests/integration/helper.py", "integration")
# tests/ 直接下面的文件 —— 仍走 dir 兜底命中
assert _is_internal_test("f", "/p/tests/foo.py", "tests")
def test_profile_structure_default_filters_zero_time_and_test_code():
# scope=all + 默认过滤:测试代码 + tottime=0 不进 functions[]
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
res = profile_structure(fx, scope="all")
assert all(f.tottime > 0 for f in res.functions), (
f"默认过滤下不应有 tottime=0 的帧: {[f.name for f in res.functions if f.tottime == 0]}"
)
for f in res.functions:
assert not _is_internal_test(f.name, f.file, f.module), (
f"默认过滤下不应有测试代码帧: {f.name} ({f.module})"
)
def test_profile_structure_hide_internal_false_keeps_more():
# hide_internal=False 时不过滤 —— result 含全量数据
fx = os.path.join(os.path.dirname(__file__), "fixtures", "nested_calls.py")
res_full = profile_structure(fx, scope="all", hide_internal=False)
res_filtered = profile_structure(fx, scope="all", hide_internal=True)
assert len(res_full.functions) >= len(res_filtered.functions)
# ──────────────────────────────────────────────────────────────────────────────
# v6 stdlib 间接调用链过滤 —— typing/inspect/functools 等默认剔除
# ──────────────────────────────────────────────────────────────────────────────
#
# 用户场景scope=all 时 numpy 这类第三方包会通过类型注解 / importlib 钩子触发
# typing / inspect / functools / re / _py_warnings / ctypes / enum 等几十个 stdlib
# 内部帧。这些帧 tottime 通常在 1µs~1ms 之间(高于 cProfile 量化地板但低于噪声线),
# 既不是优化目标也不是用户写的代码。默认剔除,只保留用户脚本里显式 import 的
# stdlib 模块帧 —— 「用户能看到『自己 import 了什么』,而不是『numpy 的依赖图』」。
from engine.structure import _extract_user_imports
def test_extract_user_imports_simple_import():
src = "import json\nimport os\nx = 1\n"
assert _extract_user_imports(src) == {"json", "os"}
def test_extract_user_imports_from_import():
src = "from collections import OrderedDict\nfrom os.path import join\nx = 1\n"
# 顶级包名进集合,不展开子模块
assert _extract_user_imports(src) == {"collections", "os"}
def test_extract_user_imports_dotted():
src = "import numpy as np\nimport a.b.c\nx = 1\n"
# 取顶级包名,as 别名不进集合
assert _extract_user_imports(src) == {"numpy", "a"}
def test_extract_user_imports_inside_function_ignored():
"""函数/类内部的 import 不算 —— 它们是延迟副作用,不是「我需要分析的目标」。"""
src = "import json\ndef f():\n import typing\n return typing\n"
assert _extract_user_imports(src) == {"json"}
def test_extract_user_imports_relative_skipped():
"""from . import x 是相对导入 —— 依附于已知包,跳过。"""
src = "from . import utils\nimport json\n"
assert _extract_user_imports(src) == {"json"}
def test_extract_user_imports_syntax_error_returns_empty():
# 语法错误的源码(走不到 profile_and_measure,这里只是防御)
assert _extract_user_imports("def f(:\n pass") == set()
def test_profile_structure_default_filters_indirect_stdlib(tmp_path):
"""scope=all 默认应剔除间接 stdlib 调用链 —— 用户显式 import 的保留,其他剔除。
fixture 设计:用户只 import json + os.path + collections,显式调用 list() +
json.dumps + os.path.join。间接触发 typing/inspect/functools/_py_warnings 的
帧(来自 cProfile 内部 / 解释器启动 / json 自身依赖)不应出现在 result 里。
"""
script = tmp_path / "user.py"
script.write_text(
"import json\n"
"import os.path\n"
"from collections import OrderedDict\n"
"x = json.dumps(OrderedDict([('a', 1)]))\n"
"y = os.path.join('a', 'b')\n"
"z = list(range(10))\n",
encoding="utf-8",
)
res = profile_structure(str(script), scope="all")
modules = {f.module for f in res.functions}
# 用户显式 import 的 stdlib 模块保留
assert "json" in modules, f"json 应保留(用户显式 import实际 {modules}"
# 间接触发的 stdlib 内部模块剔除
leak = modules & {"typing", "inspect", "functools", "_py_warnings", "re",
"annotationlib", "ctypes", "enum", "warnings"}
assert not leak, f"默认应剔除间接 stdlib 调用链,但 {leak} 漏进来了"
def test_profile_structure_hide_internal_false_keeps_indirect_stdlib(tmp_path):
"""--no-hide-internal间接 stdlib 帧也保留,排查调用栈时用得到。
fixture 用 traceback.print_exc() —— 它在 CPython 3.11+ 实现里会触发 inspect /
annotationlib / linecache / tokenize / dataclasses / enum / reprlib / textwrap /
ast / codeop / contextlib 等十几 stdlib 内部模块(都是 Python 层面的,而非 C)。
fixture 只显式 import traceback,其他都是「间接调用链」—— hide_internal=True
时它们被剔除,hide_internal=False 时重新出现。对照组验证 v6 过滤真的生效。
"""
script = tmp_path / "user.py"
script.write_text(
"import traceback\n"
"def f():\n"
" return 1 / 0\n"
"for i in range(50):\n"
" try:\n"
" f()\n"
" except Exception:\n"
" traceback.print_exc()\n",
encoding="utf-8",
)
res_on = profile_structure(str(script), scope="all", hide_internal=True)
res_off = profile_structure(str(script), scope="all", hide_internal=False)
on_modules = {f.module for f in res_on.functions}
off_modules = {f.module for f in res_off.functions}
# 关掉过滤时应恢复间接 stdlib 模块 —— inspect 是 traceback 的核心依赖,必然出现
extra = off_modules - on_modules
stdlib_noise = {"inspect", "annotationlib", "linecache", "tokenize",
"dataclasses", "enum", "reprlib", "textwrap", "ast",
"codeop", "contextlib"}
assert extra & stdlib_noise, (
f"--no-hide-internal 应恢复 traceback 的间接 stdlib 调用链帧,"
f"但 on vs off 的 module 差异 {extra} 里没出现 {stdlib_noise}"
f"v6 过滤可能没生效,或者测试 fixture 没触发干扰。"
)
# hide_internal=True 时不应漏出 inspect / annotationlib / linecache 等间接帧
leak = on_modules & stdlib_noise
assert not leak, (
f"hide_internal=True 时不应漏出间接 stdlib 噪声,但 {leak} 出现了。"
f"这意味着 fixture 没显式 import 它们,却进了 result.json。"
)
# ──────────────────────────────────────────────────────────────────────────────
# v7 「用户直接调用」过滤 —— 只保留用户代码 + 用户直接调用的 import 入口
# ──────────────────────────────────────────────────────────────────────────────
#
# 用户原话:「只要代码的本身和import 调用的耗时统计,其他的不需要」/「目前好像
# 仍然统计到内部测试的代码了,不合理」—— 之前的间接 stdlib 过滤解决了 stdlib 噪声,
# 但 487 帧里仍有 numpy 内部 250+ 帧、<built-in> C 函数 131 帧、<frozen> importlib
# 90 帧等「不是用户调用的」帧。v7 把规则收紧到「caller 链含用户帧才算 import 调用」,
# 并加 cumtime >= 1µs 闸门(剔除纯注册/分发占位函数如 _mean_dispatcher)。
#
# 测试矩阵:
# - 用户写的函数 → 保留
# - 用户直接调用的第三方函数 → 保留
# - 用户调用的第三方函数内部又调的第三方函数 → 剔除
# - 用户调用的第三方函数内部又调的 stdlib(非用户 import)→ 剔除
# - <built-in> C 函数 → 剔除(不是 import 调用)
# - <frozen> importlib → 剔除(不是 import 调用)
# - 用户没 import 的 stdlib 帧 → 剔除
def test_profile_structure_default_keeps_only_user_and_direct_calls(tmp_path):
"""默认过滤:只保留用户代码 + 用户直接调用的 third-party / 用户 import 的 stdlib。
fixture 用 json + 一个明显干活的 helper(json.dumps 大循环),结果应只有:
- 用户模块(脚本本身)
- 用户函数 helper
- 用户直接调的 json.dumps
不应出现:json.decoder / json.scanner / re / typing / inspect 等内部帧。
"""
script = tmp_path / "user.py"
script.write_text(
"import json\n"
"def helper(data):\n"
" return json.dumps(data)\n"
"result = helper({'a': list(range(100))})\n",
encoding="utf-8",
)
res = profile_structure(str(script), scope="all")
modules = {f.module for f in res.functions}
names = {f.name for f in res.functions}
# 1) 用户脚本本身
assert "<user>" in modules, f"用户代码应保留,实际 modules: {modules}"
assert "<module>" in names and "helper" in names
# 2) 用户直接调用的 json.dumps
assert "json" in modules, f"json 是用户显式 import 且被直接调用,应保留"
assert "dumps" in names, "用户直接调用 json.dumps,应保留"
# 3) json 内部帧被剔除(json.decoder / encoder / scanner 等)
json_internal_leak = {"decode", "encode", "scanstring", "JSONDecoder",
"JSONEncoder", "__init__"}
bad = json_internal_leak & names
assert not bad, f"json 内部帧不应漏出,但 {bad} 出现了"
# 4) stdlib 内部展开(typing/inspect/re 等)不应出现
leak = modules & {"typing", "inspect", "re", "_py_warnings", "functools"}
assert not leak, f"间接 stdlib 噪声应剔除,但 {leak} 漏进了"
def test_profile_structure_default_drops_indirect_nested_calls(tmp_path):
"""用户调用 numpy.mean,但 numpy.mean 内部调的 numpy._core._methods 不应出现。
这是 v7 的核心语义:只看「直接被用户调用」,不传递。哪怕是同包内部的辅助函数,
只要不是用户写的,就不在 result 里。
"""
script = tmp_path / "user.py"
script.write_text(
"import numpy as np\n"
"x = np.array([1, 2, 3, 4, 5])\n"
"y = np.mean(x)\n",
encoding="utf-8",
)
res = profile_structure(str(script), scope="all")
names = {f.name for f in res.functions}
# 用户直接调用的入口应保留
assert "mean" in names, f"用户直接调用 np.mean,应保留,实际 names: {names}"
# numpy 内部辅助方法不应出现(_methods / fromnumeric 之类的子帧)
# 注:numpy.mean 实际上调用 numpy._core.fromnumeric.mean,我们想要的是 fromnumeric.mean
# 出现在 result 里(因为它是 user-direct call 的代理),但其内部又调了 _methods
# 这种,不该出现。
bad = {"_methods", "_sum", "_mean", "_std", "_var"}
bad &= names
assert not bad, f"numpy 内部辅助函数不应漏出,但 {bad} 出现了"
def test_profile_structure_default_drops_builtin_and_frozen(tmp_path):
"""<built-in> C 函数 / <frozen> importlib 即使被用户代码用到也不保留。
用户原意是「import 调用」,builtin C 函数(len / dict.keys / numpy C 核等)
和 frozen import machinery 不属于「import 调用」—— 即使技术上被用户代码调到。
"""
script = tmp_path / "user.py"
script.write_text(
"import numpy as np\n"
"data = [1, 2, 3, 4, 5]\n"
"x = np.array(data)\n"
"y = len(data)\n"
"z = sorted(data)\n",
encoding="utf-8",
)
res = profile_structure(str(script), scope="all")
modules = {f.module for f in res.functions}
# <built-in> 完全不应该出现 —— 不是 import 调用
assert "<built-in>" not in modules, (
f"<built-in> C 函数不应作为 import 调用保留,实际 modules: {modules}"
)
# <frozen> 也不应出现
assert "<frozen>" not in modules, (
f"<frozen> importlib 帧不应保留,实际 modules: {modules}"
)
def test_profile_structure_default_keeps_user_function_with_no_args(tmp_path):
"""用户写的不带参数的函数(只调用 import)仍应保留 —— 用户代码一律保留。
这是 v7 的关键 invariant:用户帧不管 tottime/cumtime 多小,不管有没有调用
别的东西,都进 result —— 这是「代码本身」的定义。
"""
script = tmp_path / "user.py"
script.write_text(
"import json\n"
"def tiny():\n"
" return json.dumps({})\n"
"tiny()\n",
encoding="utf-8",
)
res = profile_structure(str(script), scope="all")
user_fns = {f.name for f in res.functions if f.module == "<user>"}
assert "tiny" in user_fns, f"用户函数 tiny 应保留,实际 user fns: {user_fns}"
def test_profile_structure_default_cumtime_floor_drops_dispatcher(tmp_path):
"""cumtime < 1µs 的 entry point 视为注册期/分发期占位 → 剔除。
numpy 在用户 import 时会触发 _mean_dispatcher / _std_dispatcher 等纯注册函数
(cProfile 把 caller 记成 <module>,但 cumtime 几乎为 0)。这些不是真正的
「用户 import 调用」,默认剔除。
"""
script = tmp_path / "user.py"
script.write_text(
"import numpy as np\n"
"y = np.mean([1, 2, 3])\n",
encoding="utf-8",
)
res = profile_structure(str(script), scope="all")
# _mean_dispatcher 是注册期占位,cumtime 应 < 1µs
dispatcher_frames = [f for f in res.functions if "_dispatcher" in f.name]
assert not dispatcher_frames, (
f"_dispatcher 注册占位函数应剔除(它们是 import 期触发的分发器,"
f"cumtime < 1µs),但 {[(f.name, f.cumtime) for f in dispatcher_frames]} 出现了"
)