restructure

scripts/tests/test_release.py

@@ -0,0 +1,187 @@
+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_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()