claude
13e396a795
`status_phrase` renders a status it does not know verbatim, deliberately:
`argparse`'s `choices=` would exit 2 on an unexpected value and the log --
the entire reason this script exists -- would never be posted.
But the verbatim value lands in a code span inside a **bold** header, so a
backtick in it closes the span early and the rest renders as markdown:
--status 'x` **loud** `y'
-> **`tests` finished with status `x` **loud** `y`**
Nothing hostile is expected: the value comes from `${{ job.status }}` or a
hand-written flag, both written by whoever wrote the workflow. It is worth
closing anyway because this repo is public and four others consume the
script as a composite action, so a branch name or a matrix value could
reach this argument later without anyone revisiting this function.
Backticks are removed rather than escaped -- there is no escape for a
backtick inside a code span, only a wider fence, and the status is a short
word rather than something whose exact bytes matter.
Found in the cold re-read of #10, not by the suite, so the test that
covers it was proved to fail without the fix.
Co-authored-by: bit <bit@das-labor.org>
332 lines
13 KiB
Python
Executable File
332 lines
13 KiB
Python
Executable File
#!/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_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)
|