205 lines
6.8 KiB
Python
205 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
||
"""经验升格链对 muse-example 真实库的集成测试。
|
||
|
||
覆盖：
|
||
- 登记校验：kind/title/detail 闭集与形状；
|
||
- 禁止自动升格：proposed 不得跳过评审直接 promoted（应用层 + DB 触发器双保险）；
|
||
- 人工升格闭环：reviewing → promoted 必须指明目标类型与落点；
|
||
- 终态不可再流转；驳回可从 proposed/reviewing 发起。
|
||
|
||
测试行用 unittest-lesson- 前缀的 run_id 隔离，结束物理清理（本表可变）。
|
||
|
||
跑法（需 Tailscale 内网可达 muse-example）：
|
||
.venv/bin/python tests/skills/记录运行证据/test_lesson_registry_db.py
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pathlib
|
||
import sys
|
||
import uuid
|
||
|
||
from psycopg.errors import RaiseException
|
||
|
||
PROJECT_ROOT = pathlib.Path(__file__).resolve().parents[3]
|
||
SKILLS_DIR = PROJECT_ROOT / ".agent" / "skills"
|
||
SCRIPT_DIR = PROJECT_ROOT / "muse" / "authority" / "evidence" / "skills" / "记录运行证据" / "scripts"
|
||
if str(SCRIPT_DIR) not in sys.path:
|
||
sys.path.insert(0, str(SCRIPT_DIR))
|
||
|
||
from muse_db import connect # noqa: E402
|
||
import lesson_registry # noqa: E402
|
||
from lesson_registry import ( # noqa: E402
|
||
LessonError, list_lessons, promote, propose_lesson, propose_lesson_dedup, reject, start_review,
|
||
)
|
||
|
||
SUFFIX = uuid.uuid4().hex[:10]
|
||
RUN_PREFIX = f"unittest-lesson-{SUFFIX}"
|
||
_seq = {"n": 0}
|
||
|
||
|
||
def _run_id() -> str:
|
||
_seq["n"] += 1
|
||
return f"{RUN_PREFIX}-{_seq['n']}"
|
||
|
||
|
||
def _cleanup() -> None:
|
||
with connect() as conn:
|
||
conn.execute("DELETE FROM example_lesson WHERE run_id LIKE %s", (f"{RUN_PREFIX}%",))
|
||
conn.commit()
|
||
|
||
|
||
def _propose(**overrides):
|
||
kwargs = dict(kind="lesson", title=f"测试经验 {SUFFIX}",
|
||
detail={"situation": "机械门通过但语义未跑", "action": "补语义兜底",
|
||
"outcome": "接受通道失败关闭"}, run_id=_run_id())
|
||
kwargs.update(overrides)
|
||
return propose_lesson(**kwargs)
|
||
|
||
|
||
def test_propose_validates_inputs() -> None:
|
||
out = _propose()
|
||
assert out["status"] == "proposed" and out["lesson_id"] >= 1
|
||
win = _propose(kind="win", title="成功锚点命中")
|
||
assert win["status"] == "proposed"
|
||
for bad_kwargs in (
|
||
dict(kind="epiphany"),
|
||
dict(title=""),
|
||
dict(title="x" * 201),
|
||
dict(detail={}),
|
||
):
|
||
try:
|
||
_propose(**bad_kwargs)
|
||
raise AssertionError(f"非法输入必须被拒绝: {bad_kwargs}")
|
||
except AssertionError:
|
||
raise
|
||
except LessonError:
|
||
pass
|
||
|
||
|
||
def test_no_auto_promotion() -> None:
|
||
out = _propose()
|
||
# 应用层：proposed 直接 promote 被拒（期望来源是 reviewing）
|
||
try:
|
||
promote(out["lesson_id"], decided_by="founder", target_type="pattern",
|
||
target_ref="pattern:test")
|
||
raise AssertionError("不得跳过评审自动升格")
|
||
except AssertionError:
|
||
raise
|
||
except LessonError:
|
||
pass
|
||
# 绕过应用层直接 UPDATE 也被触发器拒绝
|
||
try:
|
||
with connect() as conn:
|
||
conn.execute(
|
||
"UPDATE example_lesson SET status='promoted', promotion_target_type='pattern', "
|
||
"promotion_target_ref='pattern:test' WHERE id=%s", (out["lesson_id"],))
|
||
conn.commit()
|
||
raise AssertionError("触发器必须拒绝自动升格")
|
||
except AssertionError:
|
||
raise
|
||
except RaiseException:
|
||
pass
|
||
rows = list_lessons(status="proposed")
|
||
assert any(row["id"] == out["lesson_id"] for row in rows)
|
||
|
||
|
||
def test_human_promotion_requires_target() -> None:
|
||
out = _propose(kind="win")
|
||
start_review(out["lesson_id"], decided_by="founder", rationale="复利候选")
|
||
# reviewing 升格必须指明目标
|
||
try:
|
||
promote(out["lesson_id"], decided_by="founder", target_type="ritual", target_ref="x")
|
||
raise AssertionError("非法目标类型必须被拒绝")
|
||
except AssertionError:
|
||
raise
|
||
except LessonError:
|
||
pass
|
||
try:
|
||
promote(out["lesson_id"], decided_by="founder", target_type="pattern", target_ref="")
|
||
raise AssertionError("缺落点必须被拒绝")
|
||
except AssertionError:
|
||
raise
|
||
except LessonError:
|
||
pass
|
||
promoted = promote(out["lesson_id"], decided_by="founder", target_type="skill",
|
||
target_ref="写下一章", rationale="篇幅自纠环有效")
|
||
assert promoted["status"] == "promoted"
|
||
# 终态不得再流转
|
||
for action in (start_review, reject):
|
||
try:
|
||
action(out["lesson_id"], decided_by="founder")
|
||
raise AssertionError("promoted 终态不得再流转")
|
||
except AssertionError:
|
||
raise
|
||
except LessonError:
|
||
pass
|
||
|
||
|
||
def test_reject_paths() -> None:
|
||
direct = _propose()
|
||
assert reject(direct["lesson_id"], decided_by="founder", rationale="n=1 证据不足")["status"] == "rejected"
|
||
reviewed = _propose()
|
||
start_review(reviewed["lesson_id"], decided_by="founder")
|
||
assert reject(reviewed["lesson_id"], decided_by="founder")["status"] == "rejected"
|
||
# rejected 终态
|
||
try:
|
||
start_review(reviewed["lesson_id"], decided_by="founder")
|
||
raise AssertionError("rejected 终态不得再流转")
|
||
except AssertionError:
|
||
raise
|
||
except LessonError:
|
||
pass
|
||
|
||
|
||
def test_missing_decided_by_rejected() -> None:
|
||
out = _propose()
|
||
try:
|
||
start_review(out["lesson_id"], decided_by="")
|
||
raise AssertionError("decided_by 必填")
|
||
except AssertionError:
|
||
raise
|
||
except LessonError:
|
||
pass
|
||
|
||
|
||
def test_propose_dedup_idempotent() -> None:
|
||
run_id = _run_id()
|
||
title = f"dedup-{SUFFIX}"
|
||
first = propose_lesson_dedup(
|
||
kind="lesson", title=title,
|
||
detail={"skill": "unittest", "note": "first"},
|
||
run_id=run_id,
|
||
)
|
||
second = propose_lesson_dedup(
|
||
kind="lesson", title=title,
|
||
detail={"skill": "unittest", "note": "second"},
|
||
run_id=run_id,
|
||
)
|
||
assert first["deduped"] is False and second["deduped"] is True
|
||
assert first["lesson_id"] == second["lesson_id"]
|
||
rows = list_lessons(kind="lesson")
|
||
assert sum(1 for row in rows if row["run_id"] == run_id and row["title"] == title) == 1
|
||
|
||
|
||
def main() -> None:
|
||
tests = (
|
||
("登记输入校验", test_propose_validates_inputs),
|
||
("禁止自动升格", test_no_auto_promotion),
|
||
("人工升格必须指明目标", test_human_promotion_requires_target),
|
||
("驳回路径与终态", test_reject_paths),
|
||
("decided_by 必填", test_missing_decided_by_rejected),
|
||
("run_id+title 幂等", test_propose_dedup_idempotent),
|
||
)
|
||
try:
|
||
for name, test in tests:
|
||
test()
|
||
print(f"PASS: {name}")
|
||
print("PASS：经验升格链真实库集成测试全部通过")
|
||
finally:
|
||
_cleanup()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|