rustfs/scripts/test_reference_binary.py

#!/usr/bin/env python3
# Copyright 2026 RustFS Team
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Exercise the reference-binary pin resolver and unpacker without network access."""

import hashlib
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
import zipfile

ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / ".github/actions/reference-binary/reference_binary.py"
PIN = '''schema = 1
repository = "rustfs/rustfs"
version = "1.0.5"
tag = "1.0.5"

[platforms.linux-x86_64-gnu]
asset = "rustfs-linux-x86_64-gnu-v1.0.5.zip"
sha256 = "{sha}"
executable = "rustfs"
'''
SHA = "a" * 64


def run(*args):
    proc = subprocess.run([sys.executable, str(SCRIPT), *args], capture_output=True, text=True, check=False)
    return proc.returncode, proc.stdout, proc.stderr


class Base(unittest.TestCase):
    def setUp(self):
        temp = tempfile.TemporaryDirectory()
        self.addCleanup(temp.cleanup)
        self.root = Path(temp.name)

    def pin(self, text):
        path = self.root / "pin.toml"
        path.write_text(text)
        return path

    def archive(self, members):
        path = self.root / "rustfs.zip"
        with zipfile.ZipFile(path, "w") as bundle:
            for name, data in members.items():
                bundle.writestr(name, data)
        return path, hashlib.sha256(path.read_bytes()).hexdigest()


class ResolveTests(Base):
    def test_missing_pin_file_fails(self):
        code, _, err = run("resolve", "--pin", str(self.root / "absent.toml"), "--platform", "linux-x86_64-gnu")
        self.assertEqual(code, 1)
        self.assertIn("reference pin file is missing", err)

    def test_unpinned_platform_fails(self):
        code, _, err = run("resolve", "--pin", str(self.pin(PIN.format(sha=SHA))), "--platform", "linux-aarch64-gnu")
        self.assertEqual(code, 1)
        self.assertIn("no pin for platform", err)

    def test_malformed_pins_fail(self):
        good = PIN.format(sha=SHA)
        for label, text in {
            "short digest": good.replace(SHA, "a" * 63),
            "uppercase digest": good.replace(SHA, "A" * 64),
            "no digest": good.replace(f'sha256 = "{SHA}"\n', ""),
            "path in asset": good.replace('asset = "rustfs-', 'asset = "../rustfs-'),
            "non-zip asset": good.replace("v1.0.5.zip", "v1.0.5.deb"),
            "asset of another version": good.replace("v1.0.5.zip", "v1.0.4.zip"),
            "path in executable": good.replace('executable = "rustfs"', 'executable = "bin/rustfs"'),
            "wrong schema": good.replace("schema = 1", "schema = 2"),
            "boolean schema": good.replace("schema = 1", "schema = true"),
            "no repository": good.replace('repository = "rustfs/rustfs"\n', ""),
            "unknown key": good + 'url = "https://example.invalid"\n',
            "not toml": "schema = \n",
        }.items():
            with self.subTest(label):
                code, _, _ = run("resolve", "--pin", str(self.pin(text)), "--platform", "linux-x86_64-gnu")
                self.assertEqual(code, 1)

    def test_resolves_identity_as_step_outputs(self):
        code, out, err = run("resolve", "--pin", str(self.pin(PIN.format(sha=SHA))), "--platform", "linux-x86_64-gnu")
        self.assertEqual(code, 0, err)
        self.assertEqual(out.splitlines(), ["repository=rustfs/rustfs", "version=1.0.5", "tag=1.0.5",
                                            "asset=rustfs-linux-x86_64-gnu-v1.0.5.zip", f"sha256={SHA}",
                                            "executable=rustfs"])

    def test_repository_pin_is_valid(self):
        code, out, err = run("resolve", "--pin", str(ROOT / ".config/reference-binary.toml"),
                             "--platform", "linux-x86_64-gnu")
        self.assertEqual(code, 0, err)
        self.assertIn("version=1.0.5", out.splitlines())


class UnpackTests(Base):
    def unpack(self, archive, sha):
        return run("unpack", "--archive", str(archive), "--sha256", sha, "--executable", "rustfs",
                   "--destination", str(self.root / "out"))

    def test_digest_mismatch_fails_and_extracts_nothing(self):
        archive, _ = self.archive({"rustfs": b"binary"})
        code, _, err = self.unpack(archive, SHA)
        self.assertEqual(code, 1)
        self.assertIn("does not match the pinned digest", err)
        self.assertFalse((self.root / "out").exists())

    def test_missing_archive_fails(self):
        code, _, err = self.unpack(self.root / "absent.zip", SHA)
        self.assertEqual(code, 1)
        self.assertIn("missing or unreadable", err)

    def test_archive_without_top_level_executable_fails(self):
        archive, sha = self.archive({"bin/rustfs": b"binary"})
        code, _, err = self.unpack(archive, sha)
        self.assertEqual(code, 1)
        self.assertIn("exactly one top-level", err)

    def test_empty_executable_fails(self):
        archive, sha = self.archive({"rustfs": b""})
        code, _, _ = self.unpack(archive, sha)
        self.assertEqual(code, 1)

    def test_non_zip_archive_fails(self):
        archive = self.root / "rustfs.zip"
        archive.write_bytes(b"not a zip")
        code, _, err = self.unpack(archive, hashlib.sha256(b"not a zip").hexdigest())
        self.assertEqual(code, 1)
        self.assertIn("not a valid zip", err)

    def test_extracts_only_the_executable(self):
        archive, sha = self.archive({"rustfs": b"binary", "../escape": b"x", "README": b"r"})
        code, out, err = self.unpack(archive, sha)
        self.assertEqual(code, 0, err)
        target = self.root / "out" / "rustfs"
        self.assertEqual(out.strip(), f"path={target}")
        self.assertEqual(target.read_bytes(), b"binary")
        self.assertTrue(target.stat().st_mode & 0o111)
        self.assertEqual(sorted(p.name for p in (self.root / "out").iterdir()), ["rustfs"])
        self.assertFalse((self.root / "escape").exists())


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