bass-foundry/scripts/tests/test_release.py

import contextlib
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import struct
import sys
import tempfile
import unittest
import zipfile

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from plugin_catalog import archive_name, tag_version
import release

spec = importlib.util.spec_from_file_location("package_windows", Path(__file__).resolve().parents[1] / "package-windows.py")
packaging = importlib.util.module_from_spec(spec)
spec.loader.exec_module(packaging)

class FakeGitea:
    def __init__(self):
        self.release = None
        self.bytes = {}
        self.calls = []
        self.tag_exists = True
        self.fail_upload = False

    def request(self, method, path, data=None, missing=False):
        self.calls.append((method, path, data))
        if path.startswith("/tags/"):
            return {"name": path.split("/")[-1]} if self.tag_exists else None
        if method == "GET":
            return self.release
        if method == "POST" and path == "/releases":
            self.release = dict(data, id=7, assets=[])
            return self.release
        if method == "PATCH":
            self.release.update(data)
            return self.release
        raise AssertionError((method, path))

    def upload(self, release_id, path):
        if self.fail_upload and path.name.endswith(".sha256"):
            raise RuntimeError("simulated interruption")
        self.calls.append(("UPLOAD", path.name, None))
        url = "https://gitea.invalid/download/" + path.name
        self.bytes[url] = path.read_bytes()
        asset = {"name": path.name, "browser_download_url": url}
        self.release["assets"].append(asset)
        return asset

    def download(self, url):
        return self.bytes[url]

class ReleaseTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.folder = self.root / "dist/x86_64-pc-windows-gnu"
        self.folder.mkdir(parents=True)
        catalog = {}
        for plugin, name, version in [("rivet", "Rivet", "0.3.0"), ("patina", "Patina", "0.1.0")]:
            manifest = f"plugins/{plugin}/plugin/Cargo.toml"
            catalog[plugin] = {"name": name, "description": name + " processor", "manifest": manifest,
                               "documentation": f"docs/{plugin}"}
            self.write(manifest, f'[package]\nname="{plugin}"\nversion="{version}"\n')
            self.write(f"docs/{plugin}/README.md", name + " guide")
            self.write(f"docs/{plugin}/CHANGELOG.md", f"# Changelog\n\n## {version}\n\nFirst {name} release.\n")
            self.write(f"dist/x86_64-pc-windows-gnu/docs/{plugin}/README.md", name + " guide")
            self.write(f"dist/x86_64-pc-windows-gnu/docs/{plugin}/CHANGELOG.md", f"# Changelog\n\n## {version}\n\nFirst {name} release.\n")
            binary = bytearray(160)
            binary[:2] = b"MZ"
            struct.pack_into("<I", binary, 0x3C, 64)
            binary[64:68] = b"PE\0\0"
            struct.pack_into("<H", binary, 68, 0x8664)
            binary[120:131] = b"clap_entry\0"
            (self.folder / f"{name}.clap").write_bytes(binary)
        self.write("plugins/catalog.json", json.dumps(catalog))
        for name in ["README.md", "LICENSE", "docs/development.md", "docs/releases.md"]:
            self.write("dist/x86_64-pc-windows-gnu/" + name, name)
        (self.folder / "Legacy.clap").write_bytes(b"unused")

    def write(self, path, value):
        p = self.root / path
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(value)

    def package(self, plugin):
        with contextlib.redirect_stdout(io.StringIO()):
            return packaging.package(plugin=plugin, root=self.root)

    def publish(self, client, plugin="patina"):
        tag = "patina-v0.1.0" if plugin == "patina" else "rivet-v0.3.0"
        with contextlib.redirect_stdout(io.StringIO()):
            release.publish(plugin, tag, client, self.root)

    def test_tag_and_version_are_independent(self):
        self.assertEqual(tag_version("rivet", "rivet-v0.3.0", self.root), "0.3.0")
        self.assertEqual(release.plan("refs/tags/patina-v0.1.0", self.root)["plugin"], "patina")
        for plugin, tag in [("rivet", "patina-v0.1.0"), ("patina", "patina-v0.3.0"), ("patina", "v0.1.0")]:
            with self.assertRaises(ValueError):
                tag_version(plugin, tag, self.root)

    def test_unknown_tag_names_every_catalogued_plugin(self):
        with self.assertRaisesRegex(ValueError, "rivet-v1.0.0, patina-v1.0.0"):
            release.plan("refs/tags/unknown-v1.0.0", self.root)

    def test_repository_catalog_is_complete(self):
        root = Path(__file__).resolve().parents[2]
        products = json.loads((root / "plugins/catalog.json").read_text())
        check = (root / ".gitea/workflows/check.yml").read_text()
        triggers = (root / ".gitea/workflows/release.yml").read_text()
        ids = [p["clap_id"] for p in products.values()]
        self.assertEqual(len(ids), len(set(ids)), "CLAP IDs must be unique")
        for plugin, product in products.items():
            self.assertTrue((root / product["manifest"]).is_file(), plugin)
            self.assertTrue((root / product["documentation"] / "CHANGELOG.md").is_file(), plugin)
            self.assertIn(f'"{plugin}-v*"', triggers)
            self.assertIn(f"--plugin {plugin} --target", check)
            for source in product.get("licenses", {}):
                self.assertTrue((root / source).is_file(), source)

    def test_individual_zip_excludes_other_plugin_and_is_reproducible(self):
        path = self.package("patina")
        first = path.read_bytes()
        self.package("patina")
        self.assertEqual(first, path.read_bytes())
        with zipfile.ZipFile(path) as z:
            self.assertIn("Patina.clap", z.namelist())
            self.assertNotIn("Rivet.clap", z.namelist())
            self.assertNotIn("Legacy.clap", z.namelist())
            self.assertFalse(any(n.startswith("docs/rivet/") for n in z.namelist()))
            for line in z.read("SHA256SUMS").decode().splitlines():
                digest, name = line.split("  ", 1)
                self.assertEqual(hashlib.sha256(z.read(name)).hexdigest(), digest)

    def test_draft_publish_and_retry_preserve_notes(self):
        self.package("patina")
        client = FakeGitea()
        self.publish(client)
        self.assertFalse(client.release["draft"])
        self.assertEqual(len(client.release["assets"]), 2)
        client.release["body"] = "User edited notes"
        client.calls.clear()
        self.publish(client)
        self.assertEqual(client.release["body"], "User edited notes")
        self.assertTrue(all(call[0] == "GET" for call in client.calls))

    def test_partial_upload_is_repaired_before_draft_is_published(self):
        self.package("patina")
        client = FakeGitea()
        client.fail_upload = True
        with self.assertRaises(RuntimeError):
            self.publish(client)
        self.assertTrue(client.release["draft"])
        self.assertEqual(len(client.release["assets"]), 1)
        client.fail_upload = False
        self.publish(client)
        self.assertFalse(client.release["draft"])
        self.assertEqual(len(client.release["assets"]), 2)

    def test_different_existing_asset_is_never_replaced(self):
        self.package("patina")
        client = FakeGitea()
        self.publish(client)
        client.bytes[client.release["assets"][0]["browser_download_url"]] = b"different"
        client.calls.clear()
        with self.assertRaises(ValueError):
            self.publish(client)
        self.assertTrue(all(call[0] == "GET" for call in client.calls))

    def test_bad_checksum_or_missing_tag_cannot_publish(self):
        path = self.package("rivet")
        path.with_suffix(".zip.sha256").write_text("invalid")
        client = FakeGitea()
        with self.assertRaises(ValueError):
            self.publish(client, "rivet")
        self.assertEqual(client.calls, [])
        self.package("rivet")
        client.tag_exists = False
        with self.assertRaises(ValueError):
            self.publish(client, "rivet")
        self.assertFalse(any(call[0] != "GET" for call in client.calls))

    def test_no_release_notes_fails_before_network(self):
        self.package("patina")
        self.write("docs/patina/CHANGELOG.md", "# No matching release")
        client = FakeGitea()
        with self.assertRaises(ValueError):
            self.publish(client)
        self.assertEqual(client.calls, [])

    def test_foreign_plugin_in_archive_cannot_publish(self):
        path = self.package("patina")
        with zipfile.ZipFile(path, "a") as z:
            z.writestr("Rivet.clap", b"wrong plugin")
        path.with_suffix(".zip.sha256").write_text(f"{hashlib.sha256(path.read_bytes()).hexdigest()}  {path.name}\n")
        client = FakeGitea()
        with self.assertRaises(ValueError):
            self.publish(client)
        self.assertEqual(client.calls, [])

if __name__ == "__main__":
    unittest.main()