fix(quality): close coverage ratchet

scripts/architecture/coverage_ratchet.py

@@ -1,9 +1,11 @@
-"""对 Application 与 Domain 维护不可退化的行覆盖率阈值。"""
+"""对 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
 
@@ -14,22 +16,55 @@ 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]] = {}
-    files = report.get("files", {})
     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
-            summary = details["summary"]
-            statements += int(summary["num_statements"])
-            covered += int(summary["covered_lines"])
-        percent = round(covered * 100 / statements, 2) if statements else 100.0
+            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,
@@ -38,18 +73,110 @@ def collect_package_coverage(report: dict[str, Any]) -> dict[str, dict[str, int
     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]:
-    """返回包覆盖率低于已提交阈值的问题。"""
-    problems = []
-    for name in PACKAGE_PREFIXES:
-        expected = float(baseline[name]["percent"])
-        actual = float(current[name]["percent"])
-        if actual < expected:
-            problems.append(f"{name}: 行覆盖率下降 {expected:.2f}%->{actual:.2f}%")
-    return problems
+    """返回覆盖率回退和尚未固化的新低水位。"""
+    regressions, stale = classify_coverage(baseline, current)
+    return [*regressions, *stale]
 
 
 def main() -> int:
@@ -59,25 +186,61 @@ def main() -> int:
     parser.add_argument("--report", type=Path, default=DEFAULT_REPORT)
     parser.add_argument("--baseline", type=Path, default=DEFAULT_BASELINE)
     args = parser.parse_args()
-    report = json.loads(args.report.read_text(encoding="utf-8"))
-    current = collect_package_coverage(report)
+    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",
         )
-        print(f"已写入 {args.baseline.relative_to(PROJECT_ROOT)}")
+        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
-    baseline = json.loads(args.baseline.read_text(encoding="utf-8"))
-    problems = compare_coverage(baseline, current)
+    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}）")
+    print(f"覆盖率 ratchet 通过（低水位已同步：{summary}）")
     return 0