PostHog Python 测试成本测量指南：本地基准、性能剖析与 CI 计时数据分析-创锋一号

PostHog Python 测试成本测量指南：本地基准、性能剖析与 CI 计时数据分析

【免费下载链接】posthog:hedgehog: PostHog is the leading platform for building self-driving products. Our developer tools – AI observability, analytics, session replay, flags, experiments, error tracking, logs, and more – capture all the context agents need to diagnose problems, uncover opportunities, and ship fixes. Steer it all from Slack, web, desktop, or the MCP.项目地址: https://gitcode.com/GitHub_Trending/po/posthog

在优化任何 Python 测试之前，先用同一种测量口径记录"改动前"与"改动后"的数据——这是本仓库维护 Python 测试套件（pytest / Django test）的第一原则。本文以 PostHog 仓库的测试维护技能文档为主线，完整讲解本地测量的三种命令、pytest 时间阶段的正确解读、cProfile / py-spy 的剖析方法，以及如何基于 CI 上报到posthog.trace_spans的计时数据编写 ClickHouse 查询，在合并前后客观验证测试优化是否真的生效。

1. 测量总原则：先测量，后改动

PostHog 仓库中的maintaining-python-tests技能（见 SKILL.md）明确了六个核心原则，其中测量贯穿始终：

先测量再改代码：按"总观测工作量"而非"某一次本地慢跑"来排序候选测试。
保留行为覆盖：保留覆盖不同校验、持久化、集成或输出路径的用例。
只删除过期或冗余的覆盖：删除测试前必须获得明确批准。
共享昂贵基础设施，而非共享可变测试状态：用唯一 ID、schema、表、topic 或租户来保证隔离。
合并后再次测量：本地结果只能证明机制成立，CI 中的真实结果必须以合并后新鲜的master数据为准。
区分"测试用例工作量"与"套件墙钟时间"：一个改动可能让"用例耗时总和"下降，但最慢的 pytest 套件墙钟时间毫无变化。

对应到测量文档本身，核心要求只有一句话：改动前后使用完全相同的测量方式，并明确写出命令、范围和计时类型（详见 measurement.md）。任何"百分比提升"如果来自两种不同的测量类型，都不算有效结论。

2. 本地测量：三种命令与适用场景

2.1 首选：仓库测试运行器`hogli test`

PostHog 将测试运行封装为hogli test命令，它会根据文件路径自动识别测试类型（pytest、Jest、Playwright、Rust、Go、Turbo）并调用正确的底层 runner。其实现位于 tools/hogli-commands/hogli_commands/test_runner.py。

hogli test path/to/test.py::TestClass::test_name

值得注意的实现细节（来自源码）：

支持 pytest node ID（文件路径::类::方法）形式的精确目标，也会从路径片段解析出::之前的部分并重新拼接为仓库相对路径（见_resolve_to_repo_relative）。
对于 Python 文件，只有位于posthog/、ee/、products/、common/、dags/、tools/、services/这些 Python 根目录下的才走 pytest 分支。
底层 pytest 命令默认带-s（流式输出）；在云端任务沙箱中（存在POSTHOG_TASK_RUN_ID且未设置HOGLI_TEST_VERBOSE）会自动把-s替换为-q，避免把海量 print 输出当作 token 消耗掉（见_quiet_pytest_in_cloud_sandbox）。
Python 测试运行会注入REDIS_URL=redis:///环境变量（macOS 上还会设置OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES，见_python_env）。
额外参数会原样透传给底层 runner；还支持--changed（运行当前分支相对master变更涉及的测试）与--watch（nodemon 监听posthog/、common/hogvm/python、ee/、dags/、products/下的改动自动重跑）两个模式。

2.2 需要 pytest 计时输出：直接使用`uv run pytest`

当你想看到 pytest 自身的耗时排名（--durations）时，绕开hogli直接调用 pytest：

uv run pytest -q path/to/test.py::TestClass::test_name --durations=20

仓库使用uv管理 Python 环境（根目录存在 pyproject.toml 与 uv.lock），所以用uv run在锁定的依赖环境里执行 pytest。--durations=20会在测试结束时打印最慢的 20 个用例及其 setup / call / teardown 各阶段耗时。

2.3 完整墙钟时间：`/usr/bin/time`

pytest 自身的计时不含进程启动与收集阶段。需要完整命令墙钟时间时，使用系统级计时工具，并显式指定输出格式：

/usr/bin/time -f 'wall=%e user=%U system=%S max_rss_kb=%M' \ uv run pytest -q path/to/test.py -k 'target_family' --durations=20

wall=%e：真实流逝时间（秒）；
user=%U/system=%S：用户态与内核态 CPU 时间；
max_rss_kb=%M：峰值常驻内存（KB），用于观察优化是否以内存为代价。

2.4 参数化族与冷/热基线

参数化族要记录全部用例：-k 'target_family'会选中一族参数化用例。不要在"改动前只测一个用例、改动后测整个族"之间做对比——两种采样的体量不同，结论无效。
冷基线 vs 热基线：
- 当优化目标是进程启动/导入/收集这类一次性成本时，运行冷基线（fresh process）；
- 当优化目标是同一进程内的重复工作（如 fixture 复用）时，运行热基线（warm run）。
- 报告时必须标注结果是 cold 还是 warm，因为两者不可直接比较。

3. 解读 pytest 时间：四个阶段，别混淆

pytest 的--durations输出可以分开报告多个阶段：

阶段	含义
`setup`	测试调用前的 fixture 准备
`call`	测试函数体本身的执行
`teardown`	测试调用后的 fixture 清理
`wall`	完整命令的墙钟时间，包含收集（collection）与进程启动

两个典型的解读陷阱：

模块级 fixture 会把成本从重复的 call 阶段搬到单次的 setup 阶段。此时所有用例的 call 耗时之和会下降，但整个命令的 wall 时间可能毫无变化。报告时必须同时给出两者，只报 call 之和会得到虚假的"提速"结论。
昂贵的 setup 不一定出现在 setup 阶段。例如 Django 迁移测试：迁移执行器是在call 阶段内运行的，所以它的 call 时间几乎等于总时间。不要仅凭"setup 阶段不长"就推断 fixture 不是成本中心。

这与技能文档中"不要把完整的 pytest wall 时间与 sum 起来的 call 时间对比"（SKILL.md 第 4 步）是同一件事：它们测量的是不同的工作。

4. 性能剖析：选对工具，命中成本中心

4.1 Python 进程内 CPU 工作：cProfile

uv run python -m cProfile -o /tmp/test.prof -m pytest -q <nodeid> uv run python - <<'PY' import pstats pstats.Stats('/tmp/test.prof').strip_dirs().sort_stats('cumulative').print_stats(40) PY

第一行把剖析结果写入/tmp/test.prof，第二行用pstats按累计耗时排序打印前 40 行。strip_dirs()让输出更易读；sort_stats('cumulative')对"定位总成本来源"最直观。

4.2 子进程、原生代码与 I/O 主导时：py-spy 或系统剖析器

cProfile 只能看到当前 Python 进程内的函数调用。当成本来自以下场景时，必须换工具：

测试启动了子进程（worker、consumer、broker、容器）；
热点在原生代码（如 C 扩展、ClickHouse 客户端、Rust UDF）；
时间主要花在 I/O 等待上。

此时使用py-spy（可对运行中的进程做 wall-clock 采样）或系统级剖析器（如perf）；如果测试启动了 worker 或容器，直接查阅对应服务的日志来定位等待点。CPU profile 会漏掉花在服务或子进程上的时间——这是 SKILL.md 第 5 步的明确告诫。

4.3 剖析范围：从最小可复现目标开始

不要一开始就剖析一个宽泛的文件。先剖析仍然能复现成本的最小目标（单个 node ID 优先）。这与技能工作流一致：第 4 步建立基线时就要求"用改动前后完全相同的命令运行精确目标"，第 6 步要求选择最小安全修复。

剖析的目的不是写计时断言——SKILL.md 明确禁止在测试里加 timing 断言（CI 计时噪声太大，不适合作为正确性测试），而是把时间归因到以下几类成本中心之一：

测试收集或环境启动；
fixture 的 setup / teardown；
数据库创建、flush 或迁移；
worker / consumer / broker / 容器启动；
测试所执行的产品代码；
快照序列化或格式化；
轮询、重试或真实等待。

5. CI 计时数据：`posthog.trace_spans`

5.1 数据来源与可用字段

Backend CI 计时上报器会把 pytest spans 写入posthog.trace_spans表（ClickHouse）。每个 span 携带以下有用字段：

service_name = 'ci-backend' resource_attributes['ci.branch'] resource_attributes['ci.run_id'] resource_attributes['ci.run_attempt'] attributes['test.runner'] attributes['test.outcome'] attributes['test.owner_team'] attributes['test.file'] attributes['test.file_source'] attributes['shard.segment'] attributes['shard.testcase_seconds'] is_root_span duration_nano

5.2 查询前的纪律

先调用/querying-posthog-data技能，再使用posthog:execute-sql工具；查询之前先核实可用的 trace 字段（schema 可能随版本演化）。
使用master分支评估合并后的影响：PR 运行不是合并后的结果（SKILL.md 的 Boundaries 明确禁止把 PR 运行当作 post-merge 结果）。
使用显式的 UTC 时间窗口，避免时区歧义。

5.3 查询一：对上报的慢 pytest 测试排序

上报器只对超过其配置时长阈值的单个测试发射 span。因此以下查询排名的是"被采样到的慢测试工作量"，而不是全部 pytest 工作量：

SELECT name, coalesce(nullIf(attributes['test.owner_team'], ''), 'unowned') AS owner_team, count() AS executions, round(quantile(0.5)(duration_nano / 1000000000), 3) AS p50_seconds, round(quantile(0.95)(duration_nano / 1000000000), 3) AS p95_seconds, round(sum(duration_nano) / 1000000000 / 3600, 2) AS observed_hours FROM posthog.trace_spans WHERE timestamp >= now() - INTERVAL 2 DAY AND service_name = 'ci-backend' AND resource_attributes['ci.branch'] = 'master' AND attributes['test.runner'] = 'pytest' AND attributes['test.outcome'] = 'passed' AND duration_nano > 0 GROUP BY name, owner_team HAVING executions >= 10 ORDER BY observed_hours DESC LIMIT 50

observed_hours（执行次数 × 时长的总观测工作量）用于发现"重复出现的中等耗时测试"，这比只看单次慢测试更有价值；
只有当样本过小或包含已知事故时段时才调整两天窗口；
"改动后少了一行"不等于测试停止运行：它可能只是掉到了上报阈值以下，需要单独确认。

5.4 查询二：合并前后对比同一测试

使用等长的两个时间窗口，并在合并时间点周围留出空隙，防止旧的 job 混入"after"组：

WITH samples AS ( SELECT multiIf( timestamp >= toDateTime('2026-01-01 00:00:00', 'UTC') AND timestamp < toDateTime('2026-01-02 00:00:00', 'UTC'), 'before', timestamp >= toDateTime('2026-01-03 00:00:00', 'UTC') AND timestamp < toDateTime('2026-01-04 00:00:00', 'UTC'), 'after', 'excluded' ) AS period, duration_nano / 1000000000 AS seconds FROM posthog.trace_spans WHERE service_name = 'ci-backend' AND resource_attributes['ci.branch'] = 'master' AND attributes['test.outcome'] = 'passed' AND name = '<exact test span name>' ) SELECT period, count() AS executions, round(quantile(0.5)(seconds), 3) AS p50_seconds, round(quantile(0.95)(seconds), 3) AS p95_seconds, round(avg(seconds), 3) AS mean_seconds FROM samples WHERE period != 'excluded' GROUP BY period ORDER BY period

使用 GitHub 上的精确合并时间戳选择窗口，并确认 after 窗口里的运行确实包含合并后的代码。该对比仅在"测试在两个时段都发射了 span"时成立。p50 反映稳态成本，p95 反映争用或尾延迟。

5.5 查询三：参数化族按每次 workflow 运行测量

WITH per_run AS ( SELECT resource_attributes['ci.run_id'] AS run_id, resource_attributes['ci.run_attempt'] AS run_attempt, sum(duration_nano) / 1000000000 AS sampled_seconds, uniq(name) AS sampled_cases FROM posthog.trace_spans WHERE timestamp >= now() - INTERVAL 2 DAY AND service_name = 'ci-backend' AND resource_attributes['ci.branch'] = 'master' AND attributes['test.outcome'] = 'passed' AND name LIKE '%::test_target_family[%]' GROUP BY run_id, run_attempt ) SELECT count() AS attempts, round(quantile(0.5)(sampled_seconds), 2) AS p50_sampled_seconds_per_attempt, round(quantile(0.95)(sampled_seconds), 2) AS p95_sampled_seconds_per_attempt, round(avg(sampled_cases), 1) AS mean_sampled_cases_per_attempt FROM per_run

name LIKE '%::test_target_family[%]'匹配参数化用例（方括号内为参数）。先核对期望用例数与被采样用例数是否一致：此查询仅在"族内每个目标用例都超过上报阈值"时有效——如果发射 span 的用例变少了，更低的时长并不是有效结论。

5.6 查询四：测量受影响的分片（shard）

WITH per_run AS ( SELECT resource_attributes['ci.run_id'] AS run_id, resource_attributes['ci.run_attempt'] AS run_attempt, max(duration_nano) / 1000000000 AS slowest_suite_seconds, sum(duration_nano) / 1000000000 AS total_suite_seconds, count() AS suites FROM posthog.trace_spans WHERE timestamp >= now() - INTERVAL 2 DAY AND service_name = 'ci-backend' AND resource_attributes['ci.branch'] = 'master' AND is_root_span AND attributes['shard.segment'] = '<segment>' GROUP BY run_id, run_attempt ) SELECT count() AS attempts, round(quantile(0.5)(slowest_suite_seconds), 2) AS p50_slowest_suite_seconds, round(quantile(0.95)(slowest_suite_seconds), 2) AS p95_slowest_suite_seconds, round(quantile(0.5)(total_suite_seconds), 2) AS p50_total_suite_seconds, round(avg(suites), 1) AS suites_per_attempt FROM per_run

最慢的 root span 近似该分段的pytest 套件关键路径（critical path）。注意它不包含checkout、缓存恢复、产物处理以及其他 GitHub Actions 步骤——它测量的是 pytest 执行本身，不是整个 CI job。

5.7 查询五：总 pytest 工作量与套件墙钟时间

WITH per_run AS ( SELECT resource_attributes['ci.run_id'] AS run_id, resource_attributes['ci.run_attempt'] AS run_attempt, sumIf(toFloatOrZero(attributes['shard.testcase_seconds']), is_root_span) AS testcase_seconds, maxIf(duration_nano, is_root_span) / 1000000000 AS slowest_suite_seconds, uniqIf(trace_id, is_root_span) AS suites FROM posthog.trace_spans WHERE timestamp >= now() - INTERVAL 2 DAY AND service_name = 'ci-backend' AND resource_attributes['ci.branch'] = 'master' GROUP BY run_id, run_attempt HAVING suites > 0 ) SELECT count() AS attempts, round(quantile(0.5)(testcase_seconds), 1) AS p50_testcase_seconds, round(quantile(0.95)(testcase_seconds), 1) AS p95_testcase_seconds, round(quantile(0.5)(slowest_suite_seconds), 1) AS p50_slowest_suite_seconds, round(quantile(0.95)(slowest_suite_seconds), 1) AS p95_slowest_suite_seconds FROM per_run

两个关键解释：

shard.testcase_seconds包含每一个 JUnit testcase（包括低于单个 span 上报阈值的测试），因此它代表完整的用例工作量；
sum 起来的 testcase 时间会低估 job 步骤的真实耗时：差距随"收集到的测试数量"增长，而不是随它们的时长增长——因为收集和模块导入对每个测试都有固定成本。不要仅凭求和时长来给套件定大小。

5.8 查询六：测量所有权覆盖

SELECT toDate(timestamp) AS day, count() AS test_spans, countIf(nullIf(attributes['test.owner_team'], '') IS NULL) AS unowned_spans, round(100 * unowned_spans / test_spans, 2) AS unowned_percent FROM posthog.trace_spans WHERE timestamp >= now() - INTERVAL 2 DAY AND service_name = 'ci-backend' AND attributes['test.runner'] = 'pytest' GROUP BY day ORDER BY day

把所有权视为路由结果（routing result），它并不证明测试成本下降了。所有权指标回答"这些测试由谁负责"，与"这些测试是否变快"是两个独立问题。

6. 报告限制：声明"改进"之前必须确认的条件

在陈述任何优化成果之前，逐条确认（这也是 SKILL.md 第 10 步"合并后验证"的落地点）：

两次样本使用同一个分支；
两次样本使用同一个测试名或族规则；
两次样本包含相同的用例集合；
两次样本使用相同的计时类型（例如不能拿 call 时间对比 wall 时间）；
after 样本确实包含合并后的代码（用精确 merge 时间戳确认窗口）；
样本量大到足以抵御一次异常运行；
证据不是被改动步骤自己写出的文件——一个步骤写出的文件不能证明该步骤本身正确；要验证某个修正或过滤器，必须在原始输入上重新运行它；
当同一时间窗口内还合并了无关改动时，必须如实说明；因果性结论只能用"精确测试"的结果，整体运行结果只能作为方向性证据（directional evidence）。

此外，技能工作流还要求按固定格式汇报本地结果（SKILL.md 第 9 步），其中必须包含 Target / Regression / Cost center / Change / Before / After / Correctness / Isolation / Follow-up 九项——"Before / After"必须使用同一指标、同一命令、同一用例集合与同一冷热状态。

7. 常见误区速查

误区	正确做法
用 call 时间下降宣称"CI 变快"	call 下降而 wall 不变时，两个事实都要报告
改动前测 1 个用例、改动后测整个参数族	族内所有用例都要记录
用`.test_durations`里的默认值做排序依据	该文件对 pytest-split 无法计时的测试写入扁平默认值（0.01、18.0、60.0），不是测量值（SKILL.md 第 2 步）
用 PR 运行当合并后结果	只使用新鲜的`master`运行
从包含无关改动的宽窗口断言因果	精确测试结果才能支撑因果声明
断言"某个测试消失了"	可能只是掉到上报阈值以下，需要单独确认
用求和 testcase 时长给套件定大小	求和会低估 job 步骤，差距随收集测试数量增长

测量只是维护 Python 测试套件的一半工作。选定优化方案前，还应阅读同一技能目录下的 optimization-patterns.md；动手之前先查看 docs/internal/ci-things-already-tried.md，避免重建已经被实测否决的并行、分片或覆盖率选择方案。本地验证机制、合并后在master上用上述 SQL 验证结果——这条闭环正是 PostHog 维护大规模 pytest + Django 测试套件的标准做法。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析