ktransformers/.github/release/test_release_stack.py

"""Adversarial promotion checks. All remote calls/uploads are mocked."""

import copy
import json
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent))
from contracts import CASES, digest, write_json
from release_stack import (
    check_evidence,
    check_unused,
    cleanup,
    existing_matches,
    promote,
)
from test_release_contracts import fixture, report


@pytest.fixture
def candidate(tmp_path, monkeypatch):
    root = tmp_path / "candidate"
    root.mkdir()
    manifest = fixture(root)
    sha = digest(root / "release.json")
    evidence = tmp_path / "evidence"
    (evidence / "tests").mkdir(parents=True)
    write_json(
        evidence / "request.json",
        {
            "mode": "release-candidate",
            "manifest_sha256": sha,
            "build_run_id": 123,
            "build_run_attempt": 1,
            "build_workflow_sha": "a" * 40,
            "harness_sha": "a" * 40,
        },
    )
    suite = {
        "status": "passed",
        "cases": {case: {"status": "passed"} for case in CASES},
    }
    for name in ("host-result.json", "tests/suite.json"):
        write_json(evidence / name, suite)
    for label in ("serving", "combined"):
        write_json(evidence / f"tests/release-{label}-install.json", report(manifest))
    monkeypatch.setenv("GITHUB_REF", "refs/heads/main")
    monkeypatch.setenv("GITHUB_EVENT_NAME", "workflow_dispatch")
    monkeypatch.setenv("GITHUB_REPOSITORY", "kvcache-ai/ktransformers")
    monkeypatch.setenv("GITHUB_RUN_ID", "123")
    monkeypatch.setenv("GITHUB_SHA", "a" * 40)
    return root, manifest, sha, evidence


def files(entry):
    return [
        {
            "filename": entry["file"],
            "digests": {"sha256": entry["sha256"]},
            "yanked": False,
        }
    ]


def test_uploads_tested_bytes_in_dependency_order_with_kt_last(candidate, tmp_path):
    root, manifest, sha, evidence = candidate
    uploaded = {}

    def query(name, version):
        return files(manifest["wheels"][name]) if name in uploaded else []

    def upload(path):
        name = next(
            name
            for name, entry in manifest["wheels"].items()
            if entry["file"] == path.name
        )
        assert digest(path) == manifest["wheels"][name]["sha256"]
        uploaded[name] = path

    output = tmp_path / "promotion.json"
    promote(root, sha, evidence, output, query=query, upload=upload)
    assert list(uploaded) == [
        "accelerate-kt",
        "transformers-kt",
        "kt-kernel",
        "sglang-kt",
        "ktransformers",
    ]
    assert json.loads(output.read_text())["status"] == "uploaded-awaiting-public-e2e"


def test_retry_reuses_only_exact_files_without_reupload(candidate, tmp_path):
    root, manifest, sha, evidence = candidate
    promote(
        root,
        sha,
        evidence,
        tmp_path / "promotion.json",
        query=lambda name, version: files(manifest["wheels"][name]),
        upload=lambda path: pytest.fail("Must not re-upload identical published files"),
    )


def test_conflict_in_last_package_fails_before_any_upload(candidate, tmp_path):
    root, manifest, sha, evidence = candidate
    conflicting = copy.deepcopy(manifest["wheels"]["ktransformers"])
    conflicting["sha256"] = "c" * 64
    with pytest.raises(ValueError, match="different bytes"):
        promote(
            root,
            sha,
            evidence,
            tmp_path / "promotion.json",
            query=lambda name, version: files(conflicting)
            if name == "ktransformers"
            else [],
            upload=lambda path: pytest.fail("No writes before full batch preflight"),
        )


@pytest.mark.parametrize(
    "field,value",
    [
        ("mode", "release-pypi"),
        ("manifest_sha256", "c" * 64),
        ("build_run_id", 456),
        ("build_run_attempt", 2),
        ("harness_sha", "c" * 40),
    ],
)
def test_wrong_acceptance_identity_fails(candidate, field, value):
    root, manifest, sha, evidence = candidate
    path = evidence / "request.json"
    request = json.loads(path.read_text())
    write_json(path, request | {field: value})
    with pytest.raises(ValueError):
        check_evidence(evidence, manifest, sha, "candidate")


@pytest.mark.parametrize("status", ["failed", "resource_unavailable", "not_run"])
def test_missing_or_failed_model_blocks_promotion(candidate, status):
    root, manifest, sha, evidence = candidate
    path = evidence / "tests/suite.json"
    result = json.loads(path.read_text())
    result["cases"]["glm53_inference"]["status"] = status
    write_json(path, result)
    with pytest.raises(ValueError, match="model acceptance"):
        check_evidence(evidence, manifest, sha, "candidate")


def test_incomplete_suite_never_passes(candidate):
    root, manifest, sha, evidence = candidate
    path = evidence / "tests/suite.json"
    result = json.loads(path.read_text())
    del result["cases"]["deepseek_v31_lora"]
    write_json(path, result)
    with pytest.raises(ValueError):
        check_evidence(evidence, manifest, sha, "candidate")


def test_pypi_check_must_use_public_downloads(candidate):
    root, manifest, sha, evidence = candidate
    request = json.loads((evidence / "request.json").read_text())
    write_json(evidence / "request.json", request | {"mode": "release-pypi"})
    with pytest.raises(ValueError, match="public PyPI"):
        check_evidence(evidence, manifest, sha, "pypi")


def test_existing_version_cannot_be_rebuilt_under_same_name():
    with pytest.raises(ValueError, match="already exists"):
        check_unused(
            [{"name": "kt-kernel", "version": "1.0"}], query=lambda *args: [{}]
        )


def test_yanked_or_extra_remote_files_are_not_skipped():
    entry = {"file": "x.whl", "sha256": "a" * 64}
    for remote in (files(entry) * 2, [files(entry)[0] | {"yanked": True}]):
        with pytest.raises(ValueError):
            existing_matches(entry, remote)


def test_partial_upload_never_claims_release_success(candidate, tmp_path):
    root, manifest, sha, evidence = candidate

    def fail(path):
        raise RuntimeError("network failure")

    output = tmp_path / "promotion.json"
    with pytest.raises(RuntimeError):
        promote(root, sha, evidence, output, query=lambda *args: [], upload=fail)
    assert json.loads(output.read_text())["status"] == "partial-or-failed-upload"


def test_cleanup_is_scoped_to_an_owned_run_directory(tmp_path, monkeypatch):
    monkeypatch.setenv("GITHUB_RUN_ID", "123")
    with pytest.raises(ValueError):
        cleanup(tmp_path, tmp_path.parent)
    owned = tmp_path / "kt-four-main.example"
    owned.mkdir()
    (owned / ".kt-release-owned").write_text("456")
    with pytest.raises(ValueError):
        cleanup(owned, tmp_path)
    (owned / ".kt-release-owned").write_text("123")
    cleanup(owned, tmp_path)
    assert not owned.exists()