361 lines
14 KiB
Python
361 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""默认测试套件：课程成绩异常诊断与报告生成。
|
||
|
||
设计原则：
|
||
- 仅用标准库（不依赖 pytest / jinja2），CI 与本地均能直接 `python tests/run_tests.py` 跑
|
||
- v1.0.1 是骨架版（仅 main.py 占位 + 空 __init__.py），v1.0.2 同步发布声明，本套件覆盖：
|
||
- util.constants 版本对齐（POLICY-RELEASE-001）
|
||
- SKILL.md / CHANGELOG.md 顶部 version / slug 一致性
|
||
- scripts/main.py CLI 路由表的命令完整性
|
||
- scripts/util/constants.py 阈值与等级常量自洽
|
||
- 退出码：0 = 全部通过，1 = 有失败
|
||
- -v 开关：详细输出每个用例的状态
|
||
|
||
release.ps1 在打 tag 前会自动跑 `python tests/run_tests.py -v`，任何一个
|
||
test_* 返回 (False, msg) 即 abort 发布。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any, Callable
|
||
|
||
# 让 scripts/ 下的模块可导入
|
||
_REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
_SCRIPTS_DIR = _REPO_ROOT / "scripts"
|
||
sys.path.insert(0, str(_SCRIPTS_DIR))
|
||
|
||
|
||
# ---------- 测试基础设施 ----------
|
||
|
||
_RESULTS: list[tuple[str, bool, str]] = []
|
||
_VERBOSE = False
|
||
|
||
|
||
def _run(name: str, fn: Callable[[], tuple[bool, str]]) -> None:
|
||
"""运行单个 test_*，记录结果。"""
|
||
import traceback
|
||
|
||
try:
|
||
ok, msg = fn()
|
||
except Exception as e:
|
||
tb = traceback.format_exc(limit=3)
|
||
ok, msg = False, f"exception: {e}\n{tb}"
|
||
_RESULTS.append((name, ok, msg))
|
||
if _VERBOSE or not ok:
|
||
marker = "PASS" if ok else "FAIL"
|
||
print(f" [{marker}] {name}: {msg}", flush=True)
|
||
|
||
|
||
# ---------- util.constants 版本对齐（POLICY-RELEASE-001） ----------
|
||
|
||
def test_skill_md_version_alignment() -> tuple[bool, str]:
|
||
"""POLICY-RELEASE-001：SKILL.md version == SKILL_VERSION == CHANGELOG 顶部版本"""
|
||
from util.constants import SKILL_VERSION
|
||
|
||
skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
||
m = re.search(r'(?m)^version:\s*["\']?(\d+\.\d+\.\d+)', skill_md)
|
||
if not m:
|
||
return False, "SKILL.md 缺顶层 version: X.Y.Z"
|
||
md_ver = m.group(1)
|
||
if md_ver != SKILL_VERSION:
|
||
return False, f"SKILL.md={md_ver} != SKILL_VERSION={SKILL_VERSION}"
|
||
|
||
cl = (_REPO_ROOT / "CHANGELOG.md").read_text(encoding="utf-8")
|
||
cl_m = re.search(r'(?m)^##\s+(\d+\.\d+\.\d+)', cl)
|
||
if not cl_m:
|
||
return False, "CHANGELOG.md 缺顶层 ## X.Y.Z"
|
||
cl_ver = cl_m.group(1)
|
||
if cl_ver != SKILL_VERSION:
|
||
return False, f"CHANGELOG 顶部={cl_ver} != SKILL_VERSION={SKILL_VERSION}"
|
||
|
||
return True, f"SKILL.md={md_ver} == CHANGELOG={cl_ver} == SKILL_VERSION={SKILL_VERSION}"
|
||
|
||
|
||
def test_skill_md_slug_alignment() -> tuple[bool, str]:
|
||
"""SKILL.md metadata.openclaw.slug == SKILL_SLUG"""
|
||
from util.constants import SKILL_SLUG
|
||
|
||
skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
||
m = re.search(r'(?m)^\s*slug:\s*["\']?([a-z0-9_-]+)', skill_md)
|
||
if not m:
|
||
return False, "SKILL.md 缺 metadata.openclaw.slug"
|
||
md_slug = m.group(1)
|
||
if md_slug != SKILL_SLUG:
|
||
return False, f"SKILL.md slug={md_slug} != SKILL_SLUG={SKILL_SLUG}"
|
||
return True, f"slug={md_slug}"
|
||
|
||
|
||
def test_skill_md_brands_declaration() -> tuple[bool, str]:
|
||
"""v1.0.2+ 要求 metadata.openclaw.brands 含 jiangchang + huashang"""
|
||
skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
|
||
# 定位 metadata.openclaw 块（顶层 key 之前都属 metadata 范围）
|
||
# 简单做法：先找 brands: 行，再往后扫 10 行，仅收录缩进 > 0 且以 - 开头的项
|
||
m = re.search(r'(?m)^[ \t]*brands:.*$', skill_md)
|
||
if not m:
|
||
return False, "SKILL.md metadata.openclaw 缺 brands: 声明"
|
||
after = skill_md[m.end():].splitlines()
|
||
# brands 列表项最长 2~3 行（jiangchang + huashang + 注释），扫 8 行足够
|
||
brands: set[str] = set()
|
||
for line in after[:8]:
|
||
# 列表项必须以 2+ 个空格 + - + 空格开头
|
||
item = re.match(r'^[ \t]{2,}-\s+([a-z0-9_-]+)\s*(?:#.*)?$', line)
|
||
if item:
|
||
brands.add(item.group(1))
|
||
continue
|
||
# 遇下一个非缩进/非 - 项则停止 brands 块
|
||
if line.strip() and not line.startswith(" ") and not line.startswith("\t"):
|
||
break
|
||
if "jiangchang" not in brands:
|
||
return False, f"brands 缺 jiangchang，实际 {sorted(brands)}"
|
||
if "huashang" not in brands:
|
||
return False, f"brands 缺 huashang，实际 {sorted(brands)}"
|
||
return True, f"brands={sorted(brands)}"
|
||
|
||
|
||
def test_changelog_has_v102_entry() -> tuple[bool, str]:
|
||
"""CHANGELOG.md 含 v1.0.2 条目"""
|
||
from util.constants import SKILL_VERSION
|
||
|
||
cl = (_REPO_ROOT / "CHANGELOG.md").read_text(encoding="utf-8")
|
||
if f"## {SKILL_VERSION}" not in cl:
|
||
return False, f"CHANGELOG 缺 ## {SKILL_VERSION} 段"
|
||
return True, f"## {SKILL_VERSION} 条目已存在"
|
||
|
||
|
||
# ---------- util.constants 阈值与等级自洽 ----------
|
||
|
||
def test_constants_anomaly_patterns() -> tuple[bool, str]:
|
||
"""ANOMALY_PATTERNS 含三种核心异常模式"""
|
||
from util.constants import ANOMALY_PATTERNS
|
||
|
||
expected = {"高分低能", "两极分化", "挂科异常"}
|
||
got = set(ANOMALY_PATTERNS)
|
||
if got != expected:
|
||
return False, f"ANOMALY_PATTERNS={sorted(got)}, want {sorted(expected)}"
|
||
return True, "3 种异常模式齐全"
|
||
|
||
|
||
def test_constants_alert_levels_order() -> tuple[bool, str]:
|
||
"""ALERT_LEVELS 与 ALERT_LEVEL_LABELS 顺序一一对应"""
|
||
from util.constants import ALERT_LEVELS, ALERT_LEVEL_LABELS
|
||
|
||
if tuple(ALERT_LEVEL_LABELS.keys()) != ALERT_LEVELS:
|
||
return False, (
|
||
f"ALERT_LEVEL_LABELS 键={tuple(ALERT_LEVEL_LABELS.keys())} "
|
||
f"!= ALERT_LEVELS={ALERT_LEVELS}"
|
||
)
|
||
for lv in ALERT_LEVELS:
|
||
if not ALERT_LEVEL_LABELS[lv]:
|
||
return False, f"等级 {lv} 标签为空"
|
||
return True, f"3 个等级 (yellow/orange/red) 标签齐全"
|
||
|
||
|
||
def test_constants_thresholds_positive() -> tuple[bool, str]:
|
||
"""核心阈值必须为正数，且大小关系合理"""
|
||
from util.constants import (
|
||
HIGH_SCORE_THRESHOLD,
|
||
LOW_SCORE_THRESHOLD,
|
||
HIGH_FAIL_RATE_THRESHOLD,
|
||
POLARIZATION_SPREAD_THRESHOLD,
|
||
)
|
||
|
||
if not (0 < HIGH_SCORE_THRESHOLD <= 100):
|
||
return False, f"HIGH_SCORE_THRESHOLD={HIGH_SCORE_THRESHOLD}, 应在 (0,100]"
|
||
if not (0 < LOW_SCORE_THRESHOLD <= 100):
|
||
return False, f"LOW_SCORE_THRESHOLD={LOW_SCORE_THRESHOLD}, 应在 (0,100]"
|
||
if LOW_SCORE_THRESHOLD >= HIGH_SCORE_THRESHOLD:
|
||
return False, f"低分阈值 {LOW_SCORE_THRESHOLD} 应 < 高分阈值 {HIGH_SCORE_THRESHOLD}"
|
||
if not (0 < HIGH_FAIL_RATE_THRESHOLD < 1):
|
||
return False, f"HIGH_FAIL_RATE_THRESHOLD={HIGH_FAIL_RATE_THRESHOLD}, 应在 (0,1)"
|
||
if POLARIZATION_SPREAD_THRESHOLD <= 0:
|
||
return False, f"POLARIZATION_SPREAD_THRESHOLD={POLARIZATION_SPREAD_THRESHOLD}, 应 > 0"
|
||
return True, "4 个阈值合规"
|
||
|
||
|
||
def test_constants_platform_kit_min_version() -> tuple[bool, str]:
|
||
"""PLATFORM_KIT_MIN_VERSION 必须形如 X.Y.Z"""
|
||
from util.constants import PLATFORM_KIT_MIN_VERSION
|
||
|
||
if not re.match(r'^\d+\.\d+\.\d+$', PLATFORM_KIT_MIN_VERSION):
|
||
return False, f"PLATFORM_KIT_MIN_VERSION={PLATFORM_KIT_MIN_VERSION} 不是 X.Y.Z"
|
||
return True, f"PLATFORM_KIT_MIN_VERSION={PLATFORM_KIT_MIN_VERSION}"
|
||
|
||
|
||
def test_constants_high_risk_labels_consistent() -> tuple[bool, str]:
|
||
"""HIGH_RISK_LABELS 同时含 True / False 两个键"""
|
||
from util.constants import HIGH_RISK_LABELS
|
||
|
||
if True not in HIGH_RISK_LABELS or False not in HIGH_RISK_LABELS:
|
||
return False, f"HIGH_RISK_LABELS 应含 True/False 两个键，实测 {list(HIGH_RISK_LABELS.keys())}"
|
||
return True, "True/False 双键齐全"
|
||
|
||
|
||
# ---------- 核心 repository / 业务骨架自检 ----------
|
||
|
||
def test_main_py_commands_complete() -> tuple[bool, str]:
|
||
"""scripts/main.py COMMANDS 含 9 个核心命令"""
|
||
expected = {
|
||
"health", "version", "config-path", "init-db",
|
||
"analyze-grade-distribution", "batch-analyze-courses",
|
||
"render-html", "flag-high-risk-course", "sync-historical-grades",
|
||
}
|
||
import importlib.util
|
||
|
||
spec = importlib.util.spec_from_file_location("_main", _SCRIPTS_DIR / "main.py")
|
||
assert spec and spec.loader
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
got = set(mod.COMMANDS.keys())
|
||
missing = expected - got
|
||
extra = got - expected
|
||
if missing or extra:
|
||
return False, f"missing={sorted(missing)} extra={sorted(extra)}"
|
||
return True, f"COMMANDS 含全部 {len(expected)} 个命令"
|
||
|
||
|
||
def test_main_py_version_command_prints_version() -> tuple[bool, str]:
|
||
"""version 命令打印的版本号必须 == SKILL_VERSION"""
|
||
from util.constants import SKILL_VERSION
|
||
|
||
import io
|
||
import contextlib
|
||
import importlib.util
|
||
|
||
spec = importlib.util.spec_from_file_location("_main", _SCRIPTS_DIR / "main.py")
|
||
assert spec and spec.loader
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
|
||
buf = io.StringIO()
|
||
with contextlib.redirect_stdout(buf):
|
||
rc = mod.version()
|
||
out = buf.getvalue()
|
||
if rc != 0:
|
||
return False, f"version() 返回 {rc}, want 0"
|
||
if SKILL_VERSION not in out:
|
||
return False, f"version 输出不含 {SKILL_VERSION}，实际 {out!r}"
|
||
return True, f"version 命令打印含 {SKILL_VERSION}"
|
||
|
||
|
||
def test_main_py_analyze_command_does_not_crash() -> tuple[bool, str]:
|
||
"""analyze-grade-distribution 接受 course_id 占位，不抛异常"""
|
||
import importlib.util
|
||
|
||
spec = importlib.util.spec_from_file_location("_main", _SCRIPTS_DIR / "main.py")
|
||
assert spec and spec.loader
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
|
||
rc = mod.analyze_grade_distribution("COURSE-TEST-001")
|
||
if rc != 0:
|
||
return False, f"analyze_grade_distribution 返回 {rc}, want 0"
|
||
return True, "analyze-grade-distribution 占位命令可调用"
|
||
|
||
|
||
def test_main_py_main_dispatch() -> tuple[bool, str]:
|
||
"""main(argv) 能正确分发到 version 命令"""
|
||
import importlib.util
|
||
|
||
spec = importlib.util.spec_from_file_location("_main", _SCRIPTS_DIR / "main.py")
|
||
assert spec and spec.loader
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
|
||
rc = mod.main(["version"])
|
||
if rc != 0:
|
||
return False, f"main(['version']) 返回 {rc}, want 0"
|
||
rc = mod.main(["__nope__"])
|
||
if rc != 1:
|
||
return False, f"main(['__nope__']) 返回 {rc}, want 1（未知命令）"
|
||
return True, "argv 分发 OK（version=0 / 未知命令=1）"
|
||
|
||
|
||
# ---------- 服务层 / 异常识别占位自检 ----------
|
||
|
||
def test_anomaly_classification_thresholds_consistent_with_constants() -> tuple[bool, str]:
|
||
"""若未来引入 service 模块，分类阈值必须与 constants 一致。
|
||
v1.0.2 仍为骨架，本测试仅校验 constants 中阈值即可，作为常量化的契约保护。
|
||
"""
|
||
from util.constants import (
|
||
HIGH_SCORE_THRESHOLD,
|
||
LOW_SCORE_THRESHOLD,
|
||
HIGH_FAIL_RATE_THRESHOLD,
|
||
POLARIZATION_SPREAD_THRESHOLD,
|
||
)
|
||
|
||
# 高分线必须显著高于低分线（差距 ≥ 10 分）才有「高分低能」识别空间
|
||
if HIGH_SCORE_THRESHOLD - LOW_SCORE_THRESHOLD < 10:
|
||
return False, (
|
||
f"高分线 {HIGH_SCORE_THRESHOLD} 与低分线 {LOW_SCORE_THRESHOLD} 差距过小，"
|
||
f"无法区分「高分低能」与「两极分化」"
|
||
)
|
||
# 挂科率阈值 ≥ 0.1 才有意义
|
||
if HIGH_FAIL_RATE_THRESHOLD < 0.1:
|
||
return False, f"挂科率阈值 {HIGH_FAIL_RATE_THRESHOLD} 过低，噪声大"
|
||
# 两极分化极差阈值 ≥ 20 分
|
||
if POLARIZATION_SPREAD_THRESHOLD < 20:
|
||
return False, f"两极分化极差阈值 {POLARIZATION_SPREAD_THRESHOLD} 过小"
|
||
return True, "4 个阈值满足业务识别能力"
|
||
|
||
|
||
def test_scripts_subdirs_present() -> tuple[bool, str]:
|
||
"""scripts/{cli,db,service,util} 四个分层目录必须存在"""
|
||
for sub in ("cli", "db", "service", "util"):
|
||
p = _SCRIPTS_DIR / sub
|
||
if not p.is_dir():
|
||
return False, f"scripts/{sub} 目录不存在"
|
||
return True, "cli/db/service/util 四个分层齐全"
|
||
|
||
|
||
# ---------- 入口 ----------
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="默认 release 测试套件")
|
||
parser.add_argument("-v", "--verbose", action="store_true",
|
||
help="详细输出每个用例结果")
|
||
args = parser.parse_args()
|
||
|
||
global _VERBOSE
|
||
_VERBOSE = args.verbose
|
||
|
||
print("=== analyze-grade-distribution-aa release tests ===", flush=True)
|
||
|
||
_run("constants/skill_md_version_alignment", test_skill_md_version_alignment)
|
||
_run("constants/skill_md_slug_alignment", test_skill_md_slug_alignment)
|
||
_run("constants/skill_md_brands_declaration", test_skill_md_brands_declaration)
|
||
_run("constants/changelog_v102_entry", test_changelog_has_v102_entry)
|
||
_run("constants/anomaly_patterns", test_constants_anomaly_patterns)
|
||
_run("constants/alert_levels_order", test_constants_alert_levels_order)
|
||
_run("constants/thresholds_positive", test_constants_thresholds_positive)
|
||
_run("constants/platform_kit_min_version", test_constants_platform_kit_min_version)
|
||
_run("constants/high_risk_labels_consistent", test_constants_high_risk_labels_consistent)
|
||
_run("scripts/subdirs_present", test_scripts_subdirs_present)
|
||
_run("main/commands_complete", test_main_py_commands_complete)
|
||
_run("main/version_prints_skill_version", test_main_py_version_command_prints_version)
|
||
_run("main/analyze_no_crash", test_main_py_analyze_command_does_not_crash)
|
||
_run("main/argv_dispatch", test_main_py_main_dispatch)
|
||
_run("domain/anomaly_thresholds_consistent", test_anomaly_classification_thresholds_consistent_with_constants)
|
||
|
||
total = len(_RESULTS)
|
||
passed = sum(1 for _, ok, _ in _RESULTS if ok)
|
||
failed = total - passed
|
||
|
||
print()
|
||
print(f"=== Summary: {passed}/{total} passed, {failed} failed ===", flush=True)
|
||
if failed:
|
||
print("\nFailures:")
|
||
for name, ok, msg in _RESULTS:
|
||
if not ok:
|
||
print(f" - {name}: {msg}")
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|