#!/usr/bin/env python3 # -*- coding: utf-8 -*- r""" 主机侧纯逻辑单元测试的**覆盖率**口径、报告与增量门禁（纯标准库）。 做什么： 1. 发现 test/host/**/run_unit_test.py（与 scripts/ci_local.py 的主机侧纯逻辑单测同一套发现规则）， 直接 import 各模块运行器、复用其中声明的 TestBuild（**单一事实源**：include 目录、 桩顺序、测试单元路径都不在这里重复一遍）； 2. 用**与门禁完全相同的编译开关**（unit_test_harness._BASE_FLAGS）另建一份带 /Zi 调试 信息的产物，交给 Microsoft.CodeCoverage.Console.exe 动态插桩运行； 3. 解析 Cobertura XML，只保留**自研生产源码**（App/**、Drivers/BSP/** 的 .c/.h）的行与 函数命中，对多套单测重复触及的同一文件取**并集**（命中一次即算覆盖）； 4. 与 scripts/test_coverage_baseline.json 比对：**已覆盖行/函数只许增不许减**；并打印 自研生产源文件里有多少从未被任何主机单测编译过（未触及面 = 覆盖率的 0 分母盲区）。 为什么要它（2026-10-23，W5 验证常态）： 原评审报告 §4.8 记的两处空白之一（该报告 2026-10-25 已删除）：无覆盖率口径—— 主机侧纯逻辑单测只回答“跑过哪些 TC”，回答不了“生产源码里哪些行根本没进过”。用例数与 TC 追溯率 会随时间上升，但覆盖率照的是另一类真相：**新增分支一行没测**（那类“读代码 才发现”的缺陷，覆盖率是最便宜的机器照法）。 工具链现实（2026-10-23 本机实测，换机必读）： - **Windows 原生侧没有 gcc / clang / gcov / lcov / gcovr**（全盘扫 0 处）→ gcov 那条最主流的路在 Windows 侧走不通。**但 WSL2 Ubuntu 里有 gcc/gcov 15.2.0**，实测 16 套单测在 gcc 下全部编过、 gcov -b -c 还给出分支覆盖——不用它当默认是因为**换编译器=换分母**（sha256.c：gcc 105 行 vs MSVC 114 行），详见 docs/测试说明/单元测试/主机单测覆盖率口径.md §2.1 的决策记录。 - **MSVC 14.51（VS 2026/18）已不认 /coverage**：cl 只报 warning D9002「忽略未知选项」， 然后**照常编译成功** → 拿它做覆盖率等于静默出假数据，绝不可用（/fsanitize=coverage 同）。 - Team Tools\Dynamic Code Coverage Tools\CodeCoverage.exe（17.0）还在但已 deprecated， 且 collect / analyze / instrument 一律只打印「用法: CodeCoverage 」——不可用。 - 可用的是 **Microsoft.CodeCoverage.Console.exe**（mscoverage 18.x，随 VS 装在 Common7\IDE\Extensions\Microsoft\CodeCoverage.Console\）：collect -f cobertura 直接吃 MSVC 原生 exe + PDB，且**子进程退出码原样透传**（实测子进程 return 3 → collect 也返回 3）。 - 结论：编译期插桩这条路在本机是断的，走**加载期动态插桩**（mscoverage collect）。 口径（一句话版，完整版见 docs/测试说明/单元测试/主机单测覆盖率口径.md）： 行覆盖 = 至少执行过一次的可执行行 / 被插桩的可执行行； 函数覆盖 = 至少有一行命中的函数 / 被插桩的函数； 统计面 = 自研生产源码（App/**、Drivers/BSP/** 的 .c/.h），**不含**测试单元自身、 test/host 下的桩、Lib/、Middlewares/、HAL/CMSIS、Windows SDK/UCRT/MSVC 头。 退出码： 0 = 全部单测通过，且覆盖率相对基线没有回退 1 = 有单测失败，或覆盖率相对基线回退（已覆盖行/函数减少、在测文件或模块消失） 2 = 环境错误（找不到 cl.exe / 找不到 mscoverage / 编译失败 / XML 缺失或解析失败） 3 = 用法错误 中间产物目录（2026-10-25 起与主机单测同家）： %TEMP%\dtu_host\<工作树哈希>\coverage\<模块>\ —— exe / PDB / .coverage / cobertura XML 都在这里，**不进仓库**。按工作树分家，是因为旧的固定共享名 %TEMP%\dtu_test_cov 在多棵工作树并行跑覆盖率时会互相覆盖（判据与来龙去脉见 test/host/common/unit_test_harness.py 顶部"产物目录归属"一节）。改根：--work-dir 或环境变量 DTU_HOST_BUILD。 用法（仓库根目录）： & '.venv\Scripts\python.exe' scripts/check_test_coverage.py # 全量跑 + 比基线 & '.venv\Scripts\python.exe' scripts/check_test_coverage.py --module sha256 # 只跑一套（调试） & '.venv\Scripts\python.exe' scripts/check_test_coverage.py --update-baseline # 跑完写回基线 & '.venv\Scripts\python.exe' scripts/check_test_coverage.py --md <路径> --json <路径> & '.venv\Scripts\python.exe' scripts/check_test_coverage.py --list # 只列模块 实测坑（都在这条链上踩过，勿重复排查）： - **必须 /Zi 且链接 /INCREMENTAL:NO**：mscoverage 靠 PDB 把指令映射回源行；增量链接会出 .ilk、PDB 与代码不同步 → 行号漂移甚至整模块 0 覆盖。 - **PDB 必须与 exe 同目录**：/Fe: 决定 exe 落点，PDB 跟着 exe；换目录再跑就找不到符号。 - **只能 /Od**：/O2 会合并删除行，行覆盖率与源码对不上（本项目按 /Od 口径计）。 - **-- 分隔符不能省**：collect [options] -- [args]；少了它被测程序的参数会被吃掉。 - **被测程序的 stdout 会被并进 collect 的 stdout**：判“测试真跑了”要在这份合并输出里找 汇总行（TestBuild.summary_marker），别指望单独一路流。 - **一次 collect 只覆盖这一次运行的进程树**：多套单测各跑各的，最后在本脚本里做并集。 - **覆盖率是"跑到的那几套的并集"**：`--module` 子集跑必然少覆盖行、少文件，**全部回退判定都 只在全量跑时生效**（首版只挡了"文件消失"，结果 `--module sha256` 报"总计 2552→111"假回退）。 - **PDB 里的源路径大小写会漂**（盘符、VS 根），比对前先按仓库实际拼写归一，否则同一文件 在基线里裂成两份。 设计依据：docs/测试说明/单元测试/主机单测覆盖率口径.md 创建日期：2026-10-23 """ import argparse import importlib.util import json import os import re import shutil import subprocess import sys import xml.etree.ElementTree as ET from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent HARNESS_DIR = REPO_ROOT / "test" / "host" / "common" HOST_TEST_DIR = REPO_ROOT / "test" / "host" BASELINE_PATH = REPO_ROOT / "scripts" / "test_coverage_baseline.json" # 统计面：只算自研生产源码（命中 INCLUDE 且不命中 EXCLUDE 才入选） INCLUDE_PREFIXES = ("App/", "Drivers/BSP/") EXCLUDE_PREFIXES = () SOURCE_SUFFIXES = (".c", ".h") # 生成物标记：文件头注释里写着这些字样的源文件不进分母（覆盖率对它没有意义） GENERATED_RE = re.compile(r"自动生成|请勿手改|DO NOT EDIT", re.I) BASELINE_SCHEMA = 1 # 覆盖率专用追加开关：PDB（行映射）+ 不优化（行号不漂移）+ 非增量链接（PDB 与代码同步） COV_FLAGS = ["/Zi", "/Od"] COV_LINK_FLAGS = ["/link", "/INCREMENTAL:NO"] # -------------------------------------------------------------------------- # harness 复用（工具链探测与编译开关的单一事实源） # -------------------------------------------------------------------------- def load_harness(): """导入 test/host/common/unit_test_harness.py；失败返回 None（调用方按环境错误处理）。""" if not (HARNESS_DIR / "unit_test_harness.py").is_file(): return None if str(HARNESS_DIR) not in sys.path: sys.path.insert(0, str(HARNESS_DIR)) try: import unit_test_harness except Exception as exc: print("[ENV] 导入 unit_test_harness 失败: {}".format(exc), file=sys.stderr) return None return unit_test_harness def discover_modules(): """{模块名: run_unit_test.py 绝对路径}；发现规则与 ci_local.py 的主机侧纯逻辑单测一致。""" out = {} if not HOST_TEST_DIR.is_dir(): return out for p in sorted(HOST_TEST_DIR.rglob("run_unit_test.py")): out[p.parent.name] = str(p) return out def load_test_build(runner_path, harness): """从模块运行器取模块级 BUILD（TestBuild 实例），不执行它的 main。 各运行器都是声明式的：顶层只 import + 构造 TestBuild，真跑在 if __name__ == "__main__" 里，所以按独立模块名 import 是安全的。 """ mod_name = "dtu_cov_" + re.sub(r"\W+", "_", Path(runner_path).parent.name) spec = importlib.util.spec_from_file_location(mod_name, runner_path) if spec is None or spec.loader is None: return None, "无法为 {} 构造 import spec".format(runner_path) mod = importlib.util.module_from_spec(spec) try: spec.loader.exec_module(mod) except SystemExit: return None, "{} 在 import 阶段就调用了 sys.exit()".format(runner_path) except Exception as exc: return None, "import {} 失败: {}".format(runner_path, exc) build = getattr(mod, "BUILD", None) if build is None: return None, "{} 未声明模块级 BUILD，覆盖率脚本无法复用其编译配置".format(runner_path) if not isinstance(build, harness.TestBuild): return None, "{} 的 BUILD 不是 TestBuild（与 harness 类型漂移）".format(runner_path) return build, None def _vs_roots(harness): """候选 VS 安装根：cl.exe 的祖先优先，再补 harness 的标准布局表（含 BuildTools）。""" roots = [] cl, _, _ = harness.locate_toolchain() if cl: # \VC\Tools\MSVC\\bin\Hostx64\x64\cl.exe -> （上溯 8 级） roots.append(Path(cl).parents[7]) for r in getattr(harness, "_default_vs_roots", list)(): p = Path(r) if p not in roots: roots.append(p) return roots def locate_coverage_console(harness, verbose=False): r"""返回 mscoverage 绝对路径；找不到返回 None。 优先级：DTU_MS_COVERAGE（设了就必须有效，**不静默回退**）→ 各 VS 安装根下的 Common7\IDE\Extensions\Microsoft\CodeCoverage.Console\ → PATH。 """ explicit = os.environ.get("DTU_MS_COVERAGE", "").strip() if explicit: if not os.path.isfile(explicit): print("[ENV] DTU_MS_COVERAGE 已设置但文件不存在: {}（请修正该变量，不做自动探测回退）" .format(explicit), file=sys.stderr) return None return explicit rel = Path("Common7") / "IDE" / "Extensions" / "Microsoft" / "CodeCoverage.Console" \ / "Microsoft.CodeCoverage.Console.exe" for root in _vs_roots(harness): cand = root / rel if verbose: print(" [probe] {}".format(cand)) if cand.is_file(): return str(cand) return shutil.which("Microsoft.CodeCoverage.Console.exe") # -------------------------------------------------------------------------- # 单模块：编译（带 PDB）-> mscoverage collect -> 解析 Cobertura # -------------------------------------------------------------------------- def build_with_pdb(harness, build, workdir, verbose=False): """按门禁同一套开关 + COV_FLAGS 编译测试单元。返回 (退出码, 合并输出)。""" cl, inc, lib = harness.locate_toolchain() if not cl: return 2, "找不到 MSVC cl.exe（设置 DTU_MSVC_CL，或安装 VS C++ 工具链）" objdir = os.path.join(workdir, "obj") os.makedirs(objdir, exist_ok=True) exe = os.path.join(workdir, build.resolved_exe()) env = dict(os.environ) env["INCLUDE"] = os.pathsep.join(inc) env["LIB"] = os.pathsep.join(lib) env["PATH"] = os.path.dirname(cl) + os.pathsep + env.get("PATH", "") cmd = [cl] + list(harness._BASE_FLAGS) + list(COV_FLAGS) for d in build.include_dirs: cmd += ["/I", d] cmd += list(build.extra_flags) cmd += [build.src, "/Fe:" + exe, "/Fo:" + objdir + os.sep] + list(COV_LINK_FLAGS) cp = subprocess.run(cmd, env=env, capture_output=True, cwd=workdir) return cp.returncode, harness.decode(cp.stdout) + harness.decode(cp.stderr) def collect_coverage(mscov, exe, workdir, xml_path, timeout_s, verbose=False): """动态插桩运行 exe 并导出 Cobertura XML。返回 (退出码, 合并输出, xml 路径或 None)。 退出码 = 被测程序退出码（mscoverage 透传）：0 = 通过，非 0 = 失败或异常。 """ cmd = [mscov, "collect", "-o", xml_path, "-f", "cobertura", "--nologo", "--", exe] env = dict(os.environ) env["PATH"] = os.path.dirname(mscov) + os.pathsep + env.get("PATH", "") try: cp = subprocess.run(cmd, env=env, capture_output=True, cwd=workdir, timeout=timeout_s) except subprocess.TimeoutExpired: return 2, "mscoverage collect 超时（{} s）".format(timeout_s), None except OSError as exc: return 2, "mscoverage 无法执行: {}".format(exc), None out = cp.stdout.decode("utf-8", "replace") + cp.stderr.decode("utf-8", "replace") xml = xml_path if os.path.isfile(xml_path) and os.path.getsize(xml_path) > 0 else None if verbose and xml is None: print(out[-1500:]) return cp.returncode, out, xml # -------------------------------------------------------------------------- # 路径与 Cobertura 解析 # -------------------------------------------------------------------------- def rel_to_repo(path_str): """PDB 里的源文件绝对路径 -> 仓库相对 posix 路径；仓库外返回 None。 Windows 下大小写不可信（盘符、VS 根都可能漂），故比较用 normcase、切片用原串， 最后再由 canonical 表统一成仓库里的实际拼写。 """ if not path_str: return None p = path_str if os.path.isabs(path_str) else os.path.join(str(REPO_ROOT), path_str) p = os.path.abspath(p) root = os.path.abspath(str(REPO_ROOT)) if os.path.normcase(p).startswith(os.path.normcase(root) + os.sep): return p[len(root) + 1:].replace("\\", "/") return None def in_scope(rel): """是否在覆盖率统计面内（自研生产源码）；大小写不敏感。""" if not rel: return False low = rel.lower() if not low.endswith(SOURCE_SUFFIXES): return False # 前缀表本身带大小写（App/、Drivers/BSP/），比较前先整体小写 if not low.startswith(tuple(x.lower() for x in INCLUDE_PREFIXES)): return False return not low.startswith(tuple(x.lower() for x in EXCLUDE_PREFIXES)) def parse_cobertura(xml_path, canon): """Cobertura XML -> {仓库相对路径: {"lines": {行: 命中}, "funcs": {函数: 命中}}}。 同一 (文件, 行) 多次出现取最大值；文件名先经 canon 表归一，避免大小写裂成两份台账。 """ tree = ET.parse(xml_path) result = {} for cls in tree.getroot().iter("class"): rel = rel_to_repo(cls.get("filename", "")) if not in_scope(rel): continue rel = canon.get(rel.lower(), rel) entry = result.setdefault(rel, {"lines": {}, "funcs": {}}) for line in cls.iter("line"): try: num = int(line.get("number", "0")) hits = int(line.get("hits", "0")) except ValueError: continue if num <= 0: continue if hits >= entry["lines"].get(num, 0): entry["lines"][num] = hits for meth in cls.iter("method"): name = meth.get("name") or "?" hits = 0 for line in meth.iter("line"): try: hits = max(hits, int(line.get("hits", "0") or 0)) except ValueError: pass if hits >= entry["funcs"].get(name, 0): entry["funcs"][name] = hits return result def merge_into(total, part): """单模块结果并入总表（同一 (文件, 行) 取并集：命中一次即算覆盖）。""" for rel, entry in part.items(): dst = total.setdefault(rel, {"lines": {}, "funcs": {}}) for num, hits in entry["lines"].items(): if hits >= dst["lines"].get(num, 0): dst["lines"][num] = hits for name, hits in entry["funcs"].items(): if hits >= dst["funcs"].get(name, 0): dst["funcs"][name] = hits def summarize(total): """{文件: {lines, funcs}} -> 文件级指标（valid/covered + 覆盖率 + 未覆盖行清单）。""" out = {} for rel, entry in sorted(total.items()): lines, funcs = entry["lines"], entry["funcs"] valid = len(lines) covered = sum(1 for h in lines.values() if h > 0) f_valid = len(funcs) f_covered = sum(1 for h in funcs.values() if h > 0) out[rel] = { "lines_valid": valid, "lines_covered": covered, "funcs_valid": f_valid, "funcs_covered": f_covered, "line_rate": round(covered / valid, 4) if valid else 0.0, "func_rate": round(f_covered / f_valid, 4) if f_valid else 0.0, "missing_lines": sorted(n for n, h in lines.items() if h == 0), } return out def pct(x): return "{:.1f}%".format(100.0 * x) if x is not None else "-" # -------------------------------------------------------------------------- # 分母侧：生产源文件全集 / 生成物 / 未触及面 # -------------------------------------------------------------------------- def head_text(rel_path, n_lines=15): try: with open(REPO_ROOT / rel_path, "r", encoding="utf-8", errors="replace") as f: return "".join(f.readlines()[:n_lines]) except OSError: return "" def is_generated(rel_path): """文件头 15 行内出现「自动生成 / 请勿手改 / DO NOT EDIT」即判为生成物。""" return bool(GENERATED_RE.search(head_text(rel_path))) def production_sources(): """统计面内的全部生产源文件（仓库相对 posix 路径，按文件系统实际拼写）。""" out = [] for prefix in INCLUDE_PREFIXES: base = REPO_ROOT / prefix if not base.is_dir(): continue for p in sorted(base.rglob("*")): if not p.is_file() or not p.name.lower().endswith(SOURCE_SUFFIXES): continue rel = str(p.relative_to(REPO_ROOT)).replace("\\", "/") if rel.lower().startswith(EXCLUDE_PREFIXES): continue out.append(rel) return out def count_code_lines(rel_path): """粗粒度有效代码行（去空行、去整行注释、去 include）。 **只用于未触及面的体量提示**，不进覆盖率分母——分母一律来自插桩报告，不靠猜。 """ try: with open(REPO_ROOT / rel_path, "r", encoding="utf-8", errors="replace") as f: text = f.read() except OSError: return 0 n, in_block = 0, False for raw in text.splitlines(): s = raw.strip() if not s: continue if in_block: if "*/" in s: in_block = False continue if s.startswith("/*"): in_block = "*/" not in s continue if s.startswith("*") or s.startswith("//") or s.startswith("#include"): continue n += 1 return n # -------------------------------------------------------------------------- # 基线：读取 / 写出 / 比对（增量拦截：拦的是回退，不拦进步） # -------------------------------------------------------------------------- def load_baseline(path): if not Path(path).is_file(): return None try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except (OSError, ValueError) as exc: print("[ENV] 基线读取失败 {}: {}".format(path, exc), file=sys.stderr) return None BASELINE_FILE_FIELDS = ("lines_valid", "lines_covered", "funcs_valid", "funcs_covered", "line_rate", "func_rate") def write_baseline(path, modules, files, totals, tested_sources, untested): # 基线只留可比字段：missing_lines 每次跑都会漂，进了台账就是无意义的提交抖动 payload = { "schema": BASELINE_SCHEMA, "generated_by": "scripts/check_test_coverage.py --update-baseline", "note": "主机单测覆盖率基线。口径与工具链限制见 docs/测试说明/单元测试/主机单测覆盖率口径.md。" "门禁判据：files[].lines_covered 与 funcs_covered 只增不减，下降即拦提交。", "totals": totals, "modules": modules, "files": {rel: {k: v for k, v in rec.items() if k in BASELINE_FILE_FIELDS} for rel, rec in files.items()}, "tested_sources": tested_sources, "untested_sources": untested, } with open(path, "w", encoding="utf-8", newline="\n") as f: json.dump(payload, f, ensure_ascii=False, indent=2, sort_keys=True) f.write("\n") def diff_baseline(baseline, modules, files, totals, full_run): """返回 (回退项, 提示项)；回退项非空即门禁失败。""" regress, notes = [], [] if baseline is None: notes.append("无基线文件：先跑 scripts/check_test_coverage.py --update-baseline 生成 " "scripts/test_coverage_baseline.json，否则本门禁只是又跑了一遍测试，" "没有任何比较基准") return regress, notes if baseline.get("schema") != BASELINE_SCHEMA: notes.append("基线 schema={} 与当前 {} 不一致：用 --update-baseline 重建" .format(baseline.get("schema"), BASELINE_SCHEMA)) return regress, notes base_files = baseline.get("files", {}) base_mods = baseline.get("modules", {}) for rel, b in sorted(base_files.items()): cur = files.get(rel) if cur is None: if full_run: regress.append("{}: 基线里在测，本次未出现在任何单测编译单元里" "（include 链被删？还是那套单测被跳过了？）基线 {}/{} 行".format( rel, b.get("lines_covered", 0), b.get("lines_valid", 0))) continue # 覆盖率是"跑到的套数的并集"，子集必然比全量少覆盖行 → 不是回退，别假报警 if not full_run: continue if cur["lines_covered"] < b.get("lines_covered", 0): regress.append("{}: 已覆盖行 {} -> **{}**（-{}）".format( rel, b.get("lines_covered", 0), cur["lines_covered"], b.get("lines_covered", 0) - cur["lines_covered"])) if cur["funcs_covered"] < b.get("funcs_covered", 0): regress.append("{}: 已覆盖函数 {} -> **{}**".format( rel, b.get("funcs_covered", 0), cur["funcs_covered"])) if cur["lines_valid"] > int(b.get("lines_valid", 0) * 1.3) + 120: notes.append("{}: 可执行行 {} -> {}（源码长大或被更多测试 include，百分比会被摊薄）" "：确认后 --update-baseline 登记".format( rel, b.get("lines_valid", 0), cur["lines_valid"])) for name, b in sorted(base_mods.items()): cur = modules.get(name) if cur is None: if full_run: regress.append("模块 {}: 基线里有，本次未跑到（运行器缺失 / import 失败？）".format(name)) elif not cur.get("ok"): regress.append("模块 {}: 本次未通过（{}）".format(name, cur.get("reason", "未知"))) bt = baseline.get("totals", {}) # 子集运行的"总计"必然只包含被跑到的那几套，拿它比全量基线是假回退 if full_run and totals["lines_covered"] < bt.get("lines_covered", 0): regress.append("总计: 已覆盖行 {} -> **{}**".format(bt.get("lines_covered", 0), totals["lines_covered"])) for rel in sorted(set(files) - set(base_files)): notes.append("新纳入统计: {}（{}/{} 行，{}）——确认无误后 --update-baseline 登记".format( rel, files[rel]["lines_covered"], files[rel]["lines_valid"], pct(files[rel]["line_rate"]))) if not full_run: notes.append("本次是 --module 子集运行：并集必然变小，跳过全部覆盖率回退判定" "（文件消失 / 行减少 / 函数减少 / 总计下降）；只保留" "「单测必须通过」与「新文件登记提示」两项。要判回退请跑全量") return regress, notes # -------------------------------------------------------------------------- # 输出 # -------------------------------------------------------------------------- def print_report(modules, files, totals, untested, generated, notes, regress, verbose): print("=== 主机单测覆盖率（口径：docs/测试说明/单元测试/主机单测覆盖率口径.md）===") print("") print("--- 各套单测（插桩运行）---") for name in sorted(modules): m = modules[name] if not m["ok"]: print(" {:<12} FAIL {}".format(name, m.get("reason", ""))) continue print(" {:<12} ok 触及 {:>2} 个生产文件 / {:>5} 行".format( name, len(m["files"]), m["lines_total"])) print("") print("--- 自研生产源码（全部套件并集）---") print(" {:<56} {:>9} {:>7} {:>8} {:>8}".format("文件", "覆盖行", "总行", "函数", "覆盖率")) for rel in sorted(files): s = files[rel] print(" {:<56} {:>9} {:>7} {:>8} {:>8}".format( rel[-56:], s["lines_covered"], s["lines_valid"], "{}/{}".format(s["funcs_covered"], s["funcs_valid"]), pct(s["line_rate"]))) print("") print("--- 未触及面：生产源码里从未进过任何主机单测的文件 ---") print(" {} 个 / 有效代码约 {} 行（这些文件覆盖率是 0 分母，绿灯不代表它们被测）".format( len(untested), sum(u["code_lines"] for u in untested))) show = untested if verbose else untested[:12] for u in show: print(" {:<56} {:>6} 行".format(u["path"][-56:], u["code_lines"])) if len(show) < len(untested): print(" ...（另有 {} 个，--verbose 全列）".format(len(untested) - len(show))) if generated: print(" 生成物已排除（不进分母）: {}".format(", ".join(generated))) if notes: print("") print("--- 提示（不拦提交）---") for n in notes: print(" [NOTE] " + n) if regress: print("") print("--- 回退（拦提交）---") for r in regress: print(" [REGRESS] " + r) print("") print("[SUMMARY] coverage: line={}/{} ({}) func={}/{} ({}) suites={}/{} " "tested_files={} untested_files={}".format( totals["lines_covered"], totals["lines_valid"], pct(totals["line_rate"]), totals["funcs_covered"], totals["funcs_valid"], pct(totals["func_rate"]), totals["modules_ok"], totals["modules_total"], totals["files_tested"], totals["files_untested"])) def write_markdown(path, modules, files, totals, untested, notes, regress): """一次性 Markdown 报告（建议放临时目录，不入库）。 ⚠ 行拼装一律走 code() 显式拼接：`.format` 的优先级高于 `+`，写成 `a + "{}".format(x)` 只有最后一段会被格式化，占位符原样留在表里（实测踩过一次）。 """ def code(s): return BT_MARK + str(s) + BT_MARK out = [ "# 主机单测覆盖率报告", "", "> 由 " + code("scripts/check_test_coverage.py") + " 生成；口径、工具链限制与盲区见 " + code("docs/测试说明/单元测试/主机单测覆盖率口径.md") + "。本报告是一次性快照，重跑即覆盖。", "", "## 总计", "", "| 指标 | 覆盖 | 总数 | 覆盖率 |", "|---|---|---|---|", "| 行 | {} | {} | {} |".format(totals["lines_covered"], totals["lines_valid"], pct(totals["line_rate"])), "| 函数 | {} | {} | {} |".format(totals["funcs_covered"], totals["funcs_valid"], pct(totals["func_rate"])), "| 套件（通过/总数） | {} | {} | - |".format(totals["modules_ok"], totals["modules_total"]), "| 在测文件 / 未触及文件 | {} | {} | - |".format(totals["files_tested"], totals["files_untested"]), "", "## 各套单测（插桩运行）", "", "| 套件 | 结果 | 触及生产文件 | 插桩行 |", "|---|---|---|---|", ] for name in sorted(modules): m = modules[name] out.append("| " + code(name) + " | " + ("ok" if m["ok"] else "**FAIL**") + " | " + str(len(m.get("files", {}))) + " | " + str(m.get("lines_total", 0)) + " |") out += ["", "## 生产文件明细（多套取并集）", "", "| 文件 | 覆盖行 | 总行 | 覆盖率 | 函数 | 未覆盖行 |", "|---|---|---|---|---|---|"] for rel in sorted(files): s = files[rel] miss = ", ".join(str(n) for n in s["missing_lines"][:40]) if len(s["missing_lines"]) > 40: miss += " …(+" + str(len(s["missing_lines"]) - 40) + ")" out.append("| " + code(rel) + " | " + str(s["lines_covered"]) + " | " + str(s["lines_valid"]) + " | " + pct(s["line_rate"]) + " | " + str(s["funcs_covered"]) + "/" + str(s["funcs_valid"]) + " | " + (miss or "—") + " |") out += ["", "## 未触及的生产源文件", "", "共 " + str(len(untested)) + " 个（有效代码约 " + str(sum(u["code_lines"] for u in untested)) + " 行）。主机单测从未编译过它们——**对它们而言覆盖率是 0 分母，不是 0%**。", "", "| 文件 | 有效代码行 |", "|---|---|"] for u in untested: out.append("| " + code(u["path"]) + " | " + str(u["code_lines"]) + " |") if notes: out += ["", "## 提示（不拦提交）", ""] + ["- [NOTE] " + n for n in notes] if regress: out += ["", "## 回退（拦提交）", ""] + ["- [REGRESS] " + r for r in regress] out.append("") Path(path).write_text("\n".join(out), encoding="utf-8", newline="\n") # -------------------------------------------------------------------------- # 主流程 # -------------------------------------------------------------------------- def run(args): for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(encoding="utf-8", errors="replace") except Exception: pass harness = load_harness() if harness is None: print("[ENV] 取不到 test/host/common/unit_test_harness.py", file=sys.stderr) return 2 all_runners = discover_modules() if not all_runners: print("[ENV] 未发现 test/host/**/run_unit_test.py", file=sys.stderr) return 2 runners = all_runners if args.module: want = set(args.module) unknown = want - set(all_runners) if unknown: print("[USAGE] 未知模块 {}；可选：{}".format( ", ".join(sorted(unknown)), ", ".join(sorted(all_runners))), file=sys.stderr) return 3 runners = {k: v for k, v in all_runners.items() if k in want} full_run = len(runners) == len(all_runners) if args.list: for name in sorted(all_runners): print(name) return 0 mscov = None if not args.build_only: mscov = locate_coverage_console(harness, verbose=args.verbose) if not mscov: print("[ENV] 找不到 Microsoft.CodeCoverage.Console.exe（随 VS 装在 " "Common7\\IDE\\Extensions\\Microsoft\\CodeCoverage.Console\\）；" "可用 DTU_MS_COVERAGE 指定绝对路径。本机实测：MSVC 14.51 已无 /coverage，" "且 PATH 上没有 gcc/gcov。", file=sys.stderr) return 2 prod = production_sources() canon = {rel.lower(): rel for rel in prod} # 中间产物根：**跟着工作树走**，与主机单测同一个家（task-10）。旧写法是 # tempfile.gettempdir()/dtu_test_cov —— 全机一个名字，多棵工作树并行跑覆盖率就 # 互相覆盖 exe/PDB/cov 文件（docs/测试说明/单元测试/主机单测覆盖率口径.md 里"并行期每人 # 只跑 --module 自己模块"那条规避，就是为了绕开它；分家之后不再需要）。 # 复用 harness 的同一套判据（含 DTU_HOST_BUILD 覆盖），避免两处口径漂移。 _host_build_root = getattr(harness, "host_build_root", None) if _host_build_root is None: print("[ENV] unit_test_harness 缺 host_build_root()——覆盖率产物无法与主机" "单测同家（harness 版本过旧？），拒绝跑。", file=sys.stderr) return 2 work_root = args.work_dir or os.path.join(_host_build_root(), "coverage") if args.clean: shutil.rmtree(work_root, ignore_errors=True) os.makedirs(work_root, exist_ok=True) total_files = {} modules = {} for name, runner in sorted(runners.items()): workdir = os.path.join(work_root, name) shutil.rmtree(workdir, ignore_errors=True) os.makedirs(workdir, exist_ok=True) build, err = load_test_build(runner, harness) if err: modules[name] = {"ok": False, "reason": err, "files": {}, "lines_total": 0} print("[FAIL] {}".format(err), file=sys.stderr) continue rc, out = build_with_pdb(harness, build, workdir, verbose=args.verbose) if rc != 0: modules[name] = {"ok": False, "reason": "编译失败（cl 退出码 {}）".format(rc), "files": {}, "lines_total": 0} print("[ENV] {} 编译失败：".format(name), file=sys.stderr) print(out[-2500:], file=sys.stderr) continue if args.build_only: modules[name] = {"ok": True, "files": {}, "lines_total": 0, "build_only": True} print("[OK] {} 覆盖率形态编译通过".format(name)) continue exe = os.path.join(workdir, build.resolved_exe()) xml_path = os.path.join(workdir, "coverage.cobertura.xml") crc, cout, xml = collect_coverage(mscov, exe, workdir, xml_path, args.timeout, verbose=args.verbose) marker = build.summary_marker or "" if marker and marker not in cout: modules[name] = {"ok": False, "reason": "插桩运行未产生测试汇总行（找不到 " + repr(marker) + "）", "files": {}, "lines_total": 0} print("[ENV] {} 测试汇总行缺失（rc={}）".format(name, crc), file=sys.stderr) continue if crc != 0: modules[name] = {"ok": False, "reason": "单测失败（退出码 {}）".format(crc), "files": {}, "lines_total": 0} print("[FAIL] {} 在插桩运行下失败（rc={}）".format(name, crc), file=sys.stderr) continue if xml is None: modules[name] = {"ok": False, "reason": "未生成覆盖率 XML", "files": {}, "lines_total": 0} print("[ENV] {} 缺覆盖率 XML: {}".format(name, xml_path), file=sys.stderr) continue try: part = parse_cobertura(xml, canon) except ET.ParseError as exc: modules[name] = {"ok": False, "reason": "覆盖率 XML 解析失败: {}".format(exc), "files": {}, "lines_total": 0} print("[ENV] {} XML 解析失败: {}".format(name, exc), file=sys.stderr) continue modules[name] = { "ok": True, "files": {rel: len(e["lines"]) for rel, e in part.items()}, "lines_total": sum(len(e["lines"]) for e in part.values()), } merge_into(total_files, part) files = summarize(total_files) tested = sorted(files) untested, generated = [], [] for rel in prod: if rel in files: continue if is_generated(rel): generated.append(rel) continue untested.append({"path": rel, "code_lines": count_code_lines(rel)}) untested.sort(key=lambda u: (-u["code_lines"], u["path"])) totals = { "lines_valid": sum(s["lines_valid"] for s in files.values()), "lines_covered": sum(s["lines_covered"] for s in files.values()), "funcs_valid": sum(s["funcs_valid"] for s in files.values()), "funcs_covered": sum(s["funcs_covered"] for s in files.values()), "modules_total": len(modules), "modules_ok": sum(1 for m in modules.values() if m["ok"]), "files_tested": len(files), "files_untested": len(untested), "files_generated_excluded": len(generated), "line_rate": 0.0, "func_rate": 0.0, } if totals["lines_valid"]: totals["line_rate"] = round(totals["lines_covered"] / totals["lines_valid"], 4) if totals["funcs_valid"]: totals["func_rate"] = round(totals["funcs_covered"] / totals["funcs_valid"], 4) if args.json: payload = {"totals": totals, "modules": modules, "files": files, "untested_sources": untested, "generated_excluded": generated} Path(args.json).write_text( json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", newline="\n") baseline = load_baseline(args.baseline) if args.build_only: regress, notes = [], ["--build-only：只编译覆盖率形态（带 PDB），不采集数据"] else: regress, notes = diff_baseline(baseline, modules, files, totals, full_run) print_report(modules, files, totals, untested, generated, notes, regress, args.verbose) if args.md: write_markdown(args.md, modules, files, totals, untested, notes, regress) print("[OK] Markdown 报告: {}".format(args.md)) if args.update_baseline: write_baseline(args.baseline, modules, files, totals, tested, untested) print("[OK] 基线已写入 {}".format(os.path.relpath(args.baseline, REPO_ROOT))) return 0 if totals["modules_ok"] != totals["modules_total"]: return 1 return 1 if regress else 0 def main(): ap = argparse.ArgumentParser( prog="check_test_coverage.py", description="主机侧纯逻辑单测覆盖率：插桩跑测试 + 统计自研生产源码行/函数覆盖 + 基线增量门禁") ap.add_argument("--module", action="append", metavar="NAME", help="只处理指定模块（可重复；默认全部）") ap.add_argument("--list", action="store_true", help="只列出发现的模块名") ap.add_argument("--baseline", default=str(BASELINE_PATH), help="基线 JSON 路径（默认 scripts/test_coverage_baseline.json）") ap.add_argument("--update-baseline", action="store_true", help="把本次结果写回基线") ap.add_argument("--md", metavar="PATH", help="额外输出 Markdown 报告") ap.add_argument("--json", metavar="PATH", help="额外输出 JSON 明细") ap.add_argument("--work-dir", metavar="PATH", help="中间产物目录（默认与主机单测同家：TEMP 下 dtu_host/<工作树哈希>/coverage）") ap.add_argument("--clean", action="store_true", help="跑前清空中间目录") ap.add_argument("--timeout", type=int, default=600, metavar="SEC", help="单模块插桩运行超时（默认 600 秒）") ap.add_argument("--build-only", action="store_true", help="只按覆盖率开关编译、不采集（排障用）") ap.add_argument("--verbose", action="store_true", help="多打探测过程与未触及文件全表") args = ap.parse_args() if args.timeout <= 0: print("[USAGE] --timeout 必须为正整数", file=sys.stderr) return 3 try: return run(args) except KeyboardInterrupt: print("[ABORT] 用户中断", file=sys.stderr) return 2 BT_MARK = chr(96) if __name__ == "__main__": sys.exit(main())