fix(rework): require verified completion comment receipt
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
---
|
||||
name: gitea-issue-rework
|
||||
description: Gitea PR 검토 지적을 계약에 맞춰 재작업하고 검증한다.
|
||||
version: 0.2.11
|
||||
version: 0.2.12
|
||||
author: genichin, Hermes Agent
|
||||
license: MIT
|
||||
platforms: [linux]
|
||||
@@ -122,6 +122,8 @@ reason: <bounded 판정 근거>
|
||||
|
||||
재작업 branch를 전달한 뒤 reviewer가 새 head를 자동으로 재검토할 수 있도록 exact PR의 top-level conversation에 comment 하나를 게시한다. Adapter target은 object kind가 pull request이고 index가 exact `<pull-number>`인 record여야 한다. Provider가 PR comment를 underlying issue comment로 저장하면 exact comment endpoint target의 issue index가 `<pull-number>`이고 fresh PR metadata가 그 record를 같은 pull request로 식별함을 증명해야 한다. Comment object가 target kind를 반복하지 않는 API에서는 검증된 endpoint target과 PR metadata binding을 사용한다. Source Issue `<issue-number>`, inline comment, review reply 또는 native review로 대체하지 않는다. Marker를 제외한 body를 먼저 canonical UTF-8/LF bytes로 직렬화하고 그 SHA-256을 `result-sha256`으로 계산한다.
|
||||
|
||||
Canonical body와 marker는 `scripts/rework_comment.py render`로만 생성하고, 게시 또는 exact comment 재사용 뒤에는 authoritative normalized readback을 `scripts/rework_comment.py verify`로 검증한다. `status=verified`와 `readback_verified=true`인 `gitea-rework-handoff/v1` receipt가 없으면 완료 조건을 충족하지 않으며 PR body update나 write 성공 response는 receipt를 대체하지 않는다.
|
||||
|
||||
```markdown
|
||||
## 구현 재작업 결과
|
||||
|
||||
@@ -253,13 +255,13 @@ Reviewer가 rework head를 새 snapshot으로 다시 검토해야 하며 이 com
|
||||
3. Push 결과만 신뢰하지 않고 remote branch SHA와 PR head SHA가 final local HEAD와 exact-match하는지 readback한다.
|
||||
4. PR body 갱신이 필요한 경우 write 직전에 PR 전체 필드, review·comment·commit collection cardinality와 authenticated impl identity를 다시 읽는다. Current body의 `Closes #<issue-number>`와 기존 구현 내용을 보존하고 재작업 review ID·reviewed head·finding별 조치·최종 검증과 unresolved limitation만 반영한 exact body payload를 명시해 한 번만 호출한다. 성공 응답만 신뢰하지 않고 fresh PR과 관련 collection을 읽어 PR author stable account ID, exact body payload, `updated_at`, base·head·commit/ref, state·draft·merge 상태, labels·assignees·reviewer와 collection cardinality를 baseline과 비교한다. Exact body와 provider가 해당 body update로 파생한 `updated_at` 외의 변경은 승인되지 않았으며 인과를 증명할 수 없으면 partial로 중단한다.
|
||||
5. PR changed files, commits, state, merge 상태, statuses와 workflow runs를 다시 읽는다. Local test, commit status와 Actions run을 별도 사실로 기록한다.
|
||||
6. Finding matrix와 최종 검증 기록으로 canonical 재작업 결과 comment를 렌더링하고 marker-free digest와 correlation을 계산·검증한다.
|
||||
6. Finding matrix와 최종 검증 기록을 bounded JSON input으로 만들고 `scripts/rework_comment.py render --input <input.json> --comment <comment.md> --expected <expected.json>`를 실행해 canonical 재작업 결과 comment, marker-free digest와 correlation을 생성·검증한다. Comment body와 marker를 수동 조립하지 않는다.
|
||||
7. Exact PR top-level comment endpoint target과 fresh PR metadata binding을 검증해 Source Issue comments, inline comments, review replies와 native reviews에서 분리하고 scoped collection을 complete pagination한 뒤 canonical deduplication 규칙을 적용한다. Existing exact comment가 없으면 exact PR conversation에 body를 한 번 게시하고, 있으면 write 없이 재사용한다.
|
||||
8. Exact comment와 scoped complete collection을 fresh readback해 ID, impl author stable account ID, body, timestamp, verified endpoint target, PR metadata binding, final head binding, digest, correlation과 cardinality를 검증한다. 신규 게시로 허용된 PR-underlying-issue projection 외의 PR field, Source Issue field와 기존 review collection에 변경이 없어야 한다.
|
||||
8. Exact comment와 scoped complete collection을 fresh normalized JSON으로 readback한 뒤 `scripts/rework_comment.py verify --expected <expected.json> --pr-readback <pr.json> --comments-readback <comments.json> --identity-readback <identity.json> --action <create|reuse> --baseline-count <N> --output <handoff-receipt.json>`를 실행한다. Receipt의 comment ID, impl author stable account ID, body, timestamp, final head, digest, correlation과 cardinality가 검증돼야 하며, 신규 게시로 허용된 PR-underlying-issue projection 외의 PR field, Source Issue field와 기존 review collection에 변경이 없어야 한다.
|
||||
9. Native review를 작성·dismiss·reply하거나 새 review를 요청하지 않는다. 기존 `REQUEST_CHANGES`는 reviewer가 새 head를 재검토하기 전까지 자동 해소된 것으로 주장하지 않는다.
|
||||
10. Push, PR body 또는 comment write가 timeout·오류이면 blind retry하지 않고 branch, PR과 complete comment collection을 authoritative하게 reconcile한다. Exact write가 반영됐음을 증명할 수 없으면 partial 상태로 중단한다.
|
||||
|
||||
완료 조건: 같은 PR의 remote head가 final local HEAD와 일치하고 canonical 재작업 결과 comment가 정확히 하나 readback됐으며, 모든 authorized external write와 후속 review 필요성이 명확하다.
|
||||
완료 조건: 같은 PR의 remote head가 final local HEAD와 일치하고 canonical 재작업 결과 comment가 정확히 하나 readback됐으며, `gitea-rework-handoff/v1` receipt가 `status=verified`, `readback_verified=true`로 생성되고 모든 authorized external write와 후속 review 필요성이 명확하다. Receipt가 없거나 검증에 실패하면 lock은 정상 해제하되 결과를 `partial|failed`로 보고하며 `completed`를 주장하지 않는다.
|
||||
|
||||
### 8. 종료와 다음 review handoff
|
||||
|
||||
@@ -307,7 +309,7 @@ Reviewer가 rework head를 새 snapshot으로 다시 검토해야 하며 이 com
|
||||
- Base 및 reviewed head 대비 전체 modified file과 범위 근거
|
||||
- Local commit, remote branch와 PR final head SHA readback
|
||||
- PR body write 여부, status와 Actions run의 success·failure·pending·없음 상태
|
||||
- 재작업 결과 comment ID·URL·author·result·digest·correlation과 exact final head binding
|
||||
- 재작업 결과 comment ID·URL·author·result·digest·correlation, exact final head binding과 verified handoff receipt
|
||||
- 기존 review가 유지됨과 새 head의 후속 review 필요 여부
|
||||
- 수행하지 않은 review·merge·Issue·release·deployment write
|
||||
- Baseline failure, unresolved finding, specification blocker, unavailable evidence와 partial write
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fail-closed canonical rework comment renderer and verifier."""
|
||||
from __future__ import annotations
|
||||
import argparse, hashlib, json, os, re, sys, tempfile
|
||||
from pathlib import Path
|
||||
|
||||
MAX = 1_000_000
|
||||
SHA = re.compile(r"^[0-9a-f]{40}$")
|
||||
DIGEST = re.compile(r"^[0-9a-f]{64}$")
|
||||
ID = re.compile(r"^[1-9][0-9]*$")
|
||||
MARK = "<!-- gitea-agent:implementation-rework/v1 "
|
||||
|
||||
class Bad(ValueError): pass
|
||||
|
||||
def read_json(p):
|
||||
b=Path(p).read_bytes()
|
||||
if len(b)>MAX: raise Bad("input too large")
|
||||
try: x=json.loads(b)
|
||||
except Exception as e: raise Bad("invalid json") from e
|
||||
if not isinstance(x,dict): raise Bad("object required")
|
||||
return x
|
||||
|
||||
def read_json_any(p):
|
||||
b=Path(p).read_bytes()
|
||||
if len(b)>MAX: raise Bad("input too large")
|
||||
try: return json.loads(b)
|
||||
except Exception as e: raise Bad("invalid json") from e
|
||||
|
||||
def text(x, name, pattern=None, maxlen=4096):
|
||||
if not isinstance(x,str) or not x or len(x)>maxlen or "\r" in x or "\n" in x or x.rstrip()!=x: raise Bad(f"invalid {name}")
|
||||
if pattern and not pattern.fullmatch(x): raise Bad(f"invalid {name}")
|
||||
return x
|
||||
|
||||
def number(x,name):
|
||||
if isinstance(x,bool) or not isinstance(x,int) or x<1: raise Bad(f"invalid {name}")
|
||||
return str(x)
|
||||
|
||||
def safe_human(s,name):
|
||||
if not isinstance(s,str) or not s or len(s)>10000 or "\r" in s or "\n" in s or s.rstrip()!=s: raise Bad(f"invalid {name}")
|
||||
# Keep the renderer deliberately conservative: no markup, mentions, URLs,
|
||||
# credentials, shell-like payloads, or HTML in untrusted fields.
|
||||
if re.search(r"<[^>]*>|(?:https?://|www\.)|(?:^|\s)@[A-Za-z0-9_.-]+|(?:password|passwd|secret|token|api[_ -]?key|authorization|bearer|credential)\s*[:=]",s,re.I): raise Bad(f"unsafe {name}")
|
||||
return s
|
||||
|
||||
def canon(d):
|
||||
for k in ("owner","repository"): text(d.get(k),k,re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$"))
|
||||
for k in ("reviewed_head","final_head"): text(d.get(k),k,SHA)
|
||||
text(d.get("contract_sha256"),"contract_sha256",DIGEST)
|
||||
vals={k:number(d.get(k),k) for k in ("issue","pr","review")}
|
||||
result=d.get("result")
|
||||
if result not in ("rework_delivered","partial"): raise Bad("invalid result")
|
||||
fs=d.get("findings")
|
||||
if not isinstance(fs,list) or not fs or len(fs)>100: raise Bad("invalid findings")
|
||||
out=[]; seen=set()
|
||||
for f in fs:
|
||||
if not isinstance(f,dict): raise Bad("invalid finding")
|
||||
fid=text(f.get("id"),"finding id",re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$"))
|
||||
if fid in seen: raise Bad("duplicate finding")
|
||||
seen.add(fid)
|
||||
status=f.get("status"); cls=f.get("classification")
|
||||
if status not in ("resolved","unresolved") or cls not in ("valid_fix","evidence_gap","reviewer_misunderstanding"): raise Bad("invalid finding enum")
|
||||
out.append((fid,status,cls,safe_human(f.get("changes"),"changes"),safe_human(f.get("verification"),"verification"),safe_human(f.get("reason"),"reason")))
|
||||
if result=="rework_delivered" and any(status!="resolved" for _,status,cls,_,_,_ in out if cls in ("valid_fix","evidence_gap")): raise Bad("rework_delivered requires resolved actionable findings")
|
||||
extras={k:safe_human(d.get(k),k) for k in ("local_checks","runtime_browser")}
|
||||
cs=d.get("commit_status"); ar=d.get("actions_runs")
|
||||
if cs not in ("success","failure","pending","없음","조회 불가") or ar not in ("success","failure","pending","없음","조회 불가"): raise Bad("invalid check status")
|
||||
return d, vals, out, extras, cs, ar
|
||||
|
||||
def render_body(d,vals,findings,extras,cs,ar):
|
||||
lines=["## 구현 재작업 결과","",f"- Source Issue: #{vals['issue']}",f"- Pull Request: #{vals['pr']}",f"- Source Review: #{vals['review']}",f"- Reviewed head: `{d['reviewed_head']}`",f"- Rework head: `{d['final_head']}`",f"- Contract SHA-256: `{d['contract_sha256']}`",f"- Result: `{d['result']}`","","### 지적 처리",""]
|
||||
for fid,status,cls,ch,ver,reason in findings:
|
||||
lines += [f"- {fid} — `{status}`",f" - 분류: {cls}",f" - 변경: {ch}",f" - 검증: {ver}",f" - 근거: {reason}"]
|
||||
lines += ["","### 최종 검증","",f"- Local checks: {extras['local_checks']}",f"- Runtime/browser: {extras['runtime_browser']}",f"- Commit status: {cs}",f"- Actions runs: {ar}","","### 다음 단계","","Reviewer가 rework head를 새 snapshot으로 다시 검토해야 하며 이 comment는 기존 `REQUEST_CHANGES`를 해제하거나 승인을 의미하지 않는다.",""]
|
||||
return "\n".join(lines)
|
||||
|
||||
def atomic(path,data):
|
||||
p=Path(path); p.parent.mkdir(parents=True,exist_ok=True)
|
||||
fd,tmp=tempfile.mkstemp(prefix=".rework-",dir=str(p.parent));
|
||||
try:
|
||||
with os.fdopen(fd,"wb") as f: f.write(data); f.flush(); os.fsync(f.fileno())
|
||||
os.replace(tmp,p)
|
||||
finally:
|
||||
if os.path.exists(tmp): os.unlink(tmp)
|
||||
|
||||
def do_render(a):
|
||||
d,vals,fs,ex,cs,ar=canon(read_json(a.input))
|
||||
body=render_body(d,vals,fs,ex,cs,ar); bb=body.encode()
|
||||
rh=hashlib.sha256(bb).hexdigest()
|
||||
corr=hashlib.sha256("\0".join([d['owner'],d['repository'],vals['issue'],vals['pr'],vals['review'],d['reviewed_head'],d['final_head'],d['contract_sha256']]).encode()).hexdigest()
|
||||
marker=f"{MARK}owner={d['owner']} repository={d['repository']} issue={vals['issue']} pr={vals['pr']} review={vals['review']} reviewed-head={d['reviewed_head']} head={d['final_head']} contract-sha256={d['contract_sha256']} result-sha256={rh} correlation={corr} -->\n"
|
||||
atomic(a.comment,(body+marker).encode())
|
||||
expected={"body":body+marker,"marker":marker,"marker_free_body":body,"result_sha256":rh,"correlation":corr,"target":{"owner":d['owner'],"repository":d['repository'],"issue":int(vals['issue']),"pr":int(vals['pr'])},"review":int(vals['review']),"reviewed_head":d['reviewed_head'],"head":d['final_head'],"contract_sha256":d['contract_sha256']}
|
||||
atomic(a.expected,(json.dumps(expected,ensure_ascii=False,sort_keys=True,separators=(",",":"))+"\n").encode())
|
||||
|
||||
def do_verify(a):
|
||||
exp=read_json(a.expected); pr=read_json(a.pr_readback); comment_readback=read_json(a.comments_readback); ident=read_json(a.identity_readback)
|
||||
body=exp.get("body"); rh=exp.get("result_sha256"); corr=exp.get("correlation"); head=exp.get("head")
|
||||
marker=exp.get("marker","")
|
||||
marker_free=exp.get("marker_free_body", body[:-len(marker)] if marker and body.endswith(marker) else body)
|
||||
target=exp.get("target")
|
||||
if not isinstance(target,dict): raise Bad("expected target")
|
||||
owner=text(target.get("owner"),"owner",re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$"))
|
||||
repository=text(target.get("repository"),"repository",re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$"))
|
||||
issue=number(target.get("issue"),"issue"); pr_number=number(target.get("pr"),"pr"); review=number(exp.get("review"),"review")
|
||||
reviewed_head=text(exp.get("reviewed_head"),"reviewed_head",SHA); head=text(head,"head",SHA); contract=text(exp.get("contract_sha256"),"contract_sha256",DIGEST)
|
||||
computed_corr=hashlib.sha256("\0".join([owner,repository,issue,pr_number,review,reviewed_head,head,contract]).encode()).hexdigest()
|
||||
expected_marker=f"{MARK}owner={owner} repository={repository} issue={issue} pr={pr_number} review={review} reviewed-head={reviewed_head} head={head} contract-sha256={contract} result-sha256={rh} correlation={computed_corr} -->\n"
|
||||
if not isinstance(body,str) or body.encode().decode()!=body or not isinstance(marker_free,str) or body != marker_free+marker or hashlib.sha256(marker_free.encode()).hexdigest()!=rh or not DIGEST.fullmatch(rh or "") or corr!=computed_corr or marker!=expected_marker: raise Bad("expected digest")
|
||||
phead=pr.get("head",{}).get("sha") if isinstance(pr.get("head"),dict) else None
|
||||
if pr.get("object_kind")!="pull_request" or pr.get("owner")!=owner or pr.get("repository")!=repository or pr.get("number")!=int(pr_number) or phead!=head or pr.get("state")!="open" or pr.get("metadata_binding_verified") is not True or not pr.get("html_url"): raise Bad("PR mismatch")
|
||||
aid=ident.get("id"); login=ident.get("login")
|
||||
if isinstance(aid,bool) or not isinstance(aid,int) or aid<1 or not isinstance(login,str) or not login or ident.get("role")!="gitea-impl-agent" or ident.get("authenticated") is not True or ident.get("stable_account_verified") is not True: raise Bad("identity mismatch")
|
||||
if comment_readback.get("object_kind")!="pull_request_top_level_comments" or comment_readback.get("owner")!=owner or comment_readback.get("repository")!=repository or comment_readback.get("pr")!=int(pr_number) or comment_readback.get("complete") is not True or comment_readback.get("endpoint_binding_verified") is not True or comment_readback.get("source_issue_unchanged") is not True or comment_readback.get("review_collection_unchanged") is not True or comment_readback.get("pr_metadata_unchanged") is not True: raise Bad("comments scope")
|
||||
comments=comment_readback.get("comments")
|
||||
if not isinstance(comments,list) or len(comments)>10000 or comment_readback.get("total_count")!=len(comments) or isinstance(comment_readback.get("page_count"),bool) or not isinstance(comment_readback.get("page_count"),int) or comment_readback["page_count"]<1: raise Bad("comments incomplete")
|
||||
provider_counts=comment_readback.get("provider_counts")
|
||||
if not isinstance(provider_counts,dict): raise Bad("provider counts")
|
||||
count_names=("top_level","reviews","inline")
|
||||
if any(isinstance(provider_counts.get(k),bool) or not isinstance(provider_counts.get(k),int) or provider_counts[k]<0 for k in (*count_names,"aggregate_comments")): raise Bad("provider counts")
|
||||
components=provider_counts.get("aggregate_components")
|
||||
if not isinstance(components,list) or not components or len(set(components))!=len(components) or any(k not in count_names for k in components) or provider_counts["top_level"]!=len(comments) or provider_counts["aggregate_comments"]!=sum(provider_counts[k] for k in components): raise Bad("provider counts")
|
||||
matches=[]
|
||||
for c in comments:
|
||||
if not isinstance(c,dict): raise Bad("invalid comment")
|
||||
if corr in str(c.get("body","")): matches.append(c)
|
||||
if len(matches)!=1: raise Bad("correlation cardinality")
|
||||
c=matches[0]; cb=c.get("body")
|
||||
if cb != body: raise Bad("body mismatch")
|
||||
author=c.get("user",c.get("author",{})); cid=author.get("id") if isinstance(author,dict) else None; clog=author.get("login") if isinstance(author,dict) else None
|
||||
if cid!=aid or clog!=login: raise Bad("author mismatch")
|
||||
if isinstance(c.get("id"),bool) or not isinstance(c.get("id"),int) or c["id"]<1 or not c.get("html_url",c.get("url")) or not c.get("created_at") or not c.get("updated_at"): raise Bad("comment metadata")
|
||||
baseline=a.baseline_count
|
||||
if baseline<0 or (a.action=="create" and len(comments)!=baseline+1) or (a.action=="reuse" and len(comments)!=baseline): raise Bad("cardinality")
|
||||
receipt={"schema":"gitea-rework-handoff/v1","status":"verified","target":{"owner":owner,"repository":repository,"issue":int(issue),"pr":int(pr_number),"review":int(review),"object_kind":"pull_request","endpoint_binding_verified":True},"comment_id":c["id"],"comment_url":c.get("html_url",c.get("url")),"comment_body":cb,"marker":marker,"created_at":c["created_at"],"updated_at":c["updated_at"],"final_head":head,"result_sha256":rh,"correlation":corr,"author":{"id":aid,"login":login,"role":"gitea-impl-agent","authenticated":True,"stable_account_verified":True},"pr_metadata_binding_verified":True,"pr_metadata_unchanged":True,"source_issue_unchanged":True,"review_collection_unchanged":True,"provider_counts":provider_counts,"readback_verified":True,"action":a.action,"cardinality":len(comments),"page_count":comment_readback["page_count"]}
|
||||
atomic(a.output,(json.dumps(receipt,sort_keys=True,separators=(",",":"))+"\n").encode())
|
||||
|
||||
def main():
|
||||
p=argparse.ArgumentParser(); sp=p.add_subparsers(dest="cmd",required=True)
|
||||
r=sp.add_parser("render"); r.add_argument("--input",required=True); r.add_argument("--comment",required=True); r.add_argument("--expected",required=True)
|
||||
v=sp.add_parser("verify"); v.add_argument("--expected",required=True); v.add_argument("--pr-readback",required=True); v.add_argument("--comments-readback",required=True); v.add_argument("--identity-readback",required=True); v.add_argument("--action",choices=("create","reuse"),required=True); v.add_argument("--baseline-count",type=int,required=True); v.add_argument("--output",required=True)
|
||||
args=p.parse_args()
|
||||
try: do_render(args) if args.cmd=="render" else do_verify(args)
|
||||
except (Bad, OSError, KeyError, TypeError, UnicodeError) as e: print(f"error: {e}",file=sys.stderr); return 2
|
||||
return 0
|
||||
if __name__=="__main__": raise SystemExit(main())
|
||||
@@ -0,0 +1,70 @@
|
||||
import hashlib, json, subprocess, sys, tempfile, unittest
|
||||
from pathlib import Path
|
||||
|
||||
SCRIPT = Path(__file__).parents[2] / "agents/impl/skills/gitea-issue-rework/scripts/rework_comment.py"
|
||||
|
||||
class ReworkCommentTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.t = tempfile.TemporaryDirectory(); self.d = Path(self.t.name)
|
||||
self.data = {"owner":"acme","repository":"widget","issue":7,"pr":12,"review":99,
|
||||
"reviewed_head":"a"*40,"final_head":"b"*40,"contract_sha256":"c"*64,
|
||||
"result":"partial","findings":[{"id":"F-1","status":"resolved","classification":"valid_fix","changes":"src/main.py","verification":"pytest -q (0), pass","reason":"bounded fix"}],
|
||||
"local_checks":"pytest -q (0), pass","runtime_browser":"해당 없음·미검증","commit_status":"success","actions_runs":"없음"}
|
||||
self.write("input.json", self.data)
|
||||
self.invoke("render", "--input", self.p("input.json"), "--comment", self.p("comment.md"), "--expected", self.p("expected.json"))
|
||||
self.expected = json.loads(self.d.joinpath("expected.json").read_text())
|
||||
|
||||
def tearDown(self): self.t.cleanup()
|
||||
def p(self,n): return str(self.d/n)
|
||||
def write(self,n,x): self.d.joinpath(n).write_text(json.dumps(x,ensure_ascii=False) if not isinstance(x,str) else x)
|
||||
def invoke(self,*args): return subprocess.run([sys.executable,str(SCRIPT),*args],text=True,capture_output=True)
|
||||
def read(self,n): return json.loads(self.d.joinpath(n).read_text())
|
||||
def readback(self, comments=None, head=None):
|
||||
self.write("pr.json", {"object_kind":"pull_request","owner":"acme","repository":"widget","number":12,"state":"open","head":{"sha":head or self.data["final_head"]},"html_url":"https://gitea/acme/widget/pulls/12","metadata_binding_verified":True})
|
||||
self.write("identity.json", {"id":42,"login":"bot","role":"gitea-impl-agent","authenticated":True,"stable_account_verified":True})
|
||||
items=comments if comments is not None else [{"id":501,"html_url":"https://gitea/x/501","created_at":"2026-01-01T00:00:00Z","updated_at":"2026-01-01T00:00:00Z","body":self.expected["body"],"user":{"id":42,"login":"bot"}}]
|
||||
self.write("comments.json", {"object_kind":"pull_request_top_level_comments","owner":"acme","repository":"widget","pr":12,"complete":True,"endpoint_binding_verified":True,"source_issue_unchanged":True,"review_collection_unchanged":True,"pr_metadata_unchanged":True,"page_count":1,"total_count":len(items),"provider_counts":{"top_level":len(items),"reviews":2,"inline":3,"aggregate_comments":len(items),"aggregate_components":["top_level"]},"comments":items})
|
||||
def verify(self, action="create", baseline=0):
|
||||
return self.invoke("verify","--expected",self.p("expected.json"),"--pr-readback",self.p("pr.json"),"--comments-readback",self.p("comments.json"),"--identity-readback",self.p("identity.json"),"--action",action,"--baseline-count",str(baseline),"--output",self.p("receipt.json"))
|
||||
|
||||
def test_render_digest_and_correlation(self):
|
||||
self.assertEqual(hashlib.sha256(self.expected["marker_free_body"].encode()).hexdigest(), self.expected["result_sha256"])
|
||||
parts=["acme","widget","7","12","99","a"*40,"b"*40,"c"*64]
|
||||
self.assertEqual(hashlib.sha256("\0".join(parts).encode()).hexdigest(), self.expected["correlation"])
|
||||
self.assertTrue(self.d.joinpath("comment.md").read_text().endswith("-->\n"))
|
||||
|
||||
def test_no_comment_and_marker_body_mismatch(self):
|
||||
self.readback([]); self.assertNotEqual(self.verify().returncode,0)
|
||||
self.readback([{"id":1,"body":self.expected["body"].replace("correlation=","correlation=x"),"user":{"id":42,"login":"bot"}}]); self.assertNotEqual(self.verify().returncode,0)
|
||||
|
||||
def test_wrong_head_and_author(self):
|
||||
self.readback(head="d"*40); self.assertNotEqual(self.verify().returncode,0)
|
||||
self.readback(); c=self.read("comments.json"); c["comments"][0]["user"]={"id":43,"login":"bot"}; self.write("comments.json",c); self.assertNotEqual(self.verify().returncode,0)
|
||||
|
||||
def test_digest_and_correlation_tampering_duplicate(self):
|
||||
self.expected["result_sha256"]="0"*64; self.write("expected.json",self.expected); self.readback(); self.assertNotEqual(self.verify().returncode,0)
|
||||
self.setUp_render_again(); self.expected["correlation"]="0"*64; self.write("expected.json",self.expected); self.readback(); self.assertNotEqual(self.verify().returncode,0)
|
||||
self.setUp_render_again(); self.readback(); c=self.read("comments.json"); c["comments"].append(dict(c["comments"][0],id=502)); c["total_count"]=2; self.write("comments.json",c); self.assertNotEqual(self.verify("reuse",1).returncode,0)
|
||||
|
||||
def setUp_render_again(self):
|
||||
self.invoke("render","--input",self.p("input.json"),"--comment",self.p("comment.md"),"--expected",self.p("expected.json")); self.expected=self.read("expected.json")
|
||||
def test_create_reuse_cardinality(self):
|
||||
self.readback(); self.assertEqual(self.verify("create",0).returncode,0); self.assertEqual(self.read("receipt.json")["status"],"verified")
|
||||
original=self.read("comments.json")["comments"][0]
|
||||
self.readback([original,{"id":2,"html_url":"u","created_at":"t","updated_at":"t","body":"other","user":{"id":42,"login":"bot"}}]); self.assertEqual(self.verify("reuse",2).returncode,0)
|
||||
def test_exact_success_and_receipt_preserved(self):
|
||||
self.readback(); self.assertEqual(self.verify().returncode,0)
|
||||
old=self.d.joinpath("receipt.json").read_bytes(); self.readback(head="e"*40); self.assertNotEqual(self.verify().returncode,0); self.assertEqual(old,self.d.joinpath("receipt.json").read_bytes())
|
||||
|
||||
def test_unresolved_delivery_and_incomplete_or_wrong_scope_readback_fail(self):
|
||||
bad=dict(self.data); bad["result"]="rework_delivered"; bad["findings"]=[dict(self.data["findings"][0],status="unresolved")]
|
||||
self.write("bad-input.json",bad)
|
||||
self.assertNotEqual(self.invoke("render","--input",self.p("bad-input.json"),"--comment",self.p("bad.md"),"--expected",self.p("bad.json")).returncode,0)
|
||||
self.readback(); comments=self.read("comments.json"); comments["complete"]=False; self.write("comments.json",comments); self.assertNotEqual(self.verify().returncode,0)
|
||||
self.readback(); pr=self.read("pr.json"); pr["repository"]="other"; self.write("pr.json",pr); self.assertNotEqual(self.verify().returncode,0)
|
||||
|
||||
def test_impl_role_and_provider_aggregate_binding_are_required(self):
|
||||
self.readback(); identity=self.read("identity.json"); identity["role"]="other"; self.write("identity.json",identity); self.assertNotEqual(self.verify().returncode,0)
|
||||
self.readback(); comments=self.read("comments.json"); comments["provider_counts"]["aggregate_comments"]+=1; self.write("comments.json",comments); self.assertNotEqual(self.verify().returncode,0)
|
||||
|
||||
if __name__ == "__main__": unittest.main()
|
||||
Reference in New Issue
Block a user