#!/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, and accept a comment.""" comments = [] # class-level; reset per test pulls = [] 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 "/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"{}") 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.pulls = [] 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 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) 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_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)