feat(scripts): Studio 저장 어댑터 — 게시를 코드가 막고 무인 저장은 꺼 둔다
서버 계약은 tech-log-backend @ a000f87 의 openapi 와 컨트롤러에서 읽었다. Idempotency-Key 는 필수이고 200자 이하다. expectedVersion 은 스키마가 minimum 1 로 못 박으므로 빈 값은 「검사 안 함」이 아니라 「반드시 충돌」이다. 멱등 키를 기록 경로에서 유도한다. 프런트는 호출마다 새 uuid 를 만들어서 재시도가 문서를 둘 만든다. 만들기는 경로만으로 키를 만들어 몇 번을 다시 시도해도 문서가 하나이고, 저장은 보낼 내용을 키에 넣는다 — 경로만 쓰면 두 번째 저장이 첫 결과로 조용히 재생된다. 게시 요청을 만드는 코드가 없고, 계획이 게시 경로를 가리키면 거절한다. 지금까지 게시를 막고 있던 것은 스킬 문서의 문장 하나였다. 무인 저장은 꺼져 있다. 서버 권한이 studio:read·studio:write 둘뿐이라 저장 계정이 게시도 할 수 있다. 켜는 것은 studio:publish 가 갈라진 뒤의 판단이지 이 파일의 기본값이 아니다. 저장 뒤 되읽어 견줄 때 문자열 하나로 판정하지 않는다. 정규화가 지운 차이를 통과로 세면 비교 틀이 결함을 만들어 내거나 지운다. 정규화 뒤에만 같아지는 것은 whitespaceOnly 로, 되읽은 값이 보낸 값의 앞부분인 것은 truncated 로 따로 낸다. 못 보는 칸은 사유와 함께 적는다 — 조용히 건너뛰면 「전부 같다」가 「본 것만 같다」를 가린다. 회귀의 첫 줄은 대조군이다. 손대지 않은 쌍이 「같다」로 나오지 않으면 나머지 「잡았다」가 전부 틀 탓이다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Q4vKjQo9KKBBokzxqXLCfk
This commit is contained in:
DongHyeonka
co-authored by
Claude Opus 5
Claude Opus 5
parent
5472751894
commit
32a369f335
@@ -0,0 +1,375 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Studio 저장 어댑터. 승인된 최종본만 넣고, 저장한 뒤 되읽어 대조한다.
|
||||
|
||||
**게시하지 않는다.** 이 파일에 게시 요청을 만드는 코드가 없고, 만들어 낸 요청이 게시 경로를
|
||||
가리키면 `_reject_forbidden_paths` 가 막는다. 스킬 문서의 「게시 버튼을 누르지 않는다」는
|
||||
문장 하나로 막던 것을 코드로 옮긴 것이다.
|
||||
|
||||
**무인 저장은 꺼져 있다.** `UNATTENDED_SAVE_ENABLED` 가 `False` 다. 서버 권한이
|
||||
`studio:read`·`studio:write` 둘뿐이라 저장할 수 있는 계정은 게시도 할 수 있고, 그래서
|
||||
사람이 안 보는 사이에 저장을 돌리지 않는다. 이 값을 `True` 로 바꾸는 것은 서버에
|
||||
`studio:publish` 가 갈라져 들어온 뒤의 일이다.
|
||||
|
||||
**자격증명을 이 저장소에 적지 않는다.** 세션 쿠키와 CSRF 토큰은 환경 변수로만 받는다.
|
||||
|
||||
서버 계약 (tech-log-backend @ a000f87 의 src/config/openapi/studio-v1.yaml 과 소스에서 읽었다)
|
||||
|
||||
| 무엇 | 값 |
|
||||
|---|---|
|
||||
| 만들기 | `POST /api/v1/studio/documents` → 201 |
|
||||
| 읽기 | `GET /api/v1/studio/documents/{id}` |
|
||||
| 저장 | `PUT /api/v1/studio/documents/{id}` → 200 |
|
||||
| 멱등 키 | `Idempotency-Key` 헤더 **필수**, 200자 이하 (`StudioIdempotency.java:34,77-88`) |
|
||||
| 재생 여부 | 응답 헤더 `Idempotency-Replayed` |
|
||||
| 낙관적 락 | `SaveDocumentCommand.expectedVersion` **필수 · minimum 1** |
|
||||
| 충돌 | `VERSION_CONFLICT` 409, `details.latestDocument` 에 현재 문서 전체 |
|
||||
| 인증 | 쿠키 `TECHLOG_SESSION` + 변경 요청에 `X-CSRF-TOKEN` |
|
||||
|
||||
python3 scripts/studio-save.py --record <기록.md> --package <review-package.json>
|
||||
python3 scripts/studio-save.py --record <기록.md> --package <pkg.json> -o <plan.json>
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
# 서버에 studio:publish 가 갈라져 들어오기 전에는 무인 저장을 켜지 않는다.
|
||||
# 켜는 조건은 사람의 판단이지 이 파일의 기본값이 아니다 (A-studio-change-requests.md CR-001).
|
||||
UNATTENDED_SAVE_ENABLED = False
|
||||
UNATTENDED_SAVE_BLOCKED_BY = (
|
||||
"CR-001 — 서버 권한이 studio:read·studio:write 둘뿐이라 저장 계정이 게시도 할 수 있다. "
|
||||
"studio:publish 가 갈라지기 전에는 사람이 보는 앞에서만 저장한다"
|
||||
)
|
||||
|
||||
BASE_PATH = "/api/v1/studio"
|
||||
FORBIDDEN_PATH = re.compile(r"/(publish|unpublish)(/|$)")
|
||||
MAX_KEY_LENGTH = 200 # StudioIdempotency.java:34
|
||||
IDEMPOTENCY_HEADER = "Idempotency-Key"
|
||||
REPLAYED_HEADER = "Idempotency-Replayed"
|
||||
|
||||
# 기록의 `##` 이름 → Studio 입력 칸. studio-form-map.md 가 정본이다
|
||||
FIELD_MAP = {
|
||||
"CASE": {"문제": "problem", "결론": "conclusion", "검증 환경": "environment",
|
||||
"재현 조건": "reproduction", "본문": "bodyMarkdown"},
|
||||
"CONCEPT": {"본문": "bodyMarkdown"},
|
||||
"REFERENCE": {"목적": "purpose", "규칙": "rules", "적용 조건": "appliesWhen",
|
||||
"예외": "exceptions", "예시": "examples"},
|
||||
"QUESTION": {"사실": "facts", "가정": "assumptions", "미지수": "unknowns",
|
||||
"제약": "constraints", "선택지": "options", "다음 검증": "nextVerification"},
|
||||
"PROJECT_DECISION": {"근거": "basis", "결정문": "decision", "판단 이유": "rationale",
|
||||
"영향": "impact"},
|
||||
}
|
||||
BODY_START, BODY_END = "<!-- body:start -->", "<!-- body:end -->"
|
||||
|
||||
|
||||
class Refused(Exception):
|
||||
"""어댑터가 스스로 거절한 것. 서버 오류가 아니다."""
|
||||
|
||||
|
||||
def _sha256_file(path: str) -> str:
|
||||
with open(path, "rb") as fh:
|
||||
return hashlib.sha256(fh.read()).hexdigest()
|
||||
|
||||
|
||||
def _sha256_text(text: str) -> str:
|
||||
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def idempotency_key(op: str, record_rel: str, payload: dict | None = None) -> str:
|
||||
"""기록 경로에서 유도한 안정 키.
|
||||
|
||||
프런트는 호출마다 `crypto.randomUUID()` 로 새 키를 만든다(`local-id.ts:8`). 그래서
|
||||
「만들다 타임아웃 → 다시 시도」가 문서를 둘 만든다. 서버의 멱등 기능이 정확히 막으라고
|
||||
만들어진 상황을 키 정책이 빠져나간다.
|
||||
|
||||
- **만들기**는 경로만으로 키를 만든다. 몇 번을 다시 시도해도 같은 키라 문서가 하나다.
|
||||
- **저장**은 경로와 보낼 내용을 함께 넣는다. 같은 내용의 재시도는 서버가 첫 결과를
|
||||
재생하고, 내용이 바뀌면 새 저장이 된다. 경로만 쓰면 두 번째 저장이 첫 결과로
|
||||
조용히 재생돼 버린다.
|
||||
"""
|
||||
base = _sha256_text(record_rel)[:32]
|
||||
key = f"studio-{op}-{base}"
|
||||
if payload is not None:
|
||||
key += "-" + _sha256_text(json.dumps(payload, ensure_ascii=False, sort_keys=True))[:32]
|
||||
if len(key) > MAX_KEY_LENGTH: # 서버가 200자 초과를 거절한다
|
||||
raise Refused(f"멱등 키가 {len(key)}자다. 서버 상한은 {MAX_KEY_LENGTH}자")
|
||||
return key
|
||||
|
||||
|
||||
def _front_matter(text: str) -> dict:
|
||||
if not text.startswith("---"):
|
||||
return {}
|
||||
end = text.find("\n---", 3)
|
||||
out: dict[str, str] = {}
|
||||
for line in text[3:end].splitlines():
|
||||
m = re.match(r"^([a-zA-Z_]+):\s*(.*)$", line)
|
||||
if m:
|
||||
out[m.group(1)] = m.group(2).strip().strip('"')
|
||||
return out
|
||||
|
||||
|
||||
def _sections(text: str) -> dict[str, str]:
|
||||
a, b = text.find(BODY_START), text.find(BODY_END)
|
||||
marks = []
|
||||
for m in re.finditer(r"^##\s+(.+)$", text, re.M):
|
||||
if a >= 0 <= b and a < m.start() < b:
|
||||
continue
|
||||
marks.append((m.group(1).strip(), m.end()))
|
||||
out: dict[str, str] = {}
|
||||
for i, (name, start) in enumerate(marks):
|
||||
stop = marks[i + 1][1] - len(f"## {marks[i + 1][0]}") if i + 1 < len(marks) else len(text)
|
||||
chunk = text[start:stop]
|
||||
if name == "본문":
|
||||
chunk = chunk.replace(BODY_START, "").replace(BODY_END, "")
|
||||
out[name] = chunk.strip()
|
||||
return out
|
||||
|
||||
|
||||
def _summary(text: str) -> str:
|
||||
m = re.search(r"^#\s+.+$", text, re.M)
|
||||
if not m:
|
||||
return ""
|
||||
after = re.split(r"^##\s", text[m.end():], maxsplit=1, flags=re.M)[0]
|
||||
for para in (p.strip() for p in after.split("\n\n")):
|
||||
if para and not para.startswith("<!--"):
|
||||
return para
|
||||
return ""
|
||||
|
||||
|
||||
def build_input(record_path: str) -> dict:
|
||||
"""기록 `.md` 를 `WorkingCopyInput` 으로. 없는 칸을 지어내지 않는다."""
|
||||
text = open(record_path, encoding="utf-8").read()
|
||||
fm = _front_matter(text)
|
||||
kind = (fm.get("kind") or "").upper()
|
||||
if kind not in FIELD_MAP:
|
||||
raise Refused(f"모르는 kind: {fm.get('kind')!r}")
|
||||
found = _sections(text)
|
||||
doc: dict = {
|
||||
"kind": kind,
|
||||
"title": fm.get("title", ""),
|
||||
"slug": fm.get("slug", ""),
|
||||
"summary": _summary(text),
|
||||
"topicId": None, # 이름→uuid 해석은 catalog 조회가 필요하다. 지어내지 않는다
|
||||
"projectId": None,
|
||||
"relations": [],
|
||||
}
|
||||
for section, field in FIELD_MAP[kind].items():
|
||||
if section in found:
|
||||
doc[field] = found[section]
|
||||
if kind == "CASE":
|
||||
doc["lastVerifiedOn"] = fm.get("lastVerifiedOn") or None
|
||||
return doc
|
||||
|
||||
|
||||
def approved(record_path: str, package_path: str) -> dict:
|
||||
"""검토를 지난 최종본만 통과시킨다.
|
||||
|
||||
묶음의 `target.sha256` 은 검토가 본 파일의 해시다. 지금 디스크의 파일이 그것과 다르면
|
||||
**검토 뒤에 바뀐 것**이라 그 판정을 이 파일에 붙일 수 없다.
|
||||
"""
|
||||
pkg = json.load(open(package_path, encoding="utf-8"))
|
||||
now = _sha256_file(record_path)
|
||||
want = (pkg.get("target") or {}).get("sha256")
|
||||
if not want:
|
||||
raise Refused(f"묶음에 target.sha256 이 없다: {package_path}")
|
||||
if want != now:
|
||||
raise Refused(
|
||||
"검토가 본 파일과 지금 파일이 다르다 — 그 판정을 이 파일에 붙일 수 없다\n"
|
||||
f" 검토 시점 {want}\n 지금 {now}")
|
||||
failed = [g["cmd"] for g in pkg.get("gates", []) if g.get("exit") not in (0, "0")]
|
||||
if failed:
|
||||
raise Refused("관문이 통과하지 못한 묶음이다:\n " + "\n ".join(failed))
|
||||
return pkg
|
||||
|
||||
|
||||
def _reject_forbidden_paths(plan: list[dict]) -> None:
|
||||
"""게시 경로가 계획에 들어 있으면 멈춘다.
|
||||
|
||||
한 번이라도 게시한 문서는 게시를 취소해도 삭제가 409 로 거절된다. 되돌릴 수 없는
|
||||
동작을 프롬프트가 아니라 코드가 막는다.
|
||||
"""
|
||||
for step in plan:
|
||||
if FORBIDDEN_PATH.search(step["path"]):
|
||||
raise Refused(f"게시 경로는 이 어댑터가 만들지 않는다: {step['path']}")
|
||||
|
||||
|
||||
def plan_requests(record_path: str, doc: dict, document_id: str | None,
|
||||
expected_version: int | None) -> list[dict]:
|
||||
"""보낼 요청을 그대로 적어 낸다. 보내지 않는다."""
|
||||
rel = os.path.relpath(os.path.abspath(record_path), ROOT)
|
||||
steps: list[dict] = []
|
||||
if not document_id:
|
||||
steps.append({
|
||||
"op": "create", "method": "POST", "path": f"{BASE_PATH}/documents",
|
||||
"headers": {IDEMPOTENCY_HEADER: idempotency_key("create", rel),
|
||||
"X-CSRF-TOKEN": "<env STUDIO_CSRF_TOKEN>"},
|
||||
"cookies": {"TECHLOG_SESSION": "<env STUDIO_SESSION_COOKIE>"},
|
||||
"body": doc,
|
||||
"expect": {"status": 201, "readHeader": REPLAYED_HEADER},
|
||||
})
|
||||
else:
|
||||
if expected_version is None:
|
||||
raise Refused(
|
||||
"expectedVersion 을 모른 채 저장하지 않는다. 먼저 GET 으로 현재 version 을 "
|
||||
"읽는다 — 안 보내면 서버가 0 으로 채워 사실상 항상 충돌한다"
|
||||
" (StudioDocumentController.java:196)")
|
||||
if expected_version < 1:
|
||||
raise Refused(f"expectedVersion 은 1 이상이어야 한다 (스키마 minimum 1): {expected_version}")
|
||||
body = {"expectedVersion": expected_version, "document": doc}
|
||||
steps.append({
|
||||
"op": "save", "method": "PUT", "path": f"{BASE_PATH}/documents/{document_id}",
|
||||
"headers": {IDEMPOTENCY_HEADER: idempotency_key("save", rel, body),
|
||||
"X-CSRF-TOKEN": "<env STUDIO_CSRF_TOKEN>"},
|
||||
"cookies": {"TECHLOG_SESSION": "<env STUDIO_SESSION_COOKIE>"},
|
||||
"body": body,
|
||||
"expect": {"status": 200, "readHeader": REPLAYED_HEADER,
|
||||
"onConflict": "VERSION_CONFLICT 409 — 덮어쓰지 않고 멈춘다"},
|
||||
})
|
||||
steps.append({
|
||||
"op": "verify", "method": "GET",
|
||||
"path": f"{BASE_PATH}/documents/{document_id or '<생성된 id>'}",
|
||||
"headers": {}, "cookies": {"TECHLOG_SESSION": "<env STUDIO_SESSION_COOKIE>"},
|
||||
"body": None,
|
||||
"expect": {"status": 200,
|
||||
"compare": "정규화한 본문·칸·자료를 보낸 것과 견준다",
|
||||
"private": "currentPublication 이 null 이거나 status != PUBLISHED"},
|
||||
})
|
||||
_reject_forbidden_paths(steps)
|
||||
return steps
|
||||
|
||||
|
||||
# 이 어댑터가 값을 대조하지 않는 칸. 서버가 이름을 uuid 로 바꿔 돌려주거나 목록을 다시
|
||||
# 정렬하는 자리라 문자열로 견줄 수 없다. **빼먹은 것이 아니라 못 보는 것**이므로
|
||||
# 보고서에 그대로 적는다. 조용히 건너뛰면 「전부 같다」가 「본 것만 같다」를 가린다
|
||||
NOT_COMPARED = {
|
||||
"topicId": "이름→uuid 해석을 이 어댑터가 하지 않는다",
|
||||
"projectId": "같음",
|
||||
"relations": "서버가 순서를 다시 매긴다. 목록 비교는 따로 만들어야 한다",
|
||||
}
|
||||
|
||||
|
||||
def normalize(value) -> str:
|
||||
"""되읽어 견주기 전에 줄 끝 공백과 줄바꿈 표기만 맞춘다.
|
||||
|
||||
**정규화가 차이를 지울 수 있다.** 그래서 이 함수는 뜻이 바뀌지 않는 것만 건드리고,
|
||||
`compare_saved` 는 정규화한 값과 **원문 둘 다** 견준다. 정규화 뒤에만 같아지는 차이는
|
||||
통과가 아니라 `whitespace-only` 로 보고한다.
|
||||
"""
|
||||
if value is None:
|
||||
return ""
|
||||
text = str(value).replace("\r\n", "\n")
|
||||
return "\n".join(line.rstrip() for line in text.split("\n")).strip()
|
||||
|
||||
|
||||
def compare_saved(sent: dict, fetched: dict) -> dict:
|
||||
"""보낸 것과 되읽은 것을 견준다.
|
||||
|
||||
문자열 비교 하나로 「같다」를 판정하지 않는다. 세 가지를 나눠 낸다.
|
||||
|
||||
- `differences` — 정규화한 뒤에도 다르다. 값이 잘렸거나 바뀐 것이다
|
||||
- `whitespaceOnly` — 정규화 뒤에만 같아진다. 통과로 세지 않는다
|
||||
- `notCompared` · `unexpected` — 이 어댑터가 못 본 칸과 서버가 더 준 칸
|
||||
|
||||
**잘림은 길이로 한 번 더 본다.** Studio 는 제목 30/31자·본문 4157/4187자로 조용히 자른
|
||||
적이 있다. 앞이 같고 뒤가 없는 모양은 바뀐 것과 다르게 읽힌다.
|
||||
"""
|
||||
differences, whitespace_only, truncated = [], [], []
|
||||
for field, want in sent.items():
|
||||
if field in NOT_COMPARED:
|
||||
continue
|
||||
got = fetched.get(field)
|
||||
nw, ng = normalize(want), normalize(got)
|
||||
raw_same = str(want or "") == str(got or "")
|
||||
if nw != ng:
|
||||
entry = {"field": field, "sentChars": len(nw), "gotChars": len(ng),
|
||||
"sent": nw[:120], "got": ng[:120]}
|
||||
if ng and nw.startswith(ng):
|
||||
entry["shape"] = "잘림 — 되읽은 값이 보낸 값의 앞부분이다"
|
||||
truncated.append(entry)
|
||||
differences.append(entry)
|
||||
elif not raw_same:
|
||||
whitespace_only.append({"field": field,
|
||||
"note": "정규화 뒤에만 같다. 공백이나 줄바꿈이 달라졌다"})
|
||||
unexpected = [k for k in fetched
|
||||
if k not in sent and k not in NOT_COMPARED and k not in ("id", "version")]
|
||||
return {
|
||||
"same": not differences and not whitespace_only,
|
||||
"differences": differences,
|
||||
"truncated": truncated,
|
||||
"whitespaceOnly": whitespace_only,
|
||||
"notCompared": NOT_COMPARED,
|
||||
"unexpected": unexpected,
|
||||
"comparedFields": [f for f in sent if f not in NOT_COMPARED],
|
||||
}
|
||||
|
||||
|
||||
def is_private(detail: dict) -> tuple[bool, str]:
|
||||
pub = detail.get("currentPublication")
|
||||
if pub is None:
|
||||
return True, "currentPublication 이 없다"
|
||||
status = pub.get("status")
|
||||
if status == "PUBLISHED":
|
||||
return False, f"이미 게시된 문서다 — publicPath={pub.get('publicPath')}"
|
||||
return True, f"게시 상태가 {status} 다"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="Studio 저장 어댑터. 게시하지 않는다.")
|
||||
ap.add_argument("--record", required=True)
|
||||
ap.add_argument("--package", required=True, help="review-package.py 가 만든 묶음")
|
||||
ap.add_argument("--document-id", help="이미 있는 문서. 없으면 만들기 계획을 낸다")
|
||||
ap.add_argument("--expected-version", type=int, help="GET 으로 읽은 현재 version")
|
||||
ap.add_argument("-o", "--out", help="계획을 적을 파일")
|
||||
ap.add_argument("--send", action="store_true", help="실제로 보낸다 (지금은 막혀 있다)")
|
||||
args = ap.parse_args()
|
||||
|
||||
try:
|
||||
for p in (args.record, args.package):
|
||||
if not os.path.isfile(p):
|
||||
raise Refused(f"그런 파일이 없다: {p}")
|
||||
pkg = approved(args.record, args.package)
|
||||
doc = build_input(args.record)
|
||||
steps = plan_requests(args.record, doc, args.document_id, args.expected_version)
|
||||
|
||||
if args.send and not UNATTENDED_SAVE_ENABLED:
|
||||
raise Refused("무인 저장이 꺼져 있다 — " + UNATTENDED_SAVE_BLOCKED_BY)
|
||||
except Refused as e:
|
||||
print(f"거절: {e}", file=sys.stderr)
|
||||
return 3
|
||||
|
||||
plan = {
|
||||
"schemaVersion": 1,
|
||||
"mode": "plan",
|
||||
"unattendedSaveEnabled": UNATTENDED_SAVE_ENABLED,
|
||||
"unattendedSaveBlockedBy": UNATTENDED_SAVE_BLOCKED_BY,
|
||||
"record": os.path.relpath(os.path.abspath(args.record), ROOT),
|
||||
"recordSha256": _sha256_file(args.record),
|
||||
"packageSha256": _sha256_file(args.package),
|
||||
"approvedAgainst": pkg["target"]["sha256"],
|
||||
"checkerVersions": pkg.get("checkerVersions"),
|
||||
"requests": steps,
|
||||
"notes": [
|
||||
"이 파일의 요청을 아직 보내지 않았다. mode 가 plan 이다.",
|
||||
"게시·게시취소 경로는 이 어댑터가 만들지 않는다. 계획에 들어 있으면 거절한다.",
|
||||
"자격증명은 환경 변수 자리표시자로만 적혀 있다. 값은 이 저장소에 없다.",
|
||||
],
|
||||
}
|
||||
text = json.dumps(plan, ensure_ascii=False, indent=2) + "\n"
|
||||
if args.out:
|
||||
os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
|
||||
open(args.out, "w", encoding="utf-8").write(text)
|
||||
ops = " · ".join(f"{s['op']} {s['method']}" for s in steps)
|
||||
print(f"{args.out} — {ops} · 무인 저장 {'켬' if UNATTENDED_SAVE_ENABLED else '끔'}")
|
||||
else:
|
||||
print(text, end="")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,194 @@
|
||||
"""Studio 저장 어댑터.
|
||||
|
||||
이 묶음의 절반은 **대조군**이다. 손대지 않은 쌍이 「같다」로 나오는 것을 먼저 확인하지 않으면,
|
||||
결함을 잡았다는 결과가 비교 틀이 만들어 낸 것인지 알 수 없다.
|
||||
"""
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
_spec = importlib.util.spec_from_file_location(
|
||||
"studio_save", os.path.join(ROOT, "scripts", "studio-save.py"))
|
||||
ss = importlib.util.module_from_spec(_spec)
|
||||
_spec.loader.exec_module(ss)
|
||||
|
||||
SENT = {
|
||||
"kind": "CASE",
|
||||
"title": "파이프 뒤의 종료 코드",
|
||||
"slug": "exit-code-read-behind-a-pipe",
|
||||
"summary": "열넷 중 둘이 exit 1 이었다.",
|
||||
"problem": "이 기록의 증거 다섯 개도 그렇게 수집했다.",
|
||||
"bodyMarkdown": "본문이 길다. " * 40,
|
||||
"topicId": None,
|
||||
"projectId": None,
|
||||
"relations": [],
|
||||
}
|
||||
|
||||
|
||||
class ComparatorControlTest(unittest.TestCase):
|
||||
"""비교 틀이 한쪽으로 굳어 있지 않은지."""
|
||||
|
||||
def test_an_untouched_pair_compares_equal(self):
|
||||
"""대조군. 이것이 실패하면 아래의 모든 「잡았다」가 틀 탓이다."""
|
||||
r = ss.compare_saved(SENT, dict(SENT))
|
||||
self.assertTrue(r["same"], r)
|
||||
self.assertEqual([], r["differences"])
|
||||
self.assertEqual([], r["whitespaceOnly"])
|
||||
|
||||
def test_a_korean_numeral_change_is_caught(self):
|
||||
"""「여섯 → 다섯」. check-preservation 은 아라비아 숫자만 봐서 이걸 못 잡는다.
|
||||
|
||||
저장 대조는 칸 전체를 견주므로 잡힌다. 두 검사기가 보는 것이 다르다.
|
||||
"""
|
||||
r = ss.compare_saved(SENT, {**SENT, "problem": "이 기록의 증거 여섯 개도 그렇게 수집했다."})
|
||||
self.assertFalse(r["same"])
|
||||
self.assertEqual(["problem"], [d["field"] for d in r["differences"]])
|
||||
|
||||
def test_a_silently_truncated_value_is_named_as_truncation(self):
|
||||
"""Studio 는 제목·본문을 조용히 자른 적이 있다. 바뀐 것과 잘린 것은 다르게 읽힌다."""
|
||||
short = SENT["bodyMarkdown"][:60]
|
||||
r = ss.compare_saved(SENT, {**SENT, "bodyMarkdown": short})
|
||||
self.assertFalse(r["same"])
|
||||
self.assertEqual(1, len(r["truncated"]))
|
||||
self.assertGreater(r["truncated"][0]["sentChars"], r["truncated"][0]["gotChars"])
|
||||
|
||||
def test_a_whitespace_only_difference_is_not_counted_as_same(self):
|
||||
"""정규화가 지운 차이를 통과로 세지 않는다."""
|
||||
r = ss.compare_saved(SENT, {**SENT, "title": SENT["title"] + " "})
|
||||
self.assertFalse(r["same"])
|
||||
self.assertEqual(["title"], [w["field"] for w in r["whitespaceOnly"]])
|
||||
self.assertEqual([], r["differences"])
|
||||
|
||||
def test_fields_it_cannot_compare_are_named_not_skipped(self):
|
||||
"""못 본 칸을 조용히 건너뛰면 「전부 같다」가 「본 것만 같다」를 가린다."""
|
||||
r = ss.compare_saved(SENT, dict(SENT))
|
||||
self.assertEqual({"topicId", "projectId", "relations"}, set(r["notCompared"]))
|
||||
self.assertNotIn("topicId", r["comparedFields"])
|
||||
|
||||
def test_a_field_the_server_added_is_reported(self):
|
||||
r = ss.compare_saved(SENT, {**SENT, "publishedAt": "2026-09-10"})
|
||||
self.assertIn("publishedAt", r["unexpected"])
|
||||
|
||||
|
||||
class IdempotencyKeyTest(unittest.TestCase):
|
||||
def test_create_key_is_stable_across_retries(self):
|
||||
"""재시도가 문서를 둘 만들지 않게 하는 것이 이 키의 목적이다."""
|
||||
a = ss.idempotency_key("create", "docs/p/tech-log-studio/t/case/x.md")
|
||||
b = ss.idempotency_key("create", "docs/p/tech-log-studio/t/case/x.md")
|
||||
self.assertEqual(a, b)
|
||||
|
||||
def test_create_keys_differ_per_record(self):
|
||||
self.assertNotEqual(
|
||||
ss.idempotency_key("create", "docs/p/t/case/x.md"),
|
||||
ss.idempotency_key("create", "docs/p/t/case/y.md"))
|
||||
|
||||
def test_save_key_follows_the_payload(self):
|
||||
"""같은 내용의 재시도는 재생되고, 내용이 바뀌면 새 저장이 된다.
|
||||
|
||||
경로만으로 키를 만들면 두 번째 저장이 첫 결과로 조용히 재생돼 버린다.
|
||||
"""
|
||||
rel = "docs/p/t/case/x.md"
|
||||
one = {"expectedVersion": 3, "document": {"title": "가"}}
|
||||
two = {"expectedVersion": 3, "document": {"title": "나"}}
|
||||
self.assertEqual(ss.idempotency_key("save", rel, one),
|
||||
ss.idempotency_key("save", rel, one))
|
||||
self.assertNotEqual(ss.idempotency_key("save", rel, one),
|
||||
ss.idempotency_key("save", rel, two))
|
||||
|
||||
def test_key_stays_within_the_server_limit(self):
|
||||
rel = "docs/" + "a" * 300 + "/x.md"
|
||||
key = ss.idempotency_key("save", rel, {"expectedVersion": 1, "document": SENT})
|
||||
self.assertLessEqual(len(key), ss.MAX_KEY_LENGTH)
|
||||
|
||||
|
||||
class SafetyTest(unittest.TestCase):
|
||||
def test_no_plan_may_contain_a_publish_path(self):
|
||||
"""게시는 되돌릴 수 없다. 프롬프트가 아니라 코드가 막는다."""
|
||||
with self.assertRaises(ss.Refused):
|
||||
ss._reject_forbidden_paths(
|
||||
[{"path": "/api/v1/studio/documents/abc/publish"}])
|
||||
with self.assertRaises(ss.Refused):
|
||||
ss._reject_forbidden_paths(
|
||||
[{"path": "/api/v1/studio/publications/abc/unpublish"}])
|
||||
|
||||
def test_the_generated_plan_has_no_publish_path(self):
|
||||
steps = ss.plan_requests("docs/p/t/case/x.md", dict(SENT), None, None)
|
||||
for s in steps:
|
||||
self.assertNotIn("publish", s["path"])
|
||||
|
||||
def test_saving_without_expected_version_is_refused(self):
|
||||
"""안 보내면 서버가 0 으로 채워 사실상 항상 충돌한다. 빈 값은 「검사 안 함」이 아니다."""
|
||||
with self.assertRaises(ss.Refused):
|
||||
ss.plan_requests("docs/p/t/case/x.md", dict(SENT), "doc-1", None)
|
||||
|
||||
def test_expected_version_below_one_is_refused(self):
|
||||
with self.assertRaises(ss.Refused):
|
||||
ss.plan_requests("docs/p/t/case/x.md", dict(SENT), "doc-1", 0)
|
||||
|
||||
def test_unattended_save_is_off(self):
|
||||
self.assertFalse(ss.UNATTENDED_SAVE_ENABLED)
|
||||
self.assertIn("CR-001", ss.UNATTENDED_SAVE_BLOCKED_BY)
|
||||
|
||||
|
||||
class PrivacyTest(unittest.TestCase):
|
||||
def test_a_document_with_no_publication_is_private(self):
|
||||
ok, why = ss.is_private({"currentPublication": None})
|
||||
self.assertTrue(ok, why)
|
||||
|
||||
def test_a_published_document_is_not_private(self):
|
||||
ok, why = ss.is_private(
|
||||
{"currentPublication": {"status": "PUBLISHED", "publicPath": "/x"}})
|
||||
self.assertFalse(ok)
|
||||
self.assertIn("/x", why)
|
||||
|
||||
def test_an_unpublished_document_is_private_again(self):
|
||||
ok, _ = ss.is_private({"currentPublication": {"status": "UNPUBLISHED"}})
|
||||
self.assertTrue(ok)
|
||||
|
||||
|
||||
class ApprovalTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.dir = tempfile.mkdtemp()
|
||||
self.record = os.path.join(self.dir, "r.md")
|
||||
open(self.record, "w", encoding="utf-8").write("---\nkind: CASE\n---\n\n# 제목\n\n요약.\n")
|
||||
self.pkg = os.path.join(self.dir, "p.json")
|
||||
|
||||
def _write_pkg(self, sha, gates):
|
||||
json.dump({"target": {"sha256": sha}, "gates": gates},
|
||||
open(self.pkg, "w", encoding="utf-8"))
|
||||
|
||||
def test_a_package_that_matches_passes(self):
|
||||
self._write_pkg(ss._sha256_file(self.record), [{"cmd": "x", "exit": 0}])
|
||||
self.assertIn("target", ss.approved(self.record, self.pkg))
|
||||
|
||||
def test_a_record_changed_after_review_is_refused(self):
|
||||
"""검토가 본 파일과 지금 파일이 다르면 그 판정을 이 파일에 붙일 수 없다."""
|
||||
self._write_pkg(ss._sha256_file(self.record), [{"cmd": "x", "exit": 0}])
|
||||
open(self.record, "a", encoding="utf-8").write("한 줄 더.\n")
|
||||
with self.assertRaises(ss.Refused):
|
||||
ss.approved(self.record, self.pkg)
|
||||
|
||||
def test_a_package_with_a_failed_gate_is_refused(self):
|
||||
self._write_pkg(ss._sha256_file(self.record), [{"cmd": "check_body", "exit": 1}])
|
||||
with self.assertRaises(ss.Refused):
|
||||
ss.approved(self.record, self.pkg)
|
||||
|
||||
|
||||
class CliTest(unittest.TestCase):
|
||||
def _cli(self, *args):
|
||||
return subprocess.run(
|
||||
["python3", os.path.join(ROOT, "scripts", "studio-save.py"), *args],
|
||||
cwd=ROOT, capture_output=True, text=True)
|
||||
|
||||
def test_send_is_refused_while_unattended_save_is_off(self):
|
||||
p = self._cli("--record", "nope.md", "--package", "nope.json", "--send")
|
||||
self.assertEqual(3, p.returncode)
|
||||
self.assertIn("거절", p.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user