#!/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] = {} api_authorizations: list[str | None] = [] public_authorizations: list[str | None] = [] posts = 0 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: self.answer(200, {"id": 7}) elif "/tags/" in self.path: self.answer(200, {"commit": {"sha": REVISION}}) 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")) self.__class__.posts += 1 name = urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query)["name"][0] length = int(self.headers["Content-Length"]) body = self.rfile.read(length) 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): FakeGitea.assets = dict(initial or {}) FakeGitea.api_authorizations = [] FakeGitea.public_authorizations = [] FakeGitea.posts = 0 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_uploads_each_selected_byte_and_reads_it_back_anonymously(self) -> None: with tempfile.TemporaryDirectory() as temporary: directory = Path(temporary) expected = release_bundle(directory) with fake_gitea(): release_assets.upload(directory, f"v{VERSION}", REVISION) self.assertEqual(FakeGitea.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.upload(directory, f"v{VERSION}", REVISION) self.assertEqual(FakeGitea.posts, 0) conflicting = {expected[1].name: b"different published bytes"} with fake_gitea(conflicting): with self.assertRaisesRegex(release_assets.Refused, "public .* expected"): release_assets.upload(directory, f"v{VERSION}", REVISION) self.assertEqual(FakeGitea.posts, 0) if __name__ == "__main__": unittest.main()