mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-28 19:06:36 +08:00
d51320103c
* fix: make local setup and bundled skill operations recoverable * test: verify user updates with real transaction runners
435 lines
16 KiB
Python
435 lines
16 KiB
Python
"""
|
||
检查除官方插件外全部 app Python 文件的覆盖率低水位。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import math
|
||
import os
|
||
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"
|
||
# 第 2 版覆盖率快照以逐文件计数为准，同时保留包级和全量摘要。
|
||
SCHEMA_VERSION = 2
|
||
SCOPE_ROOT = "app"
|
||
EXCLUDED_PREFIXES = ("app/plugins/",)
|
||
# 新受治理文件必须覆盖至少 80% 的可执行语句。
|
||
NEW_FILE_COVERAGE_PERCENT = 80.0
|
||
_METRIC_FIELDS = {"statements", "covered_lines", "percent"}
|
||
|
||
|
||
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 _normalise_path(path: str) -> str:
|
||
"""
|
||
将 coverage.py 路径中的反斜杠统一为 POSIX 分隔符。
|
||
"""
|
||
return path.replace("\\", "/")
|
||
|
||
|
||
def _is_governed_path(path: str) -> bool:
|
||
"""
|
||
判断路径是否属于 app 且不在官方插件目录。
|
||
"""
|
||
normalised = _normalise_path(path)
|
||
return (
|
||
normalised.startswith("app/")
|
||
and not any(
|
||
normalised == prefix.rstrip("/") or normalised.startswith(prefix)
|
||
for prefix in EXCLUDED_PREFIXES
|
||
)
|
||
and normalised.endswith(".py")
|
||
)
|
||
|
||
|
||
def _metric(statements: int, covered: int) -> dict[str, int | float]:
|
||
"""
|
||
根据 coverage.py 计数构造稳定的逐文件覆盖率记录。
|
||
"""
|
||
percent = round(covered * 100 / statements, 2) if statements else 0.0
|
||
return {"statements": statements, "covered_lines": covered, "percent": percent}
|
||
|
||
|
||
def _validate_metric(
|
||
values: object, *, name: str, allow_zero: bool = True
|
||
) -> list[str]:
|
||
"""
|
||
校验一条覆盖率记录的字段、计数边界和派生百分比。
|
||
"""
|
||
problems: list[str] = []
|
||
if not isinstance(values, Mapping):
|
||
return [f"{name}: 覆盖率数据必须是对象"]
|
||
if set(values) != _METRIC_FIELDS:
|
||
return [f"{name}: 覆盖率数据字段不完整或包含未知字段"]
|
||
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:
|
||
return [str(error)]
|
||
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 必须是有限数值")
|
||
return problems
|
||
if not 0 <= covered <= statements:
|
||
problems.append(f"{name}: 已覆盖行数越界 {covered}/{statements}")
|
||
if not allow_zero and statements <= 0:
|
||
problems.append(f"{name}: 覆盖率报告语句数必须大于 0")
|
||
expected_percent = round(covered * 100 / statements, 2) if statements else 0.0
|
||
if float(percent) != expected_percent:
|
||
problems.append(
|
||
f"{name}: percent 与计数不一致 {float(percent):.2f}!={expected_percent:.2f}"
|
||
)
|
||
return problems
|
||
|
||
|
||
def _aggregate(
|
||
metrics: Mapping[str, Mapping[str, int | float]],
|
||
) -> dict[str, int | float]:
|
||
"""
|
||
按真实语句数聚合逐文件覆盖率，避免平均百分比失真。
|
||
"""
|
||
statements = sum(int(values["statements"]) for values in metrics.values())
|
||
covered = sum(int(values["covered_lines"]) for values in metrics.values())
|
||
return _metric(statements, covered)
|
||
|
||
|
||
def _package_name(path: str) -> str:
|
||
"""
|
||
返回 app 文件对应的一级包名。
|
||
"""
|
||
relative = path.removeprefix("app/")
|
||
return relative.split("/", 1)[0] if "/" in relative else "__root__"
|
||
|
||
|
||
def _snapshot_from_files(
|
||
files: dict[str, dict[str, int | float]],
|
||
) -> dict[str, Any]:
|
||
"""
|
||
从逐文件记录构造带包级和全量摘要的快照。
|
||
"""
|
||
package_files: dict[str, dict[str, dict[str, int | float]]] = {}
|
||
for path, values in files.items():
|
||
package_files.setdefault(_package_name(path), {})[path] = values
|
||
packages = {
|
||
name: _aggregate(files_for_package)
|
||
for name, files_for_package in sorted(package_files.items())
|
||
}
|
||
return {
|
||
"schema_version": SCHEMA_VERSION,
|
||
"scope": {"root": SCOPE_ROOT, "excluded": ["app/plugins"]},
|
||
"files": dict(sorted(files.items())),
|
||
"packages": packages,
|
||
"total": _aggregate(files),
|
||
}
|
||
|
||
|
||
def collect_package_coverage(report: dict[str, Any]) -> dict[str, Any]:
|
||
"""
|
||
收集全部受治理文件，并附带一级包与全量聚合摘要。
|
||
|
||
:param report: coverage.py 导出的 JSON 报告，必须含有效的 files 对象
|
||
:return: 按文件、一级包和全部源码汇总的覆盖率快照
|
||
:raises ValueError: 报告结构、文件路径或计数不符合要求
|
||
"""
|
||
if not isinstance(report, Mapping):
|
||
raise ValueError("coverage JSON 顶层必须是对象")
|
||
files = report.get("files")
|
||
if not isinstance(files, Mapping):
|
||
raise ValueError("coverage JSON 缺少 files 对象")
|
||
collected: dict[str, dict[str, int | float]] = {}
|
||
for raw_path, details in files.items():
|
||
if not isinstance(raw_path, str):
|
||
raise ValueError("coverage files 键必须是字符串路径")
|
||
path = _normalise_path(raw_path)
|
||
if not _is_governed_path(path):
|
||
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 对象")
|
||
statements = _non_negative_int(
|
||
summary.get("num_statements"), field=f"{path}.num_statements"
|
||
)
|
||
covered = _non_negative_int(
|
||
summary.get("covered_lines"), field=f"{path}.covered_lines"
|
||
)
|
||
if covered > statements:
|
||
raise ValueError(f"{path}: 已覆盖行数越界 {covered}/{statements}")
|
||
collected[path] = _metric(statements, covered)
|
||
|
||
return _snapshot_from_files(collected)
|
||
|
||
|
||
def validate_coverage(current: object, *, require_data: bool = False) -> list[str]:
|
||
"""
|
||
拒绝超出治理范围、字段不完整或计数与百分比不一致的快照。
|
||
|
||
:param current: 待检查的覆盖率快照
|
||
:param require_data: 是否拒绝空快照或没有可执行语句的数据
|
||
:return: 格式与计数错误列表；空列表表示有效
|
||
"""
|
||
problems: list[str] = []
|
||
if not isinstance(current, Mapping):
|
||
return ["覆盖率快照必须是对象"]
|
||
if current.get("schema_version") != SCHEMA_VERSION:
|
||
problems.append(f"覆盖率快照 schema_version 必须是 {SCHEMA_VERSION}")
|
||
if current.get("scope") != {"root": SCOPE_ROOT, "excluded": ["app/plugins"]}:
|
||
problems.append("覆盖率快照 scope 必须覆盖 app 并排除 app/plugins")
|
||
files = current.get("files")
|
||
if not isinstance(files, Mapping):
|
||
problems.append("覆盖率快照缺少 files 对象")
|
||
return problems
|
||
if require_data and not files:
|
||
problems.append("覆盖率报告缺少受治理文件")
|
||
for path, values in files.items():
|
||
if not isinstance(path, str):
|
||
problems.append("覆盖率快照 files 键必须是字符串路径")
|
||
continue
|
||
if not _is_governed_path(path):
|
||
problems.append(f"{path}: 路径不在覆盖率治理范围")
|
||
continue
|
||
problems.extend(_validate_metric(values, name=path))
|
||
packages = current.get("packages")
|
||
if not isinstance(packages, Mapping):
|
||
problems.append("覆盖率快照缺少 packages 对象")
|
||
else:
|
||
for name, values in packages.items():
|
||
if not isinstance(name, str):
|
||
problems.append("packages 键必须是字符串")
|
||
continue
|
||
problems.extend(_validate_metric(values, name=f"packages.{name}"))
|
||
total = current.get("total")
|
||
problems.extend(_validate_metric(total, name="total"))
|
||
if require_data and isinstance(total, Mapping) and total.get("statements") == 0:
|
||
problems.append("覆盖率报告语句数必须大于 0")
|
||
if isinstance(files, Mapping) and isinstance(packages, Mapping) and isinstance(total, Mapping):
|
||
expected_packages: dict[str, dict[str, Mapping[str, int | float]]] = {}
|
||
for path, values in files.items():
|
||
if not isinstance(path, str) or not isinstance(values, Mapping):
|
||
continue
|
||
expected_packages.setdefault(_package_name(path), {})[path] = values
|
||
for name in sorted(set(packages) - set(expected_packages)):
|
||
problems.append(f"packages.{name}: 包级摘要没有对应文件")
|
||
for name, values in expected_packages.items():
|
||
if packages.get(name) != _aggregate(values):
|
||
problems.append(f"packages.{name}: 聚合结果与 files 不一致")
|
||
if dict(total) != _aggregate(files):
|
||
problems.append("total: 聚合结果与 files 不一致")
|
||
return problems
|
||
|
||
|
||
def _file_ratio_is_lower(
|
||
previous: Mapping[str, int | float], current: Mapping[str, int | float]
|
||
) -> bool:
|
||
"""
|
||
使用真实计数比较两个文件的覆盖率，避免四舍五入误判。
|
||
"""
|
||
previous_statements = int(previous["statements"])
|
||
current_statements = int(current["statements"])
|
||
if previous_statements == 0:
|
||
return False
|
||
if current_statements == 0:
|
||
return True
|
||
return int(current["covered_lines"]) * previous_statements < int(
|
||
previous["covered_lines"]
|
||
) * current_statements
|
||
|
||
|
||
def _fewer_statements_without_new_misses(
|
||
previous: Mapping[str, int | float], current: Mapping[str, int | float]
|
||
) -> bool:
|
||
"""
|
||
允许语句减少且未覆盖行数不增加造成的比例下降。
|
||
"""
|
||
previous_statements = int(previous["statements"])
|
||
current_statements = int(current["statements"])
|
||
return (
|
||
current_statements < previous_statements
|
||
and current_statements - int(current["covered_lines"])
|
||
<= previous_statements - int(previous["covered_lines"])
|
||
)
|
||
|
||
|
||
def classify_coverage(
|
||
baseline: Mapping[str, Any], current: Mapping[str, Any]
|
||
) -> tuple[list[str], list[str]]:
|
||
"""
|
||
拒绝存量文件新增未覆盖行造成的回退，并要求新文件达到 80%。
|
||
|
||
:param baseline: 已确认的逐文件覆盖率低水位
|
||
:param current: 当前逐文件覆盖率快照
|
||
:return: 回退问题与可提升低水位的条目两个列表
|
||
"""
|
||
regressions: list[str] = []
|
||
improvements: list[str] = []
|
||
baseline_files = baseline.get("files", {})
|
||
current_files = current.get("files", {})
|
||
if not isinstance(baseline_files, Mapping) or not isinstance(current_files, Mapping):
|
||
return ["Coverage 基线或当前报告缺少 files 对象"], []
|
||
for path, values in current_files.items():
|
||
previous = baseline_files.get(path)
|
||
if previous is None:
|
||
statements = int(values["statements"])
|
||
covered = int(values["covered_lines"])
|
||
if covered * 100 < NEW_FILE_COVERAGE_PERCENT * statements:
|
||
regressions.append(
|
||
f"{path}: 新文件覆盖率低于 {NEW_FILE_COVERAGE_PERCENT:.0f}%"
|
||
f"->{float(values['percent']):.2f}%"
|
||
)
|
||
else:
|
||
improvements.append(f"{path}: 新文件达到覆盖率门槛")
|
||
elif _file_ratio_is_lower(previous, values) and not _fewer_statements_without_new_misses(
|
||
previous, values
|
||
):
|
||
regressions.append(
|
||
f"{path}: 存量覆盖率下降 "
|
||
f"{float(previous['percent']):.2f}%->{float(values['percent']):.2f}%"
|
||
)
|
||
elif _file_ratio_is_lower(values, previous):
|
||
improvements.append(f"{path}: 覆盖率上升")
|
||
return regressions, improvements
|
||
|
||
|
||
def compare_coverage(
|
||
baseline: Mapping[str, Any], current: Mapping[str, Any]
|
||
) -> list[str]:
|
||
"""
|
||
返回存量覆盖率回退或新文件低于 80% 的问题。
|
||
"""
|
||
regressions, _ = classify_coverage(baseline, current)
|
||
return regressions
|
||
|
||
|
||
def _merge_baseline(
|
||
baseline: Mapping[str, Any], current: Mapping[str, Any]
|
||
) -> dict[str, Any]:
|
||
"""
|
||
使用同一份当前计数生成逐文件和汇总低水位；调用方已拒绝覆盖率回退。
|
||
"""
|
||
regressions = compare_coverage(baseline, current)
|
||
if regressions:
|
||
raise ValueError("拒绝写入覆盖率回退：" + "; ".join(regressions))
|
||
return _snapshot_from_files(current["files"])
|
||
|
||
|
||
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
|
||
if args.report.resolve() == DEFAULT_REPORT.resolve():
|
||
source_paths = {
|
||
path.relative_to(PROJECT_ROOT).as_posix()
|
||
for path in (PROJECT_ROOT / SCOPE_ROOT).rglob("*.py")
|
||
if path.relative_to(PROJECT_ROOT).parts[:2] != ("app", "plugins")
|
||
}
|
||
missing_source_paths = source_paths - set(current["files"])
|
||
stale_report_paths = set(current["files"]) - source_paths
|
||
if os.environ.get("CI", "").lower() == "true" and (
|
||
missing_source_paths or stale_report_paths
|
||
):
|
||
print(
|
||
"Coverage 报告未完整覆盖当前 app 源码，"
|
||
"CI 只接受 canonical 全量报告。"
|
||
)
|
||
return 1
|
||
if args.write and (missing_source_paths or stale_report_paths):
|
||
print(
|
||
"拒绝写入：默认 coverage 报告未完整覆盖当前 app 源码，"
|
||
"请使用 canonical 全量报告。"
|
||
)
|
||
return 1
|
||
current_files = {
|
||
path: values
|
||
for path, values in current["files"].items()
|
||
if path in source_paths
|
||
}
|
||
current = _snapshot_from_files(current_files)
|
||
validation_problems = validate_coverage(current, require_data=True)
|
||
if validation_problems:
|
||
print("\n".join(validation_problems))
|
||
return 1
|
||
|
||
if args.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
|
||
baseline_problems = validate_coverage(baseline)
|
||
if baseline_problems:
|
||
print("Coverage 基线无效：")
|
||
print("\n".join(baseline_problems))
|
||
return 1
|
||
else:
|
||
baseline = {
|
||
"schema_version": SCHEMA_VERSION,
|
||
"scope": {"root": SCOPE_ROOT, "excluded": ["app/plugins"]},
|
||
"files": {},
|
||
}
|
||
|
||
regressions, _ = classify_coverage(baseline, current)
|
||
if regressions:
|
||
print("\n".join(regressions))
|
||
if args.write:
|
||
print("拒绝写入：覆盖率低于既有低水位或新文件 80% 门槛。")
|
||
return 1
|
||
if args.write:
|
||
args.baseline.parent.mkdir(parents=True, exist_ok=True)
|
||
args.baseline.write_text(
|
||
json.dumps(
|
||
_merge_baseline(baseline, 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
|
||
total = current["total"]
|
||
print(
|
||
"覆盖率 ratchet 通过（全 app 排除 plugins："
|
||
f"{total['percent']:.2f}%，文件 {len(current['files'])} 个）"
|
||
)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|