#!/usr/bin/env python3 """Tests for `report_job_log.py`. Standard library only, like everything else here, and offline: the Gitea it posts to is a `http.server` on localhost that keeps the comments it is sent, so a test can *re-read* what was written instead of trusting an exit code of 0. Run them with any python3: python3 test_report_job_log.py The point of most of them is the comment **header**. `report_job_log.py` used to hardcode the word "failed", so a step guarded by `if: always()` posted a failure report for a job that had passed (weblib-ci#8, found in weblib-viewer#10). The header is the one thing a reader sees before the log, so it is the one thing worth asserting on. """ import contextlib import io import json import os import subprocess import sys import threading import unittest from http.server import BaseHTTPRequestHandler, HTTPServer import report_job_log HERE = os.path.dirname(os.path.abspath(__file__)) SCRIPT = os.path.join(HERE, "report_job_log.py") # Environment variables the script reads. Every test clears all of them, so a # stray one in the ambient shell (this suite is meant to be runnable on a # runner, where several of these are set) cannot change a result. ACTIONS_ENV = [ "GITHUB_REPOSITORY", "GITHUB_EVENT_PATH", "GITHUB_SERVER_URL", "GITHUB_SHA", "GITHUB_JOB", "GITHUB_HEAD_REF", "GITHUB_REF_NAME", "GITEA_HOST", "GITEA_TOKEN", "GITHUB_TOKEN", "JOB_STATUS", ] class FakeGitea(BaseHTTPRequestHandler): """Just enough of the API: list open PRs, resolve a commit's merged PR, accept a comment, and accept a commit status.""" comments = [] # class-level; reset per test statuses = [] pulls = [] # sha -> pull dict, mirroring GET .../commits/{sha}/pull. A sha absent # from this map 404s, the same shape Gitea returns for a commit that was # never part of any PR. pull_for_sha = {} def _send(self, code, payload): raw = json.dumps(payload).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def do_GET(self): if "/commits/" in self.path and self.path.endswith("/pull"): sha = self.path.split("/commits/", 1)[1].rsplit("/pull", 1)[0] pull = type(self).pull_for_sha.get(sha) if pull is None: return self._send( 404, {"message": "pull request does not exist " "[id: 0, issue_id: 0]"}) return self._send(200, pull) if "/pulls" in self.path: return self._send(200, type(self).pulls) self._send(404, {"message": "no"}) def do_POST(self): length = int(self.headers.get("Content-Length") or 0) payload = json.loads(self.rfile.read(length) or b"{}") if "/statuses/" in self.path: type(self).statuses.append({ "path": self.path, "auth": self.headers.get("Authorization"), "context": payload.get("context"), "state": payload.get("state"), "description": payload.get("description", ""), "target_url": payload.get("target_url"), }) return self._send(201, {"id": 1, "state": payload.get("state")}) type(self).comments.append({ "path": self.path, "auth": self.headers.get("Authorization"), "body": payload.get("body", ""), }) self._send(201, {"html_url": "http://example.invalid/c/1"}) def log_message(self, *a): # keep the test output readable pass class ServerTestCase(unittest.TestCase): """A test that runs the whole entry point against the fake forge.""" def setUp(self): FakeGitea.comments = [] FakeGitea.statuses = [] FakeGitea.pulls = [] FakeGitea.pull_for_sha = {} self.server = HTTPServer(("127.0.0.1", 0), FakeGitea) self.host = "http://127.0.0.1:%d" % self.server.server_address[1] # `shutdown()` only takes effect on the next poll, so the default # 0.5s interval charged this suite half a second per test -- measured # 4.6s for nine tests, 0.6s after. self.thread = threading.Thread( target=self.server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True) self.thread.start() self.addCleanup(self.server.server_close) self.addCleanup(self.server.shutdown) self.log = os.path.join(self.mkdtemp(), "build.log") with open(self.log, "w") as fh: fh.write("nix-build \x1b[31msomething\x1b[0m\nDISTINCTIVE-LINE\n") for name in ACTIONS_ENV: os.environ.pop(name, None) def mkdtemp(self): import tempfile d = tempfile.mkdtemp() self.addCleanup(lambda: __import__("shutil").rmtree(d, ignore_errors=True)) return d def posted(self): """What the server actually stored, read back from the server. An exit status of 0 is not evidence that a comment was written: the script deliberately swallows HTTP errors so that a failure to report does not mask the failure being reported, and returns 0 either way. """ return FakeGitea.comments def posted_statuses(self): """Same idea as `posted()`, for the commit-status fallback.""" return FakeGitea.statuses def run_main(self, *argv): """The entry point in-process, so a traceback is readable.""" out = io.StringIO() with contextlib.redirect_stdout(out): code = report_job_log.main([ self.log, "--repo", "weblib/weblib-ci", "--host", self.host, "--token", "t0ken", "--pr", "8", "--job", "tests", *argv]) self.assertEqual(code, 0) self.assertIn("reported to weblib/weblib-ci#8", out.getvalue()) return code def run_cli(self, *argv, env=None): """The entry point as a subprocess: argv parsing and shebang included.""" full = dict(os.environ) full.update(env or {}) proc = subprocess.run( [sys.executable, SCRIPT, self.log, "--repo", "weblib/weblib-ci", "--host", self.host, "--token", "t0ken", "--pr", "8", "--job", "tests", *argv], capture_output=True, text=True, env=full, cwd=HERE) return proc class TestHeader(ServerTestCase): def test_default_is_still_failed(self): """Backward compatibility: the four repos pass only the log path.""" self.run_main() body = self.posted()[0]["body"] self.assertIn("**`tests` failed**", body) self.assertNotIn("passed", body) def test_status_passed(self): self.run_main("--status", "passed") body = self.posted()[0]["body"] self.assertIn("**`tests` passed**", body) self.assertNotIn("failed", body) def test_job_status_expression_success(self): """`${{ job.status }}` yields `success`, not `passed`.""" self.run_main("--status", "success") self.assertIn("**`tests` passed**", self.posted()[0]["body"]) def test_job_status_expression_failure(self): self.run_main("--status", "failure") self.assertIn("**`tests` failed**", self.posted()[0]["body"]) def test_cancelled(self): self.run_main("--status", "cancelled") self.assertIn("**`tests` was cancelled**", self.posted()[0]["body"]) def test_case_and_whitespace_are_not_a_failure_report(self): self.run_main("--status", " Success \n") self.assertIn("**`tests` passed**", self.posted()[0]["body"]) def test_unknown_status_still_reports(self): """A status the table has not heard of must not cost us the log. `argparse(choices=...)` would exit 2 here, and the log this script exists to surface would never be posted. """ self.run_main("--status", "neutral") body = self.posted()[0]["body"] self.assertIn("**`tests` finished with status `neutral`**", body) self.assertIn("DISTINCTIVE-LINE", body) def test_empty_status_falls_back_to_the_default(self): self.run_main("--status", "") self.assertIn("**`tests` failed**", self.posted()[0]["body"]) def test_sha_and_log_survive_a_non_default_status(self): os.environ["GITHUB_SHA"] = "0123456789abcdef" self.run_main("--status", "passed") body = self.posted()[0]["body"] self.assertIn("**`tests` passed** on `01234567`.", body) self.assertIn("DISTINCTIVE-LINE", body) self.assertNotIn("\x1b[31m", body) # ANSI still stripped class TestEnvironment(ServerTestCase): def test_job_status_env_is_honoured(self): proc = self.run_cli(env={"JOB_STATUS": "success"}) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("**`tests` passed**", self.posted()[0]["body"]) def test_flag_beats_the_environment(self): proc = self.run_cli("--status", "failed", env={"JOB_STATUS": "success"}) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("**`tests` failed**", self.posted()[0]["body"]) def test_unset_environment_is_still_failed(self): proc = self.run_cli() self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("**`tests` failed**", self.posted()[0]["body"]) class TestMissingLog(ServerTestCase): def test_missing_log_when_failed(self): self.log = os.path.join(self.mkdtemp(), "absent.log") self.run_main() self.assertIn("it failed before the build started", self.posted()[0]["body"]) def test_missing_log_when_passed_does_not_claim_a_failure(self): self.log = os.path.join(self.mkdtemp(), "absent.log") self.run_main("--status", "passed") body = self.posted()[0]["body"] self.assertIn("nothing was captured", body) self.assertNotIn("failed", body) class TestEndToEnd(ServerTestCase): """The CLI, driven the way a workflow drives it: environment, no `--pr`.""" def _event_file(self, number): path = os.path.join(self.mkdtemp(), "event.json") with open(path, "w") as fh: json.dump({"pull_request": {"number": number}}, fh) return path def _run(self, *argv, env=None): full = dict(os.environ) full.pop("GITEA_HOST", None) full.update({ "GITHUB_REPOSITORY": "weblib/weblib-ci", "GITHUB_SERVER_URL": self.host, "GITEA_TOKEN": "t0ken", "GITHUB_JOB": "probe", "GITHUB_SHA": "abcdef0123456789", "GITHUB_EVENT_PATH": self._event_file(42), }) full.update(env or {}) return subprocess.run([sys.executable, SCRIPT, self.log, *argv], capture_output=True, text=True, env=full, cwd=HERE) def test_always_guarded_probe_that_passed(self): proc = self._run("--status", "success") self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("reported to weblib/weblib-ci#42", proc.stdout) # The write is only real if the resource says so. posted = self.posted() self.assertEqual(len(posted), 1, posted) self.assertEqual(posted[0]["path"], "/api/v1/repos/weblib/weblib-ci/issues/42/comments") self.assertEqual(posted[0]["auth"], "token t0ken") body = posted[0]["body"] self.assertIn("**`probe` passed** on `abcdef01`.", body) self.assertNotIn("failed", body) self.assertIn("DISTINCTIVE-LINE", body) def test_failure_guarded_job_is_unchanged(self): proc = self._run() self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("**`probe` failed** on `abcdef01`.", self.posted()[0]["body"]) def test_branch_lookup_path_still_works(self): """No event payload: the PR is found by head branch, and reported.""" FakeGitea.pulls = [{"number": 7, "head": {"ref": "fix/x"}}] proc = self._run("--status", "passed", env={"GITHUB_EVENT_PATH": "", "GITHUB_HEAD_REF": "fix/x"}) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("reported to weblib/weblib-ci#7", proc.stdout) self.assertIn("**`probe` passed**", self.posted()[0]["body"]) def test_help_still_renders(self): """argparse `%`-expands `help=`; a stray `%` there breaks `--help`.""" proc = subprocess.run([sys.executable, SCRIPT, "--help"], capture_output=True, text=True, cwd=HERE) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("--status", proc.stdout) # A commit that landed by merging a now-closed PR into development/main -- # weblib-ci#14's actual failure mode. A real 40-char sha, because # pull_request_for_sha silently treats a truncated one the same as "no PR". PUSH_SHA = "c9d6db67d9c20f8ab133f4fb5fea359e9c474274" class TestPushWithNoOpenPR(ServerTestCase): """A `push` event: no event payload, no GITHUB_HEAD_REF, and the pushed branch (GITHUB_REF_NAME) is the *target* of a merge, not a PR head -- the exact shape of weblib-ci#14 (weblib-viewer run 1769, PR#27 merged into `development`).""" def _run(self, *argv, env=None): full = dict(os.environ) full.pop("GITEA_HOST", None) full.update({ "GITHUB_REPOSITORY": "weblib/weblib-ci", "GITHUB_SERVER_URL": self.host, "GITEA_TOKEN": "t0ken", "GITHUB_JOB": "publish", "GITHUB_SHA": PUSH_SHA, "GITHUB_REF_NAME": "development", }) full.update(env or {}) return subprocess.run([sys.executable, SCRIPT, self.log, *argv], capture_output=True, text=True, env=full, cwd=HERE) def test_falls_back_to_the_pr_the_push_merged(self): """No open PR has `development` as its head -- it never does -- but the merge commit itself resolves to the PR that produced it, even though that PR is now closed.""" FakeGitea.pulls = [] FakeGitea.pull_for_sha = {PUSH_SHA: {"number": 27, "state": "closed"}} proc = self._run() self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("reported to weblib/weblib-ci#27", proc.stdout) posted = self.posted() self.assertEqual(len(posted), 1, posted) self.assertEqual(posted[0]["path"], "/api/v1/repos/weblib/weblib-ci/issues/27/comments") self.assertIn("DISTINCTIVE-LINE", posted[0]["body"]) self.assertEqual(self.posted_statuses(), [], "a resolvable PR must not also get a status") def test_branch_lookup_still_wins_when_it_applies(self): """Ordering: an *open* PR whose head matches the branch is found first, so the (potentially stale) sha lookup is never even tried.""" FakeGitea.pulls = [{"number": 5, "head": {"ref": "development"}}] FakeGitea.pull_for_sha = {PUSH_SHA: {"number": 999}} proc = self._run() self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("reported to weblib/weblib-ci#5", proc.stdout) self.assertNotIn("999", proc.stdout) def test_no_pr_anywhere_reports_a_commit_status(self): """Not even the sha lookup finds anything -- a commit that was never part of any PR. This must still report *somewhere*, not silently exit 0 the way #14 did.""" FakeGitea.pulls = [] FakeGitea.pull_for_sha = {} proc = self._run() self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn(f"reported to weblib/weblib-ci@{PUSH_SHA[:8]}", proc.stdout) self.assertEqual(self.posted(), [], "no PR exists to comment on") statuses = self.posted_statuses() self.assertEqual(len(statuses), 1, statuses) s = statuses[0] self.assertEqual(s["path"], f"/api/v1/repos/weblib/weblib-ci/statuses/{PUSH_SHA}") self.assertEqual(s["auth"], "token t0ken") self.assertEqual(s["state"], "failure") self.assertIn("publish", s["description"]) self.assertIn("DISTINCTIVE-LINE", s["description"]) self.assertIn(PUSH_SHA, s["target_url"]) def test_commit_status_state_follows_the_reported_outcome(self): FakeGitea.pulls = [] FakeGitea.pull_for_sha = {} for status, want_state in [("passed", "success"), ("cancelled", "warning"), ("skipped", "skipped"), ("neutral", "error")]: with self.subTest(status=status): FakeGitea.statuses = [] proc = self._run("--status", status) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertEqual(self.posted_statuses()[0]["state"], want_state) def test_no_sha_and_no_pr_does_not_crash_or_post_anything(self): """The one case with nowhere at all to report: still exits 0 and says so, rather than raising or posting a status with an empty sha in the URL.""" FakeGitea.pulls = [] FakeGitea.pull_for_sha = {} proc = self._run(env={"GITHUB_SHA": "", "GITHUB_REF_NAME": "development"}) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("nothing to report", proc.stdout) self.assertEqual(self.posted(), []) self.assertEqual(self.posted_statuses(), []) def test_existing_pr_event_path_is_unaffected(self): """The ordinary pull_request-event path -- unrelated to any of the push-triggered fallback -- still goes straight to the comment, and never touches the sha or status endpoints.""" path = os.path.join(self.mkdtemp(), "event.json") with open(path, "w") as fh: json.dump({"pull_request": {"number": 11}}, fh) FakeGitea.pull_for_sha = {PUSH_SHA: {"number": 999}} # must be ignored proc = self._run(env={"GITHUB_EVENT_PATH": path}) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("reported to weblib/weblib-ci#11", proc.stdout) self.assertEqual(self.posted_statuses(), []) class TestStatusPhrase(unittest.TestCase): """The unit under the header, without a server.""" def test_table(self): for status, want in [ ("failed", "failed"), ("failure", "failed"), ("passed", "passed"), ("success", "passed"), ("succeeded", "passed"), ("cancelled", "was cancelled"), ("canceled", "was cancelled"), ("skipped", "was skipped"), ("SUCCESS", "passed"), (" failure ", "failed"), (None, "failed"), ("", "failed"), ]: with self.subTest(status=status): self.assertEqual(report_job_log.status_phrase(status), want) def test_unknown_is_verbatim(self): self.assertEqual(report_job_log.status_phrase("weird"), "finished with status `weird`") def test_a_backtick_in_an_unknown_status_cannot_escape_the_code_span(self): """The verbatim value sits in a code span inside a **bold** header, so a backtick in it would close the span and let the rest render as markdown. `${{ job.status }}` is workflow-author-controlled rather than hostile, but this script is public and shared by four repos.""" phrase = report_job_log.status_phrase("x` **loud** `y") self.assertEqual(phrase, "finished with status `x **loud** y`") self.assertEqual(phrase.count("`"), 2) def test_default_constant_is_failed(self): """Named so that changing it is a deliberate act, not a typo.""" self.assertEqual(report_job_log.DEFAULT_STATUS, "failed") if __name__ == "__main__": unittest.main(verbosity=2)