ci: publish tested module release assets

ci/test_release_assets.py

@@ -0,0 +1,205 @@
+#!/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()