"""对 Application 与 Domain 维护不可退化且及时固化的行覆盖率低水位。""" from __future__ import annotations import argparse import json import math from collections.abc import Mapping from pathlib import Path from typing import Any PROJECT_ROOT = Path(__file__).resolve().parents[2] DEFAULT_REPORT = PROJECT_ROOT / "coverage.json" DEFAULT_BASELINE = PROJECT_ROOT / "tests/fixtures/architecture/coverage-baseline.json" PACKAGE_PREFIXES = { "application": "app/application/", "domain": "app/domain/", } LEGACY_ZERO_BASELINE = { name: {"statements": 0, "covered_lines": 0, "percent": 0.0} for name in PACKAGE_PREFIXES } def _non_negative_int(value: object, *, field: str) -> int: """读取 coverage 整数计数，拒绝 bool、字符串和负值。""" if type(value) is not int or value < 0: raise ValueError(f"{field} 必须是非负整数") return value def collect_package_coverage(report: dict[str, Any]) -> dict[str, dict[str, int | float]]: """按治理包聚合 coverage.py JSON 中的语句和已覆盖行。""" if not isinstance(report, Mapping): raise ValueError("coverage JSON 顶层必须是对象") files = report.get("files") if not isinstance(files, Mapping): raise ValueError("coverage JSON 缺少 files 对象") result: dict[str, dict[str, int | float]] = {} for name, prefix in PACKAGE_PREFIXES.items(): statements = 0 covered = 0 for path, details in files.items(): if not isinstance(path, str): raise ValueError("coverage files 键必须是字符串路径") if not path.replace("\\", "/").startswith(prefix): continue if not isinstance(details, Mapping): raise ValueError(f"{path}: coverage 文件详情必须是对象") summary = details.get("summary") if not isinstance(summary, Mapping): raise ValueError(f"{path}: coverage 文件详情缺少 summary 对象") file_statements = _non_negative_int( summary.get("num_statements"), field=f"{path}.num_statements", ) file_covered = _non_negative_int( summary.get("covered_lines"), field=f"{path}.covered_lines", ) if file_covered > file_statements: raise ValueError( f"{path}: 已覆盖行数越界 {file_covered}/{file_statements}" ) statements += file_statements covered += file_covered percent = round(covered * 100 / statements, 2) if statements else 0.0 result[name] = { "statements": statements, "covered_lines": covered, "percent": percent, } return result def validate_coverage( current: object, ) -> list[str]: """拒绝缺包、额外包、零快照和不一致的派生百分比。""" problems: list[str] = [] if not isinstance(current, Mapping): return ["覆盖率快照必须是对象"] expected_names = set(PACKAGE_PREFIXES) actual_names = set(current) for name in sorted(expected_names - actual_names): problems.append(f"{name}: 覆盖率报告缺少治理包") for name in sorted(actual_names - expected_names): problems.append(f"{name}: 覆盖率快照包含未知治理包") for name in PACKAGE_PREFIXES: values = current.get(name) if values is None: continue if not isinstance(values, Mapping): problems.append(f"{name}: 覆盖率数据必须是对象") continue if set(values) != {"statements", "covered_lines", "percent"}: problems.append(f"{name}: 覆盖率数据字段不完整或包含未知字段") continue try: statements = _non_negative_int( values.get("statements"), field=f"{name}.statements", ) covered = _non_negative_int( values.get("covered_lines"), field=f"{name}.covered_lines", ) except ValueError as error: problems.append(str(error)) continue percent = values.get("percent") if ( isinstance(percent, bool) or not isinstance(percent, (int, float)) or not math.isfinite(float(percent)) ): problems.append(f"{name}.percent 必须是有限数值") continue if statements <= 0: problems.append(f"{name}: 覆盖率报告语句数必须大于 0") continue if not 0 <= covered <= statements: problems.append( f"{name}: 已覆盖行数越界 {covered}/{statements}" ) continue if covered == 0: problems.append(f"{name}: 已覆盖行数必须大于 0") expected_percent = round(covered * 100 / statements, 2) if float(percent) != expected_percent: problems.append( f"{name}: percent 与计数不一致 {float(percent):.2f}!={expected_percent:.2f}" ) return problems def is_legacy_zero_baseline(baseline: object) -> bool: """识别本批次之前唯一允许被初始化替换的全零 fixture。""" return baseline == LEGACY_ZERO_BASELINE def classify_coverage( baseline: dict[str, dict[str, int | float]], current: dict[str, dict[str, int | float]], ) -> tuple[list[str], list[str]]: """把覆盖率差异分为不可写入的回退和可固化的新低水位。""" regressions: list[str] = [] stale: list[str] = [] for name in PACKAGE_PREFIXES: expected_values = baseline.get(name, {}) actual_values = current[name] expected = float(expected_values.get("percent", 0.0)) actual = float(actual_values["percent"]) expected_statements = int(expected_values.get("statements", 0)) expected_covered = int(expected_values.get("covered_lines", 0)) actual_statements = int(actual_values["statements"]) actual_covered = int(actual_values["covered_lines"]) if ( expected_statements > 0 and actual_covered * expected_statements < expected_covered * actual_statements ): regressions.append( f"{name}: 行覆盖率下降 {expected:.2f}%->{actual:.2f}%" ) elif actual_values != expected_values: stale.append( f"{name}: 覆盖率低水位未固化 {expected:.2f}%->{actual:.2f}%" ) return regressions, stale def compare_coverage( baseline: dict[str, dict[str, int | float]], current: dict[str, dict[str, int | float]], ) -> list[str]: """返回覆盖率回退和尚未固化的新低水位。""" regressions, stale = classify_coverage(baseline, current) return [*regressions, *stale] def main() -> int: """检查 coverage JSON，或显式刷新当前阈值。""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--write", action="store_true", help="写入当前覆盖率阈值") parser.add_argument("--report", type=Path, default=DEFAULT_REPORT) parser.add_argument("--baseline", type=Path, default=DEFAULT_BASELINE) args = parser.parse_args() try: report = json.loads(args.report.read_text(encoding="utf-8")) current = collect_package_coverage(report) except (OSError, ValueError, TypeError) as error: print(f"Coverage 报告无效：{error}") return 1 validation_problems = validate_coverage(current) if validation_problems: print("\n".join(validation_problems)) return 1 baseline_exists = args.baseline.exists() if baseline_exists: try: baseline = json.loads(args.baseline.read_text(encoding="utf-8")) except (OSError, ValueError, TypeError) as error: print(f"Coverage 基线无效：{error}") return 1 if not is_legacy_zero_baseline(baseline): baseline_problems = validate_coverage(baseline) if baseline_problems: print("Coverage 基线无效：") print("\n".join(baseline_problems)) return 1 else: baseline = {} regressions, stale = classify_coverage(baseline, current) if args.write: if baseline_exists and regressions: print("\n".join(regressions)) print("拒绝写入：当前结果包含覆盖率回退，--write 只能固化持平或提升后的低水位。") return 1 args.baseline.parent.mkdir(parents=True, exist_ok=True) args.baseline.write_text( json.dumps(current, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) display_path = ( args.baseline.relative_to(PROJECT_ROOT) if args.baseline.is_relative_to(PROJECT_ROOT) else args.baseline ) print(f"已写入 {display_path}") return 0 problems = [*regressions, *stale] if problems: print("\n".join(problems)) if regressions: print("先消除覆盖率回退；存在下降时禁止用 --write 覆盖基线。") else: print("提示：当前只有覆盖率持平快照变化或提升，可用 --write 固化新的低水位。") return 1 summary = ", ".join( f"{name}={values['percent']:.2f}%" for name, values in current.items() ) print(f"覆盖率 ratchet 通过（低水位已同步：{summary}）") return 0 if __name__ == "__main__": raise SystemExit(main())