367 lines
14 KiB
Python
367 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""默认测试套件：制度文件冲突检测与合规审查。
|
||
|
||
设计原则：
|
||
- 仅用标准库（不依赖 pytest / jinja2），CI 与本地均能直接 `python tests/run_tests.py` 跑
|
||
- 覆盖：
|
||
- util.constants 版本/标识对齐（SKILL.md / CHANGELOG.md / constants.py）
|
||
- util.conflict_classifier 纯函数（classify_conflict_severity / normalize_policy_text / parse_policy_draft_id）
|
||
- scripts.main CLI 命令注册表与基础调用（占位骨架可正常 dispatch）
|
||
- 退出码：0 = 全部通过，1 = 有失败
|
||
- -v 开关：详细输出每个用例的状态
|
||
|
||
release.ps1 在打 tag 前会自动跑 `python tests/run_tests.py -v`，任何一个
|
||
test_* 返回 (False, msg) 即 abort 发布。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import io
|
||
import re
|
||
import sys
|
||
import traceback
|
||
from contextlib import redirect_stdout, redirect_stderr
|
||
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_*，记录结果。"""
|
||
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}"
|
||
|
||
# CHANGELOG 顶部应有 ## X.Y.Z
|
||
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_constants_policy_draft_statuses() -> tuple[bool, str]:
|
||
"""送审稿状态机：5 状态枚举完整 + 顺序正确"""
|
||
from util.constants import POLICY_DRAFT_STATUSES
|
||
|
||
expected = ("可送审", "审查中", "已通过", "冲突中", "驳回")
|
||
if POLICY_DRAFT_STATUSES != expected:
|
||
return False, f"POLICY_DRAFT_STATUSES={POLICY_DRAFT_STATUSES} != {expected}"
|
||
return True, f"{len(POLICY_DRAFT_STATUSES)} 状态按生命周期顺序"
|
||
|
||
|
||
def test_constants_conflict_types() -> tuple[bool, str]:
|
||
"""冲突类型枚举：含条款冲突/程序不合规/重复发文/用词歧义"""
|
||
from util.constants import CONFLICT_TYPES, CONFLICT_SEVERITY_LEVELS, CONFLICT_SEVERITY_LABELS
|
||
|
||
required = {"条款直接冲突", "程序不合规", "疑似重复发文", "用词歧义"}
|
||
missing = required - set(CONFLICT_TYPES)
|
||
if missing:
|
||
return False, f"CONFLICT_TYPES 缺少: {sorted(missing)}"
|
||
if CONFLICT_SEVERITY_LEVELS != ("low", "medium", "high"):
|
||
return False, f"CONFLICT_SEVERITY_LEVELS={CONFLICT_SEVERITY_LEVELS}"
|
||
if set(CONFLICT_SEVERITY_LABELS.keys()) != set(CONFLICT_SEVERITY_LEVELS):
|
||
return False, (
|
||
f"CONFLICT_SEVERITY_LABELS keys={set(CONFLICT_SEVERITY_LABELS.keys())} "
|
||
f"!= {set(CONFLICT_SEVERITY_LEVELS)}"
|
||
)
|
||
return True, f"{len(CONFLICT_TYPES)} 冲突类型 + {len(CONFLICT_SEVERITY_LEVELS)} 等级"
|
||
|
||
|
||
# ---------- util.conflict_classifier 纯函数 ----------
|
||
|
||
def test_classify_conflict_severity_thresholds() -> tuple[bool, str]:
|
||
"""classify_conflict_severity 高/中/低 三档边界"""
|
||
from util.conflict_classifier import classify_conflict_severity
|
||
t = (0.85, 0.92)
|
||
cases = [
|
||
(1.00, "high"), # 完全相同
|
||
(0.95, "high"), # 高于 high_min
|
||
(0.92, "high"), # 边界 =
|
||
(0.91, "medium"), # 边界下
|
||
(0.88, "medium"), # 中等
|
||
(0.85, "medium"), # 边界 =
|
||
(0.84, "low"), # 低于 medium_min
|
||
(0.50, "low"), # 明显不相关
|
||
(0.00, "low"), # 完全不同
|
||
]
|
||
for sim, want in cases:
|
||
got = classify_conflict_severity(sim, t)
|
||
if got != want:
|
||
return False, f"classify_conflict_severity({sim}) = {got}, want {want}"
|
||
return True, "9 边界 + 越界场景"
|
||
|
||
|
||
def test_classify_conflict_severity_custom_thresholds() -> tuple[bool, str]:
|
||
"""classify_conflict_severity 自定义阈值"""
|
||
from util.conflict_classifier import classify_conflict_severity
|
||
# 收紧阈值
|
||
if classify_conflict_severity(0.90, thresholds=(0.80, 0.90)) != "high":
|
||
return False, "自定义阈值 high 边界错"
|
||
if classify_conflict_severity(0.85, thresholds=(0.80, 0.90)) != "medium":
|
||
return False, "自定义阈值 medium 错"
|
||
if classify_conflict_severity(0.79, thresholds=(0.80, 0.90)) != "low":
|
||
return False, "自定义阈值 low 错"
|
||
return True, "自定义阈值 high/medium/low"
|
||
|
||
|
||
def test_normalize_policy_text_fullwidth() -> tuple[bool, str]:
|
||
"""全角数字/字母/空格转半角 + 合并连续空白"""
|
||
from util.conflict_classifier import normalize_policy_text
|
||
|
||
cases = [
|
||
("第一章 总 则", "第一章 总 则"),
|
||
("第 1 条 总则", "第 1 条 总则"),
|
||
("Article １：Definition", "Article 1:Definition"),
|
||
("", ""),
|
||
(" 多个 空白 ", "多个 空白"),
|
||
("无变化", "无变化"),
|
||
("条款ＡＢＣ１２３", "条款ABC123"),
|
||
]
|
||
for inp, want in cases:
|
||
got = normalize_policy_text(inp)
|
||
if got != want:
|
||
return False, f"normalize_policy_text({inp!r}) = {got!r}, want {want!r}"
|
||
return True, "全角转半角 + 空白合并 + 边界"
|
||
|
||
|
||
def test_normalize_policy_text_none() -> tuple[bool, str]:
|
||
"""None 输入返回空字符串（不抛异常）"""
|
||
from util.conflict_classifier import normalize_policy_text
|
||
|
||
if normalize_policy_text(None) != "":
|
||
return False, "None 应返回空字符串"
|
||
return True, "None → \"\""
|
||
|
||
|
||
def test_parse_policy_draft_id() -> tuple[bool, str]:
|
||
"""parse_policy_draft_id 去空白 + 校验非空"""
|
||
from util.conflict_classifier import parse_policy_draft_id
|
||
|
||
if parse_policy_draft_id("D-001") != "D-001":
|
||
return False, "正常 ID 应原样返回"
|
||
if parse_policy_draft_id(" D-002 ") != "D-002":
|
||
return False, "前后空白应被 strip"
|
||
if parse_policy_draft_id(123) != "123":
|
||
return False, "数字应转字符串"
|
||
|
||
# 空值校验
|
||
raised = False
|
||
try:
|
||
parse_policy_draft_id("")
|
||
except ValueError:
|
||
raised = True
|
||
if not raised:
|
||
return False, "空字符串应抛 ValueError"
|
||
|
||
raised = False
|
||
try:
|
||
parse_policy_draft_id(" ")
|
||
except ValueError:
|
||
raised = True
|
||
if not raised:
|
||
return False, "全空白字符串应抛 ValueError"
|
||
|
||
raised = False
|
||
try:
|
||
parse_policy_draft_id(None)
|
||
except ValueError:
|
||
raised = True
|
||
if not raised:
|
||
return False, "None 应抛 ValueError"
|
||
|
||
return True, "正常/空白 strip/数字/空/全空白/None 6 场景"
|
||
|
||
|
||
# ---------- scripts.main CLI 命令注册 ----------
|
||
|
||
def test_main_commands_registered() -> tuple[bool, str]:
|
||
"""scripts/main.py COMMANDS 字典含全部 9 个命令"""
|
||
from main import COMMANDS
|
||
|
||
expected = {
|
||
"health",
|
||
"version",
|
||
"config-path",
|
||
"init-db",
|
||
"audit-policy-draft",
|
||
"recheck-conflict-clause",
|
||
"export-review-report",
|
||
"mark-policy-status",
|
||
"rescan-pending-drafts",
|
||
}
|
||
missing = expected - set(COMMANDS.keys())
|
||
if missing:
|
||
return False, f"COMMANDS 缺: {sorted(missing)}"
|
||
extra = set(COMMANDS.keys()) - expected
|
||
if extra:
|
||
return False, f"COMMANDS 多余: {sorted(extra)}"
|
||
return True, f"{len(COMMANDS)} 个命令全部注册"
|
||
|
||
|
||
def test_main_version_command_prints_skill_version() -> tuple[bool, str]:
|
||
"""`python main.py version` 应输出 SKILL_VERSION"""
|
||
from util.constants import SKILL_VERSION
|
||
import main as cli_main
|
||
|
||
buf = io.StringIO()
|
||
with redirect_stdout(buf):
|
||
rc = cli_main.version()
|
||
output = buf.getvalue()
|
||
if rc != 0:
|
||
return False, f"version() 返回 {rc}, want 0"
|
||
if SKILL_VERSION not in output:
|
||
return False, f"version() 输出不含 {SKILL_VERSION}: {output!r}"
|
||
return True, f"输出含 SKILL_VERSION={SKILL_VERSION}"
|
||
|
||
|
||
def test_main_dispatch_unknown_command() -> tuple[bool, str]:
|
||
"""main() 对未知命令应返回 1 并打印 usage"""
|
||
import main as cli_main
|
||
|
||
buf_out = io.StringIO()
|
||
buf_err = io.StringIO()
|
||
with redirect_stdout(buf_out), redirect_stderr(buf_err):
|
||
rc = cli_main.main(["not-a-real-command"])
|
||
if rc == 0:
|
||
return False, f"未知命令应返回非 0, got {rc}"
|
||
if "Usage:" not in buf_out.getvalue():
|
||
return False, f"未知命令应打印 Usage, got {buf_out.getvalue()!r}"
|
||
return True, f"未知命令 → rc={rc} + Usage 输出"
|
||
|
||
|
||
def test_main_dispatch_audit_policy_draft() -> tuple[bool, str]:
|
||
"""`audit-policy-draft D-001` 占位实现应 echo draft_id"""
|
||
import main as cli_main
|
||
|
||
buf = io.StringIO()
|
||
with redirect_stdout(buf):
|
||
rc = cli_main.main(["audit-policy-draft", "D-001"])
|
||
output = buf.getvalue()
|
||
if rc != 0:
|
||
return False, f"audit-policy-draft 返回 {rc}, want 0"
|
||
if "D-001" not in output:
|
||
return False, f"输出应含 D-001, got {output!r}"
|
||
return True, "audit-policy-draft 占位 echo 正常"
|
||
|
||
|
||
def test_main_mark_policy_status_validates_enum() -> tuple[bool, str]:
|
||
"""`mark-policy-status` 未知状态应返回 1（不允许静默接受）"""
|
||
import main as cli_main
|
||
|
||
buf_out = io.StringIO()
|
||
buf_err = io.StringIO()
|
||
with redirect_stdout(buf_out), redirect_stderr(buf_err):
|
||
rc = cli_main.mark_policy_status("D-001", "非法状态")
|
||
if rc == 0:
|
||
return False, "非法状态应返回非 0"
|
||
if "ERROR" not in buf_err.getvalue():
|
||
return False, f"应输出 ERROR 提示, got stderr={buf_err.getvalue()!r}"
|
||
|
||
# 合法状态应返回 0
|
||
buf_ok = io.StringIO()
|
||
with redirect_stdout(buf_ok):
|
||
rc_ok = cli_main.mark_policy_status("D-002", "审查中")
|
||
if rc_ok != 0:
|
||
return False, f"合法状态应返回 0, got {rc_ok}"
|
||
return True, "非法状态 → 非 0 + ERROR；合法状态 → 0"
|
||
|
||
|
||
# ---------- 入口 ----------
|
||
|
||
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("=== detect-policy-conflict-gs 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/policy_draft_statuses", test_constants_policy_draft_statuses)
|
||
_run("constants/conflict_types", test_constants_conflict_types)
|
||
_run("conflict_classifier/classify_thresholds", test_classify_conflict_severity_thresholds)
|
||
_run("conflict_classifier/classify_custom_thresholds", test_classify_conflict_severity_custom_thresholds)
|
||
_run("conflict_classifier/normalize_fullwidth", test_normalize_policy_text_fullwidth)
|
||
_run("conflict_classifier/normalize_none", test_normalize_policy_text_none)
|
||
_run("conflict_classifier/parse_policy_draft_id", test_parse_policy_draft_id)
|
||
_run("main/commands_registered", test_main_commands_registered)
|
||
_run("main/version_prints_skill_version", test_main_version_command_prints_skill_version)
|
||
_run("main/dispatch_unknown_command", test_main_dispatch_unknown_command)
|
||
_run("main/dispatch_audit_policy_draft", test_main_dispatch_audit_policy_draft)
|
||
_run("main/mark_policy_status_validates", test_main_mark_policy_status_validates_enum)
|
||
|
||
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())
|