ktransformers/.github/model-e2e/test_release_contracts.py

"""Release integration failure cases; no network, GPU, or upload credentials."""

import copy
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

import inside
from contracts import PACKAGES, REPOSITORIES, digest, validate_request, write_json
from release_contracts import verify_install_report, verify_release


def fixture(root):
    house = root / "wheelhouse"
    house.mkdir()
    entries = {}
    wheels = {}
    for package in sorted(PACKAGES):
        name = package.replace("-", "_") + "-1.0-py3-none-any.whl"
        (house / name).write_bytes(package.encode())
        entries[name] = {
            "name": package,
            "version": "1.0",
            "sha256": digest(house / name),
        }
        wheels[package] = {
            "file": name,
            "version": "1.0",
            "sha256": entries[name]["sha256"],
        }
    manifest = {
        "schema": 1,
        "run_id": 123,
        "run_attempt": 1,
        "workflow_sha": "a" * 40,
        "source_lock": {
            "workflow_sha": "a" * 40,
            "sources": {
                name: {"repository": repo, "ref": "refs/heads/main", "sha": "a" * 40}
                for name, repo in REPOSITORIES.items()
            },
        },
        "wheels": wheels,
        "wheelhouse": entries,
        "plans": {
            extra: {name: entry["file"] for name, entry in wheels.items()}
            for extra in ("sglang", "sglang,sft")
        },
    }
    write_json(root / "release.json", manifest)
    return manifest


def report(manifest, public=False):
    return {
        "install": [
            {
                "metadata": {"name": name, "version": entry["version"]},
                "download_info": {
                    "url": (
                        "https://files.pythonhosted.org/packages/"
                        if public
                        else "file:///input/release/wheelhouse/"
                    )
                    + entry["file"],
                    "archive_info": {"hashes": {"sha256": entry["sha256"]}},
                },
            }
            for name, entry in manifest["wheels"].items()
        ]
    }


class ReleaseContracts(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.manifest = fixture(self.root)
        self.sha = digest(self.root / "release.json")

    def test_release_and_both_installation_modes(self):
        verify_release(self.root, self.sha)
        for extra in ("sglang", "sglang,sft"):
            for public in (True, False):
                verify_install_report(
                    report(self.manifest, public), self.manifest, extra, public=public
                )

    def test_wheel_tampering(self):
        next((self.root / "wheelhouse").iterdir()).write_bytes(b"changed")
        with self.assertRaisesRegex(ValueError, "SHA256 changed"):
            verify_release(self.root, self.sha)

    def test_rewritten_manifest(self):
        manifest = self.manifest | {"run_id": 456}
        write_json(self.root / "release.json", manifest)
        with self.assertRaisesRegex(ValueError, "manifest hash"):
            verify_release(self.root, self.sha)

    def test_extra_dependency_file(self):
        (self.root / "wheelhouse/unknown.whl").touch()
        with self.assertRaisesRegex(ValueError, "file set"):
            verify_release(self.root, self.sha)

    def test_report_rejects_same_version_different_wheel(self):
        data = report(self.manifest, True)
        data["install"][0]["download_info"]["archive_info"]["hashes"]["sha256"] = (
            "b" * 64
        )
        with self.assertRaisesRegex(ValueError, "hash changed"):
            verify_install_report(data, self.manifest, "sglang,sft", public=True)

    def test_public_test_rejects_local_artifacts(self):
        with self.assertRaisesRegex(ValueError, "public PyPI"):
            verify_install_report(
                report(self.manifest), self.manifest, "sglang,sft", public=True
            )

    def test_dependency_drift_cannot_pass(self):
        data = report(self.manifest, True)
        data["install"].pop()
        with self.assertRaisesRegex(ValueError, "resolution differs"):
            verify_install_report(data, self.manifest, "sglang,sft", public=True)

    def test_release_request_cannot_impersonate_pr(self):
        request = {
            "schema": 1,
            "mode": "release-candidate",
            "harness_sha": "a" * 40,
            "manifest_sha256": self.sha,
            "build_run_id": 123,
            "build_run_attempt": 1,
            "build_workflow_sha": "a" * 40,
        }
        validate_request(request)
        with self.assertRaises(ValueError):
            validate_request(request | {"pr": {}})

    def test_wheelhouse_symlink(self):
        path = next((self.root / "wheelhouse").iterdir())
        target = self.root / "outside"
        path.rename(target)
        path.symlink_to(target)
        with self.assertRaises(ValueError):
            verify_release(self.root, self.sha)

    def test_wrong_source_branch(self):
        manifest = copy.deepcopy(self.manifest)
        manifest["source_lock"]["sources"]["sglang"]["ref"] = "refs/heads/hotfix"
        write_json(self.root / "release.json", manifest)
        with self.assertRaisesRegex(ValueError, "must be main"):
            verify_release(self.root, digest(self.root / "release.json"))

    def check_installer(self, public):
        work = self.root / "work"
        work.mkdir()
        evidence = work / "evidence"
        evidence.mkdir()
        commands = []

        def run(command, log, **kwargs):
            command = list(map(str, command))
            commands.append(command)
            if "--report" in command:
                write_json(
                    command[command.index("--report") + 1],
                    report(self.manifest, public),
                )
            if "/harness/installed.py" in command:
                write_json(command[-1], {name: {"version": "1.0"} for name in PACKAGES})

        with (
            patch.object(inside, "WORK", work),
            patch.object(inside, "EVIDENCE", evidence),
            patch.object(inside, "PYTHON", work / "venv/bin/python"),
            patch.object(inside, "verify_release", return_value=self.manifest),
            patch.object(inside, "run", side_effect=run),
            patch.object(inside.venv, "EnvBuilder") as builder,
        ):
            builder.return_value.create.side_effect = (
                lambda directory: directory.mkdir()
            )
            inside.install_release(
                {
                    "mode": "release-pypi" if public else "release-candidate",
                    "manifest_sha256": self.sha,
                },
                {},
            )
        installs = [command for command in commands if "--report" in command]
        self.assertEqual(
            [command[-1] for command in installs],
            ["ktransformers[sglang]", "ktransformers[sglang,sft]"],
        )
        self.assertTrue(installs[0][0].endswith("serving-venv/bin/python"))
        self.assertTrue(installs[1][0].endswith("/venv/bin/python"))
        for command in installs:
            self.assertIn("--index-url" if public else "--no-index", command)
            self.assertNotIn("--no-deps", command)

    def test_release_candidate_installs_user_extras_offline_into_two_fresh_venvs(self):
        self.check_installer(False)

    def test_release_pypi_installs_unpinned_user_extras_into_two_fresh_venvs(self):
        self.check_installer(True)


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