feat: add PR walkthrough generation workflow (#2933)
This commit is contained in:
Carlos Florêncio
@@ -0,0 +1,106 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Build a GitHub API payload for the owned PR walkthrough callout."""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
|
||||
START_MARKER = "<!-- kandev-pr-walkthrough-start -->"
|
||||
END_MARKER = "<!-- kandev-pr-walkthrough-end -->"
|
||||
SHA_PATTERN = re.compile(r"^[0-9a-f]{40}$")
|
||||
UNCHANGED = 3
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--github-response", type=Path, required=True)
|
||||
parser.add_argument("--output", type=Path, required=True)
|
||||
parser.add_argument("--url", required=True)
|
||||
parser.add_argument("--pr-number", type=int, required=True)
|
||||
parser.add_argument("--head-sha", required=True)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def require_mapping(value: Any, name: str) -> dict[str, Any]:
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"{name} must be a JSON object")
|
||||
return value
|
||||
|
||||
|
||||
def load_pr(path: Path, pr_number: int) -> str:
|
||||
response = require_mapping(json.loads(path.read_text(encoding="utf-8")), "response")
|
||||
if response.get("number") != pr_number:
|
||||
raise ValueError("pull request number does not match the workflow event")
|
||||
|
||||
body = response.get("body")
|
||||
if body is None:
|
||||
return ""
|
||||
if not isinstance(body, str):
|
||||
raise ValueError("pull request body must be a string or null")
|
||||
return body
|
||||
|
||||
|
||||
def validate_url(url: str, pr_number: int, head_sha: str) -> None:
|
||||
if pr_number <= 0:
|
||||
raise ValueError("pull request number must be positive")
|
||||
if not SHA_PATTERN.fullmatch(head_sha):
|
||||
raise ValueError("head SHA must be 40 lowercase hexadecimal characters")
|
||||
expected = f"https://walkthrough.kandev.ai/pr/{pr_number}/{head_sha}.html"
|
||||
if url != expected:
|
||||
raise ValueError(f"expected walkthrough URL {expected}")
|
||||
|
||||
|
||||
def walkthrough_block(url: str) -> str:
|
||||
return "\n".join(
|
||||
(
|
||||
START_MARKER,
|
||||
"> [!TIP]",
|
||||
f"> **PR walkthrough:** [Open the visual walkthrough]({url})",
|
||||
END_MARKER,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def update_body(body: str, url: str) -> str:
|
||||
start_count = body.count(START_MARKER)
|
||||
end_count = body.count(END_MARKER)
|
||||
block = walkthrough_block(url)
|
||||
|
||||
if start_count == 0 and end_count == 0:
|
||||
return block if not body else f"{block}\n\n{body}"
|
||||
if start_count != 1 or end_count != 1:
|
||||
raise ValueError("walkthrough marker block must have one start and one end marker")
|
||||
if not body.startswith(START_MARKER):
|
||||
raise ValueError("walkthrough marker block must remain at the beginning of the body")
|
||||
|
||||
end = body.find(END_MARKER)
|
||||
if end < len(START_MARKER):
|
||||
raise ValueError("walkthrough marker block is malformed")
|
||||
suffix = body[end + len(END_MARKER) :]
|
||||
return f"{block}{suffix}"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
try:
|
||||
validate_url(args.url, args.pr_number, args.head_sha)
|
||||
body = load_pr(args.github_response, args.pr_number)
|
||||
updated = update_body(body, args.url)
|
||||
if updated == body:
|
||||
return UNCHANGED
|
||||
args.output.write_text(
|
||||
json.dumps({"body": updated}, ensure_ascii=False) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
except (OSError, ValueError, json.JSONDecodeError) as error:
|
||||
print(f"error: {error}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,169 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for the trusted PR walkthrough body updater."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
SCRIPT = REPO_ROOT / "scripts" / "pr-walkthrough-pr-body"
|
||||
PR_NUMBER = 2906
|
||||
HEAD_SHA = "a" * 40
|
||||
URL = f"https://walkthrough.kandev.ai/pr/{PR_NUMBER}/{HEAD_SHA}.html"
|
||||
START = "<!-- kandev-pr-walkthrough-start -->"
|
||||
END = "<!-- kandev-pr-walkthrough-end -->"
|
||||
|
||||
|
||||
class PRWalkthroughPRBodyTest(unittest.TestCase):
|
||||
def run_updater(
|
||||
self,
|
||||
body: str | None,
|
||||
*,
|
||||
url: str = URL,
|
||||
response_number: int = PR_NUMBER,
|
||||
response_sha: str = HEAD_SHA,
|
||||
) -> tuple[subprocess.CompletedProcess[str], Path, tempfile.TemporaryDirectory[str]]:
|
||||
temp_dir = tempfile.TemporaryDirectory()
|
||||
root = Path(temp_dir.name)
|
||||
response = root / "pr.json"
|
||||
output = root / "payload.json"
|
||||
response.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"number": response_number,
|
||||
"body": body,
|
||||
"head": {"sha": response_sha},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
result = subprocess.run(
|
||||
[
|
||||
"python3",
|
||||
str(SCRIPT),
|
||||
"--github-response",
|
||||
str(response),
|
||||
"--output",
|
||||
str(output),
|
||||
"--url",
|
||||
url,
|
||||
"--pr-number",
|
||||
str(PR_NUMBER),
|
||||
"--head-sha",
|
||||
HEAD_SHA,
|
||||
],
|
||||
cwd=REPO_ROOT,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
return result, output, temp_dir
|
||||
|
||||
def payload_body(self, output: Path) -> str:
|
||||
return json.loads(output.read_text(encoding="utf-8"))["body"]
|
||||
|
||||
def test_prepends_prominent_callout_without_changing_existing_body(self) -> None:
|
||||
original = "Summary paragraph.\n\n## Important Changes\n\n- Existing content\n"
|
||||
result, output, temp_dir = self.run_updater(original)
|
||||
self.addCleanup(temp_dir.cleanup)
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
updated = self.payload_body(output)
|
||||
self.assertTrue(updated.startswith(f"{START}\n> [!TIP]\n"))
|
||||
self.assertIn(
|
||||
f"> **PR walkthrough:** [Open the visual walkthrough]({URL})",
|
||||
updated,
|
||||
)
|
||||
self.assertEqual(updated.split(f"{END}\n\n", 1)[1], original)
|
||||
|
||||
def test_replaces_the_owned_url_without_duplicating_the_block(self) -> None:
|
||||
old_sha = "b" * 40
|
||||
old_url = f"https://walkthrough.kandev.ai/pr/{PR_NUMBER}/{old_sha}.html"
|
||||
original = (
|
||||
f"{START}\n> [!TIP]\n"
|
||||
f"> **PR walkthrough:** [Open the visual walkthrough]({old_url})\n"
|
||||
f"{END}\n\nContributor content"
|
||||
)
|
||||
result, output, temp_dir = self.run_updater(original)
|
||||
self.addCleanup(temp_dir.cleanup)
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
updated = self.payload_body(output)
|
||||
self.assertEqual(updated.count(START), 1)
|
||||
self.assertEqual(updated.count(END), 1)
|
||||
self.assertIn(URL, updated)
|
||||
self.assertNotIn(old_url, updated)
|
||||
self.assertTrue(updated.endswith("\n\nContributor content"))
|
||||
|
||||
def test_reports_an_unchanged_body_without_writing_a_payload(self) -> None:
|
||||
original = (
|
||||
f"{START}\n> [!TIP]\n"
|
||||
f"> **PR walkthrough:** [Open the visual walkthrough]({URL})\n"
|
||||
f"{END}\n\nContributor content"
|
||||
)
|
||||
result, output, temp_dir = self.run_updater(original)
|
||||
self.addCleanup(temp_dir.cleanup)
|
||||
|
||||
self.assertEqual(result.returncode, 3, result.stderr)
|
||||
self.assertFalse(output.exists())
|
||||
|
||||
def test_accepts_an_empty_github_body(self) -> None:
|
||||
result, output, temp_dir = self.run_updater(None)
|
||||
self.addCleanup(temp_dir.cleanup)
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
updated = self.payload_body(output)
|
||||
self.assertTrue(updated.startswith(START))
|
||||
self.assertTrue(updated.endswith(END))
|
||||
|
||||
def test_rejects_a_url_that_does_not_match_the_event_identity(self) -> None:
|
||||
result, output, temp_dir = self.run_updater(
|
||||
"Contributor content",
|
||||
url=f"https://walkthrough.kandev.ai/pr/999/{HEAD_SHA}.html",
|
||||
)
|
||||
self.addCleanup(temp_dir.cleanup)
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertFalse(output.exists())
|
||||
self.assertIn("expected walkthrough URL", result.stderr)
|
||||
|
||||
def test_rejects_a_github_response_for_another_pull_request(self) -> None:
|
||||
result, output, temp_dir = self.run_updater(
|
||||
"Contributor content",
|
||||
response_number=PR_NUMBER + 1,
|
||||
)
|
||||
self.addCleanup(temp_dir.cleanup)
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertFalse(output.exists())
|
||||
self.assertIn("number does not match", result.stderr)
|
||||
|
||||
def test_allows_the_pr_head_to_advance_while_publication_finishes(self) -> None:
|
||||
result, output, temp_dir = self.run_updater(
|
||||
"Contributor content",
|
||||
response_sha="c" * 40,
|
||||
)
|
||||
self.addCleanup(temp_dir.cleanup)
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertIn(URL, self.payload_body(output))
|
||||
|
||||
def test_rejects_unbalanced_or_non_leading_owned_markers(self) -> None:
|
||||
for body in (
|
||||
f"{START}\nmissing end marker",
|
||||
f"Contributor content\n\n{START}\nowned\n{END}",
|
||||
f"{START}\none\n{END}\n{START}\ntwo\n{END}",
|
||||
):
|
||||
with self.subTest(body=body):
|
||||
result, output, temp_dir = self.run_updater(body)
|
||||
self.addCleanup(temp_dir.cleanup)
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertFalse(output.exists())
|
||||
self.assertIn("marker", result.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Executable
+100
@@ -0,0 +1,100 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Read one bounded UTF-8 regular file from an immutable Git commit."""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
from pathlib import PurePosixPath
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
MAX_BYTES = 512 * 1024
|
||||
SHA_RE = re.compile(r"^[0-9a-f]{40}$")
|
||||
REGULAR_MODES = {"100644", "100755"}
|
||||
|
||||
|
||||
class ReadError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def git(repo: str, *args: str, text: bool = False) -> subprocess.CompletedProcess:
|
||||
env = {"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "LC_ALL": "C"}
|
||||
return subprocess.run(
|
||||
["git", *args],
|
||||
cwd=repo,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=text,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
def validate_path(value: str) -> str:
|
||||
path = PurePosixPath(value)
|
||||
parts = value.split("/")
|
||||
if (
|
||||
not value
|
||||
or value.startswith(("/", ":"))
|
||||
or "\\" in value
|
||||
or "\x00" in value
|
||||
or any(part in {"", ".", ".."} for part in parts)
|
||||
or path.is_absolute()
|
||||
):
|
||||
raise ReadError("invalid repository-relative path")
|
||||
return value
|
||||
|
||||
|
||||
def read_blob(repo: str, head_sha: str, path: str) -> str:
|
||||
if not SHA_RE.fullmatch(head_sha):
|
||||
raise ReadError("head SHA must be 40 lowercase hexadecimal characters")
|
||||
path = validate_path(path)
|
||||
|
||||
commit = git(repo, "cat-file", "-e", f"{head_sha}^{{commit}}")
|
||||
if commit.returncode != 0:
|
||||
raise ReadError("head SHA is not an available commit")
|
||||
|
||||
listing = git(repo, "--literal-pathspecs", "ls-tree", "-z", head_sha, "--", path)
|
||||
records = [record for record in listing.stdout.split(b"\0") if record]
|
||||
if listing.returncode != 0 or len(records) != 1:
|
||||
raise ReadError("path is not a regular file at the requested commit")
|
||||
metadata, separator, matched_path = records[0].partition(b"\t")
|
||||
fields = metadata.split()
|
||||
if not separator or len(fields) != 3 or matched_path != path.encode("utf-8"):
|
||||
raise ReadError("path is not a regular file at the requested commit")
|
||||
mode, object_type, object_id = (field.decode("ascii") for field in fields)
|
||||
if mode not in REGULAR_MODES or object_type != "blob":
|
||||
raise ReadError("path is not a regular file at the requested commit")
|
||||
|
||||
size_result = git(repo, "cat-file", "-s", object_id, text=True)
|
||||
if size_result.returncode != 0:
|
||||
raise ReadError("could not inspect file object")
|
||||
size = int(size_result.stdout.strip())
|
||||
if size > MAX_BYTES:
|
||||
raise ReadError(f"file exceeds {MAX_BYTES} bytes")
|
||||
|
||||
content = git(repo, "cat-file", "blob", object_id)
|
||||
if content.returncode != 0:
|
||||
raise ReadError("could not read file object")
|
||||
try:
|
||||
return content.stdout.decode("utf-8")
|
||||
except UnicodeDecodeError as exc:
|
||||
raise ReadError("file must be UTF-8 text") from exc
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--repo", required=True)
|
||||
parser.add_argument("--head-sha", required=True)
|
||||
parser.add_argument("--path", required=True)
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
sys.stdout.write(read_blob(args.repo, args.head_sha, args.path))
|
||||
except (OSError, ReadError, ValueError) as exc:
|
||||
print(f"PR file read failed: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,112 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for immutable, regular-file reads from an untrusted PR commit."""
|
||||
|
||||
from pathlib import Path
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = ROOT / "scripts" / "pr-walkthrough-read-file"
|
||||
|
||||
|
||||
class PRWalkthroughReadFileTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.repo = Path(self.temp.name) / "repo"
|
||||
self.repo.mkdir()
|
||||
self.git("init")
|
||||
self.git("config", "user.name", "Walkthrough Test")
|
||||
self.git("config", "user.email", "walkthrough@example.invalid")
|
||||
(self.repo / "src").mkdir()
|
||||
(self.repo / "src" / "regular.txt").write_text("trusted commit bytes\n", encoding="utf-8")
|
||||
(self.repo / "executable.sh").write_text("#!/bin/sh\n", encoding="utf-8")
|
||||
os.chmod(self.repo / "executable.sh", 0o755)
|
||||
(self.repo / "binary.dat").write_bytes(b"\xff\xfe")
|
||||
(self.repo / "large.txt").write_text("x" * (512 * 1024 + 1), encoding="utf-8")
|
||||
os.symlink("/etc/passwd", self.repo / "outside-link")
|
||||
self.git("add", ".")
|
||||
self.git("commit", "-m", "test fixture")
|
||||
self.head = self.git("rev-parse", "HEAD").stdout.strip()
|
||||
|
||||
def git(self, *args: str) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
["git", *args],
|
||||
cwd=self.repo,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=True,
|
||||
)
|
||||
|
||||
def read(self, path: str, *, head: str | None = None) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
str(SCRIPT),
|
||||
"--repo",
|
||||
str(self.repo),
|
||||
"--head-sha",
|
||||
head or self.head,
|
||||
"--path",
|
||||
path,
|
||||
],
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
def test_reads_regular_file_from_exact_commit_not_worktree(self) -> None:
|
||||
(self.repo / "src" / "regular.txt").write_text("changed worktree bytes\n", encoding="utf-8")
|
||||
|
||||
result = self.read("src/regular.txt")
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertEqual(result.stdout, "trusted commit bytes\n")
|
||||
|
||||
def test_reads_executable_regular_file(self) -> None:
|
||||
result = self.read("executable.sh")
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertEqual(result.stdout, "#!/bin/sh\n")
|
||||
|
||||
def test_rejects_traversal_absolute_and_pathspec_paths(self) -> None:
|
||||
for path in ("../outside", "/etc/passwd", "src//regular.txt", ":(glob)**"):
|
||||
with self.subTest(path=path):
|
||||
result = self.read(path)
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("invalid repository-relative path", result.stderr)
|
||||
|
||||
def test_rejects_symlink(self) -> None:
|
||||
result = self.read("outside-link")
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("regular file", result.stderr)
|
||||
|
||||
def test_rejects_binary_file(self) -> None:
|
||||
result = self.read("binary.dat")
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("UTF-8 text", result.stderr)
|
||||
|
||||
def test_rejects_oversized_file(self) -> None:
|
||||
result = self.read("large.txt")
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("exceeds 524288 bytes", result.stderr)
|
||||
|
||||
def test_rejects_missing_file_and_invalid_sha(self) -> None:
|
||||
missing = self.read("missing.txt")
|
||||
invalid_sha = self.read("src/regular.txt", head="main")
|
||||
|
||||
self.assertNotEqual(missing.returncode, 0)
|
||||
self.assertIn("not a regular file", missing.stderr)
|
||||
self.assertNotEqual(invalid_sha.returncode, 0)
|
||||
self.assertIn("40 lowercase hexadecimal", invalid_sha.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Executable
+109
@@ -0,0 +1,109 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Render agent-supplied walkthrough JSON with trusted pull-request identity."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
TRUSTED_DIR = Path(".opencode-walkthrough")
|
||||
OUTPUT_DIR = Path("docs/pr-walkthrough")
|
||||
REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
|
||||
|
||||
|
||||
def required_env(name: str) -> str:
|
||||
value = os.environ.get(name, "").strip()
|
||||
if not value:
|
||||
raise ValueError(f"{name} is required")
|
||||
return value
|
||||
|
||||
|
||||
def trusted_identity() -> dict[str, object]:
|
||||
number_text = required_env("PR_NUMBER")
|
||||
if not number_text.isdigit() or int(number_text) < 1:
|
||||
raise ValueError("PR_NUMBER must be a positive integer")
|
||||
number = int(number_text)
|
||||
repo = required_env("PR_REPO")
|
||||
if not REPO_RE.fullmatch(repo):
|
||||
raise ValueError("PR_REPO must be an owner/repository slug")
|
||||
expected_url = f"https://github.com/{repo}/pull/{number}"
|
||||
url = required_env("PR_URL").rstrip("/")
|
||||
if url != expected_url:
|
||||
raise ValueError(f"PR_URL must be {expected_url}")
|
||||
return {
|
||||
"number": number,
|
||||
"title": required_env("PR_TITLE"),
|
||||
"url": url,
|
||||
"files_url": f"{url}/files",
|
||||
"repo": repo,
|
||||
"base": required_env("PR_BASE"),
|
||||
"head": required_env("PR_HEAD"),
|
||||
}
|
||||
|
||||
|
||||
def bind_trusted_metadata(data: object) -> dict[str, object]:
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("walkthrough JSON must be an object")
|
||||
pr = data.get("pr")
|
||||
if not isinstance(pr, dict):
|
||||
pr = {}
|
||||
data["pr"] = pr
|
||||
pr.update(trusted_identity())
|
||||
changes = data.get("changes")
|
||||
if isinstance(changes, list):
|
||||
for change in changes:
|
||||
if isinstance(change, dict):
|
||||
change.pop("file_url", None)
|
||||
return data
|
||||
|
||||
|
||||
def render(data: dict[str, object]) -> tuple[Path, Path]:
|
||||
build_script = TRUSTED_DIR / "build.py"
|
||||
shell = TRUSTED_DIR / "shell.html"
|
||||
if not build_script.is_file() or not shell.is_file():
|
||||
raise ValueError("trusted walkthrough renderer assets are missing")
|
||||
|
||||
number = int(data["pr"]["number"])
|
||||
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
json_path = OUTPUT_DIR / f"pr-{number}.json"
|
||||
html_path = OUTPUT_DIR / f"pr-{number}.html"
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix=".render-", dir=OUTPUT_DIR) as temp_name:
|
||||
temp_dir = Path(temp_name)
|
||||
temp_json = temp_dir / json_path.name
|
||||
temp_html = temp_dir / html_path.name
|
||||
temp_json.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(build_script), str(temp_json), str(temp_html)],
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
message = result.stderr.strip() or result.stdout.strip() or "renderer failed"
|
||||
raise ValueError(message)
|
||||
if not temp_html.is_file() or temp_html.stat().st_size == 0:
|
||||
raise ValueError("renderer did not produce non-empty HTML")
|
||||
temp_html.replace(html_path)
|
||||
temp_json.replace(json_path)
|
||||
return json_path, html_path
|
||||
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
data = json.loads(sys.stdin.read())
|
||||
bound = bind_trusted_metadata(data)
|
||||
json_path, html_path = render(bound)
|
||||
except (json.JSONDecodeError, OSError, ValueError) as exc:
|
||||
print(f"walkthrough render failed: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
print(f"rendered {json_path} and {html_path}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,98 @@
|
||||
#!/usr/bin/env python3
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = ROOT / "scripts" / "pr-walkthrough-render"
|
||||
REFERENCES = ROOT / ".agents" / "skills" / "pr-walkthrough" / "references"
|
||||
|
||||
|
||||
class PRWalkthroughRenderTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.worktree = Path(self.tmp.name)
|
||||
trusted = self.worktree / ".opencode-walkthrough"
|
||||
trusted.mkdir()
|
||||
shutil.copy2(REFERENCES / "build.py", trusted / "build.py")
|
||||
shutil.copy2(REFERENCES / "shell.html", trusted / "shell.html")
|
||||
self.data = json.loads((REFERENCES / "example.json").read_text(encoding="utf-8"))
|
||||
self.env = {
|
||||
**os.environ,
|
||||
"PR_NUMBER": "42",
|
||||
"PR_TITLE": "Trusted pull request title",
|
||||
"PR_URL": "https://github.com/kdlbs/kandev/pull/42",
|
||||
"PR_REPO": "kdlbs/kandev",
|
||||
"PR_BASE": "main",
|
||||
"PR_HEAD": "feature/walkthrough",
|
||||
}
|
||||
|
||||
def run_render(self, data: object) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
[sys.executable, str(SCRIPT)],
|
||||
cwd=self.worktree,
|
||||
env=self.env,
|
||||
input=json.dumps(data),
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
def test_binds_trusted_pr_identity_and_removes_model_link_overrides(self) -> None:
|
||||
self.data["pr"].update(
|
||||
{
|
||||
"number": 999,
|
||||
"title": "Untrusted title",
|
||||
"url": "javascript:alert(1)",
|
||||
"files_url": "https://attacker.invalid/files",
|
||||
"repo": "attacker/repo; curl attacker.invalid",
|
||||
"base": "wrong-base",
|
||||
"head": "wrong-head",
|
||||
}
|
||||
)
|
||||
self.data["changes"][0]["file_url"] = "javascript:alert(2)"
|
||||
|
||||
result = self.run_render(self.data)
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
json_path = self.worktree / "docs" / "pr-walkthrough" / "pr-42.json"
|
||||
html_path = self.worktree / "docs" / "pr-walkthrough" / "pr-42.html"
|
||||
rendered_data = json.loads(json_path.read_text(encoding="utf-8"))
|
||||
self.assertEqual(
|
||||
rendered_data["pr"],
|
||||
{
|
||||
**self.data["pr"],
|
||||
"number": 42,
|
||||
"title": "Trusted pull request title",
|
||||
"url": "https://github.com/kdlbs/kandev/pull/42",
|
||||
"repo": "kdlbs/kandev",
|
||||
"base": "main",
|
||||
"head": "feature/walkthrough",
|
||||
}
|
||||
| {"files_url": "https://github.com/kdlbs/kandev/pull/42/files"},
|
||||
)
|
||||
self.assertNotIn("file_url", rendered_data["changes"][0])
|
||||
html = html_path.read_text(encoding="utf-8")
|
||||
self.assertIn("gh pr review 42 --repo kdlbs/kandev --approve", html)
|
||||
self.assertNotIn("attacker.invalid", html)
|
||||
self.assertNotIn("javascript:", html)
|
||||
|
||||
def test_renderer_failure_leaves_no_partial_outputs(self) -> None:
|
||||
result = self.run_render({"pr": {}})
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
output = self.worktree / "docs" / "pr-walkthrough"
|
||||
self.assertFalse((output / "pr-42.json").exists())
|
||||
self.assertFalse((output / "pr-42.html").exists())
|
||||
self.assertIn("why.problem is required", result.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user