scripts/test_micromanager_pr_report.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Unit tests for scripts/micromanager-pr-report.py (stdlib unittest)."""
from __future__ import annotations
import importlib.util
import io
import json
import re
import sys
import tempfile
import threading
import unittest
import urllib.error
from contextlib import redirect_stdout
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parent.parent
def _load_report():
path = Path(__file__).with_name("micromanager-pr-report.py")
spec = importlib.util.spec_from_file_location("micromanager_pr_report", path)
if spec is None or spec.loader is None:
raise ImportError(f"cannot load {path}")
module = importlib.util.module_from_spec(spec)
# dataclasses look their module up while the class is being built.
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
report = _load_report()
CYAN = "\x1b[36;1m"
RESET = "\x1b[0m"
def stamp(lines: list[str]) -> str:
"""A job log as GitHub serves it: BOM, timestamps, CRLF."""
body = "\r\n".join(f"2026-09-23T08:01:35.2132921Z {line}" for line in lines)
return "" + body + "\r\n"
LOG = stamp(
[
"Current runner version: '2.337.0'",
"##[group]Runner Image",
"Image: windows-2025",
"##[endgroup]",
"##[group]Run actions/checkout@v7",
"with:",
" repository: SteelCrab/firecrab",
"##[endgroup]",
"##[group]Getting Git version info",
"git version 2.55.0",
"##[endgroup]",
"##[group]Run cargo test -p firecrab-cli --locked",
f"{CYAN}cargo test -p firecrab-cli --locked \\{RESET}",
f"{CYAN} --no-fail-fast{RESET}",
"shell: /usr/bin/bash -e {0}",
"##[endgroup]",
"test a ... ok",
"\x1b[32mtest result: ok. 1 passed\x1b[0m",
"",
"##[group]Run ./doctor",
f"{CYAN}./doctor{RESET}",
"##[endgroup]",
"check failed",
"##[error]Process completed with exit code 1.",
"Post job cleanup.",
"##[group]Run actions/checkout@v7 post",
"should not be read",
]
)
STEPS = [
{"number": 1, "name": "Set up job", "conclusion": "success"},
{"number": 2, "name": "Run actions/checkout@v7", "conclusion": "success"},
{
"number": 3,
"name": "Tests",
"conclusion": "success",
"started_at": "2026-09-23T08:01:35Z",
"completed_at": "2026-09-23T08:03:20Z",
},
{"number": 4, "name": "Doctor", "conclusion": "failure"},
{"number": 5, "name": "Never reached", "conclusion": "skipped"},
{"number": 6, "name": "Post Run actions/checkout@v7", "conclusion": "success"},
{"number": 7, "name": "Complete job", "conclusion": "success"},
]
class ChangedFilesTests(unittest.TestCase):
def test_platform_files_belong_to_their_platform_only(self) -> None:
files = [
"firecrab-cli/src/micromanager/windows/wsl.rs",
"micromanager-macos/Sources/FirecrabMicroManager/Launcher.swift",
"scripts/ci-qa-windows-e2e.ps1",
]
self.assertEqual(
report.changed_files(report.WINDOWS, files),
["firecrab-cli/src/micromanager/windows/wsl.rs", "scripts/ci-qa-windows-e2e.ps1"],
)
self.assertEqual(
report.changed_files(report.MACOS, files),
["micromanager-macos/Sources/FirecrabMicroManager/Launcher.swift"],
)
def test_shared_micromanager_modules_belong_to_both(self) -> None:
for shared in (
"firecrab-cli/src/micromanager.rs",
"firecrab-cli/src/micromanager/artifact.rs",
"firecrab-cli/src/micromanager/dev.rs",
"scripts/micromanager/dev-macos-guest.sh",
"scripts/test-micromanager-dev.py",
"scripts/ci-qa-guest.sh",
"firecrab-e2e/playwright.config.ts",
):
for platform in report.PLATFORMS:
self.assertEqual(report.changed_files(platform, [shared]), [shared])
def test_the_platform_roots_are_not_shared(self) -> None:
self.assertEqual(
report.changed_files(report.MACOS, ["firecrab-cli/src/micromanager/windows.rs"]),
[],
)
def test_other_files_bring_in_no_platform(self) -> None:
files = ["firecrab-api/src/main.rs", "install-cli.ps1", "README.md"]
for platform in report.PLATFORMS:
self.assertEqual(report.changed_files(platform, files), [])
class LogTests(unittest.TestCase):
def test_a_log_splits_into_run_sections_until_the_cleanup(self) -> None:
sections = report.split_log(LOG)
self.assertEqual(
[section.header for section in sections],
["actions/checkout@v7", "cargo test -p firecrab-cli --locked", "./doctor"],
)
tests = sections[1]
self.assertEqual(
tests.commands, ["cargo test -p firecrab-cli --locked \\", " --no-fail-fast"]
)
self.assertEqual(tests.output, ["test a ... ok", "test result: ok. 1 passed", ""])
self.assertEqual(
sections[2].output,
["check failed", "##[error]Process completed with exit code 1."],
)
def test_steps_pair_with_sections_and_actions_show_no_log(self) -> None:
steps = report.step_logs(STEPS, LOG)
self.assertEqual(
[step.name for step in steps],
["Run actions/checkout@v7", "Tests", "Doctor", "Never reached"],
)
checkout, tests, doctor, skipped = steps
self.assertIsNone(checkout.section)
self.assertEqual(tests.section.header, "cargo test -p firecrab-cli --locked")
self.assertEqual(tests.seconds, 105)
self.assertEqual(doctor.section.header, "./doctor")
self.assertIsNone(skipped.section)
def test_an_unexpected_section_count_shows_steps_without_logs(self) -> None:
steps = report.step_logs(STEPS[:3], LOG)
self.assertTrue(all(step.section is None for step in steps))
self.assertTrue(all(step.section is None for step in report.step_logs(STEPS, None)))
class RenderTests(unittest.TestCase):
def job_report(self, platform=None, log: str = LOG) -> object:
job = {
"html_url": "https://github.com/o/r/actions/runs/1/job/2",
"conclusion": "failure",
"started_at": "2026-09-23T08:00:00Z",
"completed_at": "2026-09-23T08:02:24Z",
}
return report.JobReport(
platform or report.WINDOWS,
["firecrab-cli/src/micromanager/windows/wsl.rs"],
job,
report.step_logs(STEPS, log),
)
def test_the_comment_has_commands_steps_and_the_failed_log(self) -> None:
body = report.render([self.job_report()], "0123456789abcdef", "https://run", (15, 80))
self.assertTrue(body.startswith(report.MARKER))
self.assertIn("Commit `0123456` · [workflow run](https://run)", body)
self.assertIn("| Windows | [microManager (GitHub-hosted Windows build)]", body)
self.assertIn("❌ failure | 2m 24s |", body)
for command in report.WINDOWS.local + report.WINDOWS.e2e:
self.assertIn(command, body)
self.assertIn("| Tests | ✅ | 1m 45s |", body)
self.assertIn("| Never reached | ⏭️ | – |", body)
self.assertIn("<details open><summary>❌ Doctor: log, last 80 lines", body)
self.assertIn("$ cargo test -p firecrab-cli --locked \\\n --no-fail-fast", body)
self.assertIn("##[error]Process completed with exit code 1.", body)
self.assertNotIn("should not be read", body)
def fit_under(self, budget: tuple[int, int]) -> str:
"""`fit` with the limit set to exactly what `budget` renders to."""
limit = len(report.render([self.job_report()], "0" * 40, "https://run", budget))
original = report.COMMENT_LIMIT
report.COMMENT_LIMIT = limit
try:
return report.fit([self.job_report()], "0" * 40, "https://run")
finally:
report.COMMENT_LIMIT = original
def test_passing_logs_shrink_first_when_the_comment_is_too_long(self) -> None:
body = self.fit_under((5, 80))
self.assertIn("✅ Tests: log, last 5 lines", body)
self.assertIn("❌ Doctor: log, last 80 lines", body)
body = self.fit_under((0, 40))
self.assertNotIn("✅ Tests: log", body)
self.assertIn("❌ Doctor: log, last 40 lines", body)
def test_a_job_that_never_ran_says_so(self) -> None:
missing = report.JobReport(report.MACOS, ["micromanager-macos/Package.swift"], None, [])
body = report.render([missing], "0" * 40, "https://run", (15, 80))
self.assertIn("| macOS | microManager (GitHub-hosted macOS build) | ⏳ not run | – |", body)
self.assertIn("The job did not run", body)
def test_log_backticks_cannot_close_the_code_block(self) -> None:
block = report.code_block(["```", "a ```` b"], "text")
self.assertTrue(block.startswith("`````text\n"))
self.assertTrue(block.endswith("\n`````"))
def test_long_log_lines_are_clipped(self) -> None:
section = report.Section("x", [], ["y" * (report.LINE_LIMIT + 50)])
(line,) = report.tail(section, 5)
self.assertEqual(len(line), report.LINE_LIMIT + 2)
self.assertTrue(line.endswith(" …"))
class CiContractTests(unittest.TestCase):
"""The report names ci.yml's jobs and repeats their commands; keep them in step."""
ci = (ROOT / ".github/workflows/ci.yml").read_text(encoding="utf-8")
def job_block(self, name: str) -> str:
lines = self.ci.splitlines()
start = lines.index(f" name: {name}")
end = next(
(i for i in range(start + 1, len(lines)) if re.match(r"^ \S", lines[i])),
len(lines),
)
return " ".join(" ".join(lines[start:end]).split())
def test_local_commands_are_what_the_job_runs(self) -> None:
for platform in report.PLATFORMS:
block = self.job_block(platform.job)
for command in platform.local:
normalized = command.removeprefix(".\\").replace("\\", "/")
self.assertIn(normalized, block, f"{platform.name}: {command}")
def test_the_report_job_waits_for_both_platform_jobs(self) -> None:
block = self.job_block("microManager PR report")
self.assertIn("needs: [micromanager-macos-build, micromanager-windows-build]", block)
self.assertIn("python3 scripts/micromanager-pr-report.py", block)
self.assertIn("pull-requests: write", block)
def test_every_platform_path_exists(self) -> None:
for platform in report.PLATFORMS:
for path in platform.paths:
self.assertTrue((ROOT / path).exists(), path)
for command in platform.e2e:
for script in re.findall(r"scripts[\\/][\w./\\-]+", command):
self.assertTrue((ROOT / script.replace("\\", "/")).exists(), script)
class ApiHandler(BaseHTTPRequestHandler):
"""A stand-in for api.github.com and the blob store logs redirect to."""
seen: list[tuple[str, str, str | None, bytes]] = []
def log_message(self, *args) -> None:
pass
def reply(self, code: int, body: bytes, headers: dict | None = None) -> None:
self.send_response(code)
for key, value in (headers or {}).items():
self.send_header(key, value)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def record(self) -> None:
length = int(self.headers.get("Content-Length") or 0)
body = self.rfile.read(length) if length else b""
self.seen.append((self.command, self.path, self.headers.get("Authorization"), body))
def do_GET(self) -> None:
self.record()
if self.path.startswith("/repos/o/r/pulls/5/files"):
page = int(self.path.rsplit("page=", 1)[1])
count = 100 if page == 1 else 1
files = [{"filename": f"f{page}-{i}"} for i in range(count)]
return self.reply(200, json.dumps(files).encode())
if self.path == "/repos/o/r/actions/jobs/9/logs":
return self.reply(302, b"", {"Location": "/blob/9.txt"})
if self.path == "/blob/9.txt":
return self.reply(200, "log ✓".encode())
self.reply(404, b"{}")
def do_POST(self) -> None:
self.record()
self.reply(201, b"{}")
class GitHubClientTests(unittest.TestCase):
def setUp(self) -> None:
ApiHandler.seen = []
self.server = ThreadingHTTPServer(("127.0.0.1", 0), ApiHandler)
threading.Thread(
target=self.server.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True
).start()
self.addCleanup(self.server.server_close)
self.addCleanup(self.server.shutdown)
api = f"http://127.0.0.1:{self.server.server_address[1]}"
self.github = report.GitHub(api, "o/r", "secret")
def test_lists_follow_pages_until_a_short_one(self) -> None:
files = self.github.pages("/pulls/5/files")
self.assertEqual(len(files), 101)
self.assertEqual(
[path for _, path, _, _ in ApiHandler.seen],
[
"/repos/o/r/pulls/5/files?per_page=100&page=1",
"/repos/o/r/pulls/5/files?per_page=100&page=2",
],
)
def test_a_log_redirect_does_not_carry_the_token(self) -> None:
self.assertEqual(self.github.job_log(9), "log ✓")
(api, blob) = ApiHandler.seen
self.assertEqual(api[2], "Bearer secret")
self.assertEqual((blob[1], blob[2]), ("/blob/9.txt", None))
def test_a_missing_log_is_retried_then_reported(self) -> None:
output = io.StringIO()
with mock.patch.object(report.time, "sleep") as sleep, redirect_stdout(output):
self.assertIsNone(self.github.job_log(8))
self.assertEqual(len(ApiHandler.seen), 3)
self.assertEqual(sleep.call_count, 2)
self.assertIn("::warning::log of job 8: HTTP 404", output.getvalue())
def test_writes_send_json_with_the_token(self) -> None:
self.github.request("POST", "/issues/5/comments", {"body": "hi"})
((method, path, token, body),) = ApiHandler.seen
self.assertEqual(
(method, path, token), ("POST", "/repos/o/r/issues/5/comments", "Bearer secret")
)
self.assertEqual(json.loads(body), {"body": "hi"})
class FakeGitHub:
"""Records writes; serves canned files, jobs, logs, and comments."""
def __init__(self, files=(), jobs=(), logs=None, comments=(), refuse_writes=False):
self.files = [{"filename": name} for name in files]
self.jobs = list(jobs)
self.logs = logs or {}
self.comments = list(comments)
self.refuse_writes = refuse_writes
self.writes: list[tuple[str, str, object]] = []
def pages(self, path: str) -> list:
if path.startswith("/pulls/"):
return self.files
return self.comments
def get(self, path: str):
assert path == "/actions/runs/77/attempts/2/jobs?per_page=100", path
return {"jobs": self.jobs}
def job_log(self, job_id: int) -> str | None:
return self.logs.get(job_id)
def request(self, method: str, path: str, body=None) -> bytes:
if self.refuse_writes:
raise urllib.error.HTTPError(path, 403, "Forbidden", None, None)
self.writes.append((method, path, body))
return b"{}"
def comment(identifier: int, login: str, body: str) -> dict:
return {"id": identifier, "user": {"login": login}, "body": body}
class PublishTests(unittest.TestCase):
def test_a_first_report_creates_the_comment(self) -> None:
github = FakeGitHub(comments=[comment(1, "someone", "lgtm")])
report.publish(github, 5, "body")
self.assertEqual(github.writes, [("POST", "/issues/5/comments", {"body": "body"})])
def test_a_later_push_edits_the_bots_own_comment_only(self) -> None:
quoted = comment(1, "someone", f"quoting {report.MARKER}")
own = comment(2, report.BOT_LOGIN, f"{report.MARKER}\nold")
github = FakeGitHub(comments=[quoted, own])
report.publish(github, 5, "new")
self.assertEqual(github.writes, [("PATCH", "/issues/comments/2", {"body": "new"})])
def test_no_report_removes_a_stale_comment(self) -> None:
github = FakeGitHub(comments=[comment(3, report.BOT_LOGIN, report.MARKER)])
report.publish(github, 5, None)
self.assertEqual(github.writes, [("DELETE", "/issues/comments/3", None)])
github = FakeGitHub()
report.publish(github, 5, None)
self.assertEqual(github.writes, [])
def test_a_read_only_token_warns_instead_of_failing(self) -> None:
output = io.StringIO()
with redirect_stdout(output):
report.publish(FakeGitHub(refuse_writes=True), 5, "body")
self.assertIn(
"::warning::cannot write the pull request comment: HTTP 403", output.getvalue()
)
class RunTests(unittest.TestCase):
def setUp(self) -> None:
self.directory = tempfile.TemporaryDirectory()
self.addCleanup(self.directory.cleanup)
self.root = Path(self.directory.name)
self.summary = self.root / "summary.md"
self.summary.write_text("", encoding="utf-8")
def env(self, event: dict) -> dict:
path = self.root / "event.json"
path.write_text(json.dumps(event), encoding="utf-8")
return {
"GITHUB_EVENT_PATH": str(path),
"GITHUB_REPOSITORY": "SteelCrab/firecrab",
"GITHUB_RUN_ID": "77",
"GITHUB_RUN_ATTEMPT": "2",
"GITHUB_SERVER_URL": "https://github.com",
"GITHUB_STEP_SUMMARY": str(self.summary),
}
def run_report(self, github: FakeGitHub, event: dict) -> str:
output = io.StringIO()
with redirect_stdout(output):
self.assertEqual(report.run(self.env(event), github), 0)
return output.getvalue()
def test_a_windows_change_reports_the_windows_job(self) -> None:
job = {
"id": 9,
"name": report.WINDOWS.job,
"conclusion": "failure",
"html_url": "https://github.com/o/r/actions/runs/77/job/9",
"steps": STEPS,
}
github = FakeGitHub(
files=["firecrab-cli/src/micromanager/windows/doctor.rs"],
jobs=[job, {"id": 8, "name": report.MACOS.job, "conclusion": "success"}],
logs={9: LOG},
)
event = {"pull_request": {"number": 5, "head": {"sha": "abcdef0123"}}}
self.assertIn("reported Windows on #5", self.run_report(github, event))
((method, path, body),) = github.writes
self.assertEqual((method, path), ("POST", "/issues/5/comments"))
self.assertIn("### Windows", body["body"])
self.assertNotIn("### macOS", body["body"])
self.assertIn("https://github.com/SteelCrab/firecrab/actions/runs/77", body["body"])
summary = self.summary.read_text(encoding="utf-8")
self.assertTrue(summary.startswith("## microManager CI"))
self.assertNotIn(report.MARKER, summary)
def test_an_unrelated_change_writes_nothing(self) -> None:
github = FakeGitHub(files=["firecrab-api/src/main.rs"])
event = {"pull_request": {"number": 5, "head": {"sha": "abc"}}}
self.assertIn("no microManager platform files changed", self.run_report(github, event))
self.assertEqual(github.writes, [])
self.assertEqual(self.summary.read_text(encoding="utf-8"), "")
def test_other_events_are_ignored(self) -> None:
self.assertIn("not a pull request", self.run_report(FakeGitHub(), {"ref": "main"}))
if __name__ == "__main__":
unittest.main()