#!/usr/bin/env python3 """Release selection and Gitea upload tests; no real server or credential is used.""" from __future__ import annotations import contextlib import hashlib import http.server import importlib.util import io import json import os import socketserver import tarfile import tempfile import threading import unittest import urllib.parse from pathlib import Path from unittest import mock HERE = Path(__file__).resolve().parent SPEC = importlib.util.spec_from_file_location("release_assets", HERE / "release-assets.py") assert SPEC is not None and SPEC.loader is not None release_assets = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(release_assets) VERSION = "1.2.3" REVISION = "a" * 40 def digest(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def release_bundle(directory: Path) -> list[Path]: package = directory / f"signage-{VERSION}-linux-x64-glibc.tar.gz" metadata = json.dumps({ "module": {"id": "signage", "version": VERSION}, "provenance": {"revision": REVISION}, }).encode() with tarfile.open(package, "w:gz") as archive: member = tarfile.TarInfo("signage/ekklesia.package.json") member.size = len(metadata) archive.addfile(member, io.BytesIO(metadata)) checksum = directory / f"{package.name}.sha256" checksum.write_text(f"{release_assets.sha256(package)} {package.name}\n", encoding="utf-8") source = directory / f"signage-{VERSION}-linux-x64-glibc-corresponding-source.tar.gz" source.write_bytes(b"corresponding source") source_checksum = directory / f"{source.name}.sha256" source_checksum.write_text(f"{release_assets.sha256(source)} {source.name}\n", encoding="utf-8") redistribution = directory / f"signage-{VERSION}-linux-x64-glibc.redistribution.json" redistribution.write_text(json.dumps({ "package": {"file": package.name, "sha256": release_assets.sha256(package)}, "correspondingSource": {"file": source.name, "sha256": release_assets.sha256(source)}, }), encoding="utf-8") notice = directory / f"signage-{VERSION}-linux-x64-glibc.source-notice.txt" notice.write_text("source notice\n", encoding="utf-8") return [package, checksum, source, source_checksum, redistribution, notice] class FakeGitea(http.server.BaseHTTPRequestHandler): assets: dict[str, bytes] = {} release_exists = True tag_revision: str | None = REVISION api_authorizations: list[str | None] = [] public_authorizations: list[str | None] = [] asset_posts = 0 release_posts = 0 release_request: dict[str, object] | None = None def log_message(self, _format: str, *_args: object) -> None: pass def answer(self, status: int, body: object = None) -> None: data = b"" if body is None else json.dumps(body).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def asset_list(self) -> list[dict[str, object]]: server = self.server assert isinstance(server, socketserver.TCPServer) host, port = server.server_address return [ {"id": index, "name": name, "browser_download_url": f"http://{host}:{port}/public/{name}"} for index, name in enumerate(self.assets, 1) ] def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler's API if self.path.startswith("/public/"): self.public_authorizations.append(self.headers.get("Authorization")) name = urllib.parse.unquote(self.path.removeprefix("/public/")) data = self.assets[name] self.send_response(200) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return self.api_authorizations.append(self.headers.get("Authorization")) if "/releases/tags/" in self.path: if self.release_exists: self.answer(200, {"id": 7, "tag_name": f"v{VERSION}"}) else: self.answer(404, {"message": "not found"}) elif "/tags/" in self.path: if self.tag_revision is not None: self.answer(200, {"commit": {"sha": self.tag_revision}}) else: self.answer(404, {"message": "not found"}) elif self.path.endswith("/releases/7/assets"): self.answer(200, self.asset_list()) else: self.answer(404, {"message": "not found"}) def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler's API self.api_authorizations.append(self.headers.get("Authorization")) length = int(self.headers["Content-Length"]) body = self.rfile.read(length) if self.path.endswith("/releases"): request = json.loads(body) self.__class__.release_posts += 1 self.__class__.release_request = request self.__class__.release_exists = True self.__class__.tag_revision = request["target_commitish"] self.answer(201, {"id": 7, "tag_name": request["tag_name"]}) return self.__class__.asset_posts += 1 name = urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query)["name"][0] boundary = self.headers["Content-Type"].split("boundary=", 1)[1].encode() payload = body.split(b"\r\n\r\n", 1)[1].rsplit(b"\r\n--" + boundary + b"--", 1)[0] self.assets[name] = payload self.answer(201, {"name": name}) @contextlib.contextmanager def fake_gitea(initial: dict[str, bytes] | None = None, *, release_exists: bool = True, tag_revision: str | None = REVISION): FakeGitea.assets = dict(initial or {}) FakeGitea.release_exists = release_exists FakeGitea.tag_revision = tag_revision FakeGitea.api_authorizations = [] FakeGitea.public_authorizations = [] FakeGitea.asset_posts = 0 FakeGitea.release_posts = 0 FakeGitea.release_request = None with socketserver.TCPServer(("127.0.0.1", 0), FakeGitea) as server: thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() host, port = server.server_address old = {name: os.environ.get(name) for name in ("GITHUB_SERVER_URL", "GITHUB_REPOSITORY", "GITHUB_TOKEN")} os.environ.update({ "GITHUB_SERVER_URL": f"http://{host}:{port}", "GITHUB_REPOSITORY": "owner/repository", "GITHUB_TOKEN": "test-token", }) try: yield finally: server.shutdown() thread.join() for name, value in old.items(): if value is None: os.environ.pop(name, None) else: os.environ[name] = value class ReleaseAssetsTest(unittest.TestCase): def test_selects_only_the_complete_bound_bundle(self) -> None: with tempfile.TemporaryDirectory() as temporary: directory = Path(temporary) expected = release_bundle(directory) self.assertEqual(release_assets.expected_paths(directory, f"v{VERSION}", REVISION), expected) (directory / "unexpected.txt").write_text("no", encoding="utf-8") with self.assertRaisesRegex(release_assets.Refused, "six required files"): release_assets.expected_paths(directory, f"v{VERSION}", REVISION) def test_refuses_wrong_tag_revision_and_checksum(self) -> None: with tempfile.TemporaryDirectory() as temporary: directory = Path(temporary) paths = release_bundle(directory) with self.assertRaisesRegex(release_assets.Refused, "tag"): release_assets.expected_paths(directory, "v9.9.9", REVISION) with self.assertRaisesRegex(release_assets.Refused, "provenance"): release_assets.expected_paths(directory, f"v{VERSION}", "b" * 40) with mock.patch.dict(os.environ, {"EXPECTED_PACKAGE_SHA256": "0" * 64}): with self.assertRaisesRegex(release_assets.Refused, "package job's bytes"): release_assets.expected_paths(directory, f"v{VERSION}", REVISION) paths[1].write_text(f"{'0' * 64} {paths[0].name}\n", encoding="utf-8") with self.assertRaisesRegex(release_assets.Refused, "does not match"): release_assets.expected_paths(directory, f"v{VERSION}", REVISION) def test_creates_release_uploads_each_byte_and_reads_it_back_anonymously(self) -> None: with tempfile.TemporaryDirectory() as temporary: directory = Path(temporary) expected = release_bundle(directory) with fake_gitea(release_exists=False, tag_revision=None): release_assets.publish(directory, f"v{VERSION}", REVISION) self.assertEqual(FakeGitea.release_posts, 1) self.assertEqual(FakeGitea.release_request, { "tag_name": f"v{VERSION}", "target_commitish": REVISION, "name": f"Ekklesia Signage v{VERSION}", "body": ( "Published by the manual Publish Signage Release workflow after package, " "external integration, and corresponding-source verification." ), "draft": False, "prerelease": True, }) self.assertEqual(FakeGitea.asset_posts, 6) self.assertEqual(FakeGitea.assets, {path.name: path.read_bytes() for path in expected}) self.assertTrue(all(value == "token test-token" for value in FakeGitea.api_authorizations)) self.assertEqual(FakeGitea.public_authorizations, [None] * 6) def test_existing_identical_assets_are_idempotent_but_conflicts_fail(self) -> None: with tempfile.TemporaryDirectory() as temporary: directory = Path(temporary) expected = release_bundle(directory) initial = {path.name: path.read_bytes() for path in expected} with fake_gitea(initial): release_assets.publish(directory, f"v{VERSION}", REVISION) self.assertEqual(FakeGitea.release_posts, 0) self.assertEqual(FakeGitea.asset_posts, 0) conflicting = {expected[1].name: b"different published bytes"} with fake_gitea(conflicting): with self.assertRaisesRegex(release_assets.Refused, "public .* expected"): release_assets.publish(directory, f"v{VERSION}", REVISION) self.assertEqual(FakeGitea.release_posts, 0) self.assertEqual(FakeGitea.asset_posts, 0) def test_refuses_existing_tag_at_another_revision_before_mutation(self) -> None: with tempfile.TemporaryDirectory() as temporary: directory = Path(temporary) release_bundle(directory) with fake_gitea(release_exists=False, tag_revision="b" * 40): with self.assertRaisesRegex(release_assets.Refused, "existing tag"): release_assets.publish(directory, f"v{VERSION}", REVISION) self.assertEqual(FakeGitea.release_posts, 0) self.assertEqual(FakeGitea.asset_posts, 0) if __name__ == "__main__": unittest.main()