221 lines
11 KiB
Python
221 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""P1-G (#14) — acceptance-event model tests. Standalone (no pytest). Exit 0 = all pass.
|
|
|
|
검증 대상:
|
|
1. new_report 가 불변 스냅샷에 계보 필드(attempt-id, supersedes-report-id)를 심는다.
|
|
2. acceptance_log: append 이벤트 + latest-accepted 질의 + supersession/rejection 제외.
|
|
3. report_tags 가 대체/거부된 리포트를 기본 제외하고 --include-superseded 로 포함한다.
|
|
4. report/acceptance-event 스키마가 유효하며, 계보 필드 없는 최소 리포트도 여전히 통과.
|
|
"""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
HOOKS = os.path.join(ROOT, ".claude", "hooks")
|
|
SCHEMAS = os.path.join(ROOT, ".claude", "schemas")
|
|
FIX = os.path.join(ROOT, ".claude", "tests", "fixtures")
|
|
PY = sys.executable
|
|
os.makedirs(FIX, exist_ok=True)
|
|
|
|
# 격리 워크스페이스(실제 _sandbox 오염 방지). 절대경로 -> _workspace 가 그대로 root 로 사용.
|
|
WS = os.path.join(FIX, "p1g-ws")
|
|
shutil.rmtree(WS, ignore_errors=True)
|
|
os.makedirs(WS, exist_ok=True)
|
|
|
|
# in-process import 를 위해 환경/경로 설정(subprocess 에도 같은 env 전달).
|
|
os.environ["CLAUDE_PROJECT_DIR"] = ROOT
|
|
os.environ["ORGOS_WORKSPACE"] = WS
|
|
sys.path.insert(0, HOOKS)
|
|
|
|
passed, failed = 0, 0
|
|
|
|
|
|
def check(name, ok):
|
|
global passed, failed
|
|
if ok:
|
|
passed += 1
|
|
print(f" PASS {name}")
|
|
else:
|
|
failed += 1
|
|
print(f" FAIL {name}")
|
|
|
|
|
|
def run(script, args=None, stdin=None):
|
|
e = dict(os.environ)
|
|
e["CLAUDE_PROJECT_DIR"] = ROOT
|
|
e["ORGOS_WORKSPACE"] = WS
|
|
return subprocess.run(
|
|
[PY, os.path.join(HOOKS, script)] + (args or []),
|
|
input=stdin, capture_output=True, text=True, env=e,
|
|
)
|
|
|
|
|
|
def rid_of(relpath):
|
|
"""새로 mint된 상대경로에서 report-id(파일명에서 .report.yaml 제거) 추출."""
|
|
base = os.path.basename(relpath.strip())
|
|
return base[:-len(".report.yaml")] if base.endswith(".report.yaml") else base
|
|
|
|
|
|
# ============================================================ 1. new_report 계보 필드
|
|
print("== new_report lineage fields (immutable snapshot) ==")
|
|
_p1 = run("new_report.py", ["--workflow", "wfG", "--role", "ROLE-X", "--stub"]).stdout.strip()
|
|
RID_A = rid_of(_p1)
|
|
_a_body = open(os.path.join(ROOT, _p1)).read()
|
|
check("new_report stub has attempt-id: 1 (첫 시도)", "attempt-id: 1" in _a_body)
|
|
check("new_report stub keeps immutable format (report-id/report-header)",
|
|
f"report-id: {RID_A}" in _a_body and "report-header:" in _a_body)
|
|
|
|
_p2 = run("new_report.py", ["--workflow", "wfG", "--role", "ROLE-X", "--stub", "--supersedes", RID_A]).stdout.strip()
|
|
RID_B = rid_of(_p2)
|
|
_b_body = open(os.path.join(ROOT, _p2)).read()
|
|
check("2nd mint gets attempt-id: 2 (파일 수 파생)", "attempt-id: 2" in _b_body)
|
|
check("--supersedes writes supersedes-report-id into snapshot", f"supersedes-report-id: {RID_A}" in _b_body)
|
|
check("mint never collides (A != B)", RID_A != RID_B)
|
|
|
|
# ============================================================ 2. acceptance_log append/query
|
|
print("== acceptance_log: append events + queries ==")
|
|
import acceptance_log as AL # noqa: E402
|
|
|
|
|
|
def _write_valid(relpath):
|
|
"""P0-4c: accepted 이벤트는 실존·validate 통과 report 만 인정한다. 스텁은 빈 BLUF 라
|
|
검증 실패하므로, 이 헬퍼로 minted 경로를 유효 report 로 채운다(role-id 는 테스트 전용 TST-GENERIC-FIXTURE)."""
|
|
p = os.path.join(ROOT, relpath.strip())
|
|
rid = rid_of(relpath)
|
|
with open(p, "w", encoding="utf-8") as f:
|
|
f.write(
|
|
"report-type: work\n" f"report-id: {rid}\n" "workflow-id: wfG\n" "role-id: TST-GENERIC-FIXTURE\n"
|
|
'work-summary: "완료"\n' "report-header:\n" ' bottom-line: "결론"\n'
|
|
" decision-needed: { needed: false }\n"
|
|
" confidence: { value: Med, derived-from: evidence }\n"
|
|
" risks: []\n evidence:\n - source-uri: README.md\n grade: E3\n")
|
|
return p
|
|
|
|
|
|
# 이벤트: B 가 A 를 대체하며 수락됨. B 는 유효 report 여야 accepted 등록 가능(P0-4c).
|
|
_write_valid(_p2)
|
|
_ap_event = AL.build_event(RID_B, "accepted", workflow="wfG", role="ROLE-X", supersedes=RID_A)
|
|
check("low-level append_event stores engine-authorized event", AL.append_event(_ap_event))
|
|
|
|
# 별도 리포트 C 는 변경요청(거부)됨. changes-requested 는 실존만 요구(validate 불필요).
|
|
RID_C = "ROLE-X-C-rejected"
|
|
_cdir = os.path.join(WS, "completion-records", "wfG")
|
|
os.makedirs(_cdir, exist_ok=True)
|
|
with open(os.path.join(_cdir, RID_C + ".report.yaml"), "w") as _cf:
|
|
_cf.write("report-header:\n bottom-line: rejected\n")
|
|
AL.append_event(AL.build_event(RID_C, "changes-requested", workflow="wfG", role="ROLE-X"))
|
|
|
|
# CLI 질의
|
|
_la = run("acceptance_log.py", ["latest-accepted", "--workflow", "wfG", "--role", "ROLE-X"]).stdout.strip()
|
|
check("CLI latest-accepted(wfG,ROLE-X) == B", _la == RID_B)
|
|
check("CLI is-superseded A -> yes", run("acceptance_log.py", ["is-superseded", "--report-id", RID_A]).stdout.strip() == "yes")
|
|
check("CLI is-superseded B -> no", run("acceptance_log.py", ["is-superseded", "--report-id", RID_B]).stdout.strip() == "no")
|
|
|
|
# import API (재사용) — subprocess 가 방금 쓴 원장을 그대로 읽는다.
|
|
check("API latest_accepted == B", AL.latest_accepted("wfG", "ROLE-X") == RID_B)
|
|
check("API is_superseded(A) True", AL.is_superseded(RID_A) is True)
|
|
check("API is_superseded(B) False", AL.is_superseded(RID_B) is False)
|
|
_excl = AL.excluded_report_ids()
|
|
check("API excluded = {A(superseded), C(rejected)}", _excl == {RID_A, RID_C})
|
|
check("API is_current(B) True / is_current(A) False", AL.is_current(RID_B) and not AL.is_current(RID_A))
|
|
|
|
# validate() import-safe
|
|
check("validate() accepts a well-formed event",
|
|
AL.validate(AL.build_event(RID_B, "accepted", workflow="wfG", role="ROLE-X")) == [])
|
|
_bad = AL.validate({"report-id": "", "decision": "nope"})
|
|
check("validate() flags bad decision + missing fields", any("decision" in e for e in _bad) and len(_bad) >= 2)
|
|
|
|
# Task 15: report-sha256 바인딩(venture-decision human-gate 위조 방지, P1 §9.4)
|
|
_ev_sha = AL.build_event("r1", "accepted", workflow="wf1", role="HUMAN-001", report_sha256="abc123")
|
|
check("build_event report-sha256 포함", _ev_sha.get("report-sha256") == "abc123")
|
|
_ev_nosha = AL.build_event("r1", "accepted", workflow="wf1")
|
|
check("report-sha256 미지정 시 부재", "report-sha256" not in _ev_nosha)
|
|
|
|
# fail-safe: workspace 미설정이면 크래시 없이 degrade
|
|
_env2 = {k: v for k, v in os.environ.items() if k not in ("ORGOS_WORKSPACE",)}
|
|
_env2["CLAUDE_PROJECT_DIR"] = ROOT
|
|
_ns = subprocess.run([PY, os.path.join(HOOKS, "acceptance_log.py"),
|
|
"latest-accepted", "--workflow", "wfG"],
|
|
capture_output=True, text=True, env=_env2)
|
|
check("workspace unset -> latest-accepted degrades (exit 0, no crash)", _ns.returncode == 0)
|
|
|
|
# ============================================================ 3. report_tags 제외
|
|
# 상호 substring 이 없는 별도 report-id 로 검증(경로 문자열 오검출 방지).
|
|
print("== report_tags excludes superseded/rejected by default ==")
|
|
_wfdir = os.path.join(WS, "completion-records", "wfT")
|
|
os.makedirs(_wfdir, exist_ok=True)
|
|
SA, SB, SC = "rpt-alpha", "rpt-bravo", "rpt-charlie"
|
|
|
|
def _write_report(rid, tag):
|
|
# P0-3/P0-5: report 는 report-type + identity 필수(role-id 는 테스트 전용 TST-GENERIC-FIXTURE). accepted 로
|
|
# 등록될 SB 는 validate 통과해야 하므로 실존 evidence(README.md)로 유효 report 를 만든다.
|
|
p = os.path.join(_wfdir, rid + ".report.yaml")
|
|
with open(p, "w") as f:
|
|
f.write(
|
|
"report-type: work\n" f"report-id: {rid}\n" "workflow-id: wfT\n" "role-id: TST-GENERIC-FIXTURE\n"
|
|
f"tags: [{tag}]\n" 'work-summary: "s"\n'
|
|
"report-header:\n bottom-line: hi\n decision-needed: { needed: false }\n"
|
|
" confidence: { value: Med, derived-from: evidence }\n risks: []\n"
|
|
" evidence:\n - source-uri: README.md\n grade: E3\n")
|
|
return p
|
|
|
|
|
|
# 리포트 파일을 먼저 만든다(P0-4c: accepted 이벤트는 실존·validate 통과 report 만 인정).
|
|
_write_report(SA, "topicG") # superseded
|
|
_write_report(SB, "topicG") # current
|
|
_write_report(SC, "topicG") # rejected
|
|
|
|
# 이벤트: SB 가 SA 를 대체하며 수락, SC 는 변경요청(거부).
|
|
AL.append_event(AL.build_event(SB, "accepted", workflow="wfT", role="ROLE-T", supersedes=SA))
|
|
AL.append_event(AL.build_event(SC, "changes-requested", workflow="wfT", role="ROLE-T"))
|
|
|
|
_def = run("report_tags.py", ["--tag", "topicG"]).stdout
|
|
check("default: current SB is shown", SB in _def)
|
|
check("default: superseded SA is hidden", SA not in _def)
|
|
check("default: rejected SC is hidden", SC not in _def)
|
|
check("default: header says 1건 (현재 유효만)", "1건" in _def and "현재 유효만" in _def)
|
|
|
|
_inc = run("report_tags.py", ["--tag", "topicG", "--include-superseded"]).stdout
|
|
check("--include-superseded: all three shown", SA in _inc and SB in _inc and SC in _inc)
|
|
check("--include-superseded: SA/SC marked superseded/rejected", _inc.count("[superseded/rejected]") == 2)
|
|
|
|
# ============================================================ 4. 스키마 유효 + 하위호환
|
|
print("== schemas valid + backward compatible ==")
|
|
import jsonschema # noqa: E402
|
|
|
|
_report_schema = json.load(open(os.path.join(SCHEMAS, "report.schema.json")))
|
|
_event_schema = json.load(open(os.path.join(SCHEMAS, "acceptance-event.schema.json")))
|
|
jsonschema.Draft7Validator.check_schema(_report_schema)
|
|
jsonschema.Draft7Validator.check_schema(_event_schema)
|
|
check("report.schema.json + acceptance-event.schema.json are valid JSON Schema", True)
|
|
|
|
_rv = jsonschema.Draft7Validator(_report_schema)
|
|
# P0-3/P0-5: report-type + identity(report-id/workflow-id/role-id)는 이제 필수.
|
|
_bare = {"report-header": {"bottom-line": "x", "decision-needed": {"needed": False},
|
|
"confidence": {"value": "Med"}, "risks": [], "evidence": []}}
|
|
check("bare report (no report-type/identity) -> schema errors (P0-5)",
|
|
list(_rv.iter_errors(_bare)) != [])
|
|
|
|
# 완전한 리포트(identity + 계보 필드)는 통과
|
|
_full = {"report-type": "work", "report-id": "r1", "workflow-id": "wf1", "role-id": "TST-GENERIC-FIXTURE",
|
|
"attempt-id": 2, "supersedes-report-id": RID_A,
|
|
"report-header": {"bottom-line": "x", "decision-needed": {"needed": False},
|
|
"confidence": {"value": "Med"}, "risks": [], "evidence": []}}
|
|
check("full report WITH identity + lineage fields validates", list(_rv.iter_errors(_full)) == [])
|
|
|
|
# 이벤트 스키마 통과
|
|
_ev = AL.build_event(RID_B, "accepted", workflow="wfG", role="ROLE-X", supersedes=RID_A)
|
|
_ev_errors = list(jsonschema.Draft7Validator(_event_schema).iter_errors(_ev))
|
|
check("built acceptance event validates against acceptance-event.schema.json", _ev_errors == [])
|
|
_bad_ev = {"acceptance-event-id": "ae-x", "report-id": "r", "decision": "nope", "effective-at": "t"}
|
|
check("event with bad decision fails schema (enum)",
|
|
list(jsonschema.Draft7Validator(_event_schema).iter_errors(_bad_ev)) != [])
|
|
|
|
# ============================================================ 정리
|
|
shutil.rmtree(WS, ignore_errors=True)
|
|
print(f"\n{passed} passed, {failed} failed")
|
|
sys.exit(1 if failed else 0)
|