vllm/.github/workflows/scripts/test_run_ci_command.py

# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

import unittest
from typing import Any

from run_ci_command import (
    COMMAND_RETRY_FAILED,
    COMMAND_RUN_CI,
    RETRY_STATES,
    BuildkiteClient,
    authorize,
    create_build_payload,
    has_trusted_approval,
    is_active_build,
    is_build_for_pr,
    parse_command,
    parse_trusted_users,
    run,
    select_latest_build,
)


def make_pr(**overrides: Any) -> dict[str, Any]:
    pr = {
        "base": {"ref": "main"},
        "draft": False,
        "head": {
            "ref": "feature",
            "repo": {"clone_url": "https://github.com/contributor/vllm.git"},
            "sha": "0123456789abcdef",
        },
        "labels": [],
        "number": 42,
        "state": "open",
        "user": {"login": "author"},
    }
    pr.update(overrides)
    return pr


def make_event(command: str, actor: str = "reviewer") -> dict[str, Any]:
    return {
        "comment": {
            "body": command,
            "id": 99,
            "user": {"login": actor},
        },
        "issue": {
            "number": 42,
            "pull_request": {},
        },
    }


class FakeGitHub:
    def __init__(
        self,
        *,
        permission: str = "write",
        permissions: dict[str, str] | None = None,
        pr: dict[str, Any] | None = None,
        review_decision: str = "REVIEW_REQUIRED",
        reviews: list[dict[str, Any]] | None = None,
    ) -> None:
        self.comments: list[str] = []
        self.permission = permission
        self.permissions = permissions or {}
        self.pr = pr or make_pr()
        self.reactions: list[str] = []
        self.review_decision = review_decision
        self.reviews = reviews or []

    def get_pr(self, number: int) -> dict[str, Any]:
        return self.pr

    def get_permission(self, actor: str) -> str:
        return self.permissions.get(actor, self.permission)

    def get_review_decision(self, number: int) -> str:
        return self.review_decision

    def list_reviews(self, number: int) -> list[dict[str, Any]]:
        return self.reviews

    def list_reactions(self, comment_id: int) -> list[dict[str, Any]]:
        return []

    def add_reaction(self, comment_id: int, content: str) -> None:
        self.reactions.append(content)

    def add_comment(self, issue_number: int, body: str) -> None:
        self.comments.append(body)


class FakeBuildkite:
    def __init__(
        self,
        build_lists: list[list[dict[str, Any]]] | None = None,
    ) -> None:
        self.build_lists = build_lists or []
        self.created_builds: list[dict[str, Any]] = []
        self.list_calls: list[tuple[str, tuple[str, str] | None]] = []
        self.retry_calls: list[tuple[int, str]] = []

    def list_builds(
        self,
        commit: str,
        *,
        metadata: tuple[str, str] | None = None,
    ) -> list[dict[str, Any]]:
        self.list_calls.append((commit, metadata))
        return self.build_lists.pop(0)

    def create_build(self, body: dict[str, Any]) -> dict[str, Any]:
        self.created_builds.append(body)
        return {
            "number": 123,
            "web_url": "https://buildkite.example/builds/123",
        }

    def retry_failed_jobs(
        self,
        build_number: int,
        states: str,
    ) -> dict[str, Any]:
        self.retry_calls.append((build_number, states))
        return {"retried_jobs_count": 3}


class FakeTransport:
    def __init__(self, response: Any) -> None:
        self.calls: list[dict[str, Any]] = []
        self.response = response

    def request(self, url: str, **kwargs: Any) -> Any:
        self.calls.append({"url": url, **kwargs})
        return self.response


class RunCiCommandTest(unittest.TestCase):
    def test_only_exact_ci_commands_are_accepted(self) -> None:
        self.assertEqual(parse_command(COMMAND_RUN_CI), COMMAND_RUN_CI)
        self.assertEqual(
            parse_command(COMMAND_RETRY_FAILED),
            COMMAND_RETRY_FAILED,
        )
        self.assertIsNone(parse_command("/ci run please"))
        self.assertIsNone(parse_command(" /ci run"))

    def test_write_access_authorizes_reviewers_and_authors(self) -> None:
        allowed, _ = authorize(
            actor="reviewer",
            permission="write",
            pr=make_pr(),
        )
        self.assertTrue(allowed)

    def test_configured_trusted_contributors_can_run_ci(self) -> None:
        trusted_users = parse_trusted_users("trusted-one, TRUSTED-TWO")
        allowed, _ = authorize(
            actor="trusted-two",
            permission="read",
            pr=make_pr(),
            trusted_users=trusted_users,
        )
        self.assertTrue(allowed)

    def test_authors_need_an_approval_or_ready_label(self) -> None:
        pending, _ = authorize(
            actor="author",
            permission="read",
            pr=make_pr(),
        )
        approved, _ = authorize(
            actor="author",
            permission="read",
            pr=make_pr(),
            trusted_approval=True,
        )
        ready, _ = authorize(
            actor="author",
            permission="read",
            pr=make_pr(labels=[{"name": "ready"}]),
        )
        self.assertFalse(pending)
        self.assertTrue(approved)
        self.assertTrue(ready)

    def test_non_author_contributors_without_write_are_denied(self) -> None:
        allowed, _ = authorize(
            actor="contributor",
            permission="read",
            pr=make_pr(),
            trusted_approval=True,
        )
        self.assertFalse(allowed)

    def test_authors_cannot_use_ready_state_on_draft_prs(self) -> None:
        allowed, _ = authorize(
            actor="author",
            permission="read",
            pr=make_pr(draft=True, labels=[{"name": "ready"}]),
            trusted_approval=True,
        )
        self.assertFalse(allowed)

    def test_only_trusted_reviewers_can_delegate_through_approval(self) -> None:
        approved_review = {
            "state": "APPROVED",
            "user": {"login": "reviewer"},
        }
        trusted = FakeGitHub(
            permission="read",
            permissions={"reviewer": "write"},
            review_decision="APPROVED",
            reviews=[approved_review],
        )
        untrusted = FakeGitHub(
            permission="read",
            review_decision="APPROVED",
            reviews=[approved_review],
        )
        self.assertTrue(has_trusted_approval(trusted, 42, set()))
        self.assertFalse(has_trusted_approval(untrusted, 42, set()))

    def test_build_matching_is_scoped_to_the_pr(self) -> None:
        self.assertTrue(is_build_for_pr({"pull_request": {"id": 42}}, 42))
        self.assertFalse(is_build_for_pr({"pull_request": {"id": 43}}, 42))
        self.assertTrue(
            is_build_for_pr(
                {"meta_data": {"github-pr-number": "42"}},
                42,
            )
        )

    def test_latest_build_selection_ignores_other_prs(self) -> None:
        latest = select_latest_build(
            [
                {
                    "created_at": "2026-07-28T02:00:00Z",
                    "number": 3,
                    "pull_request": {"id": 43},
                },
                {
                    "created_at": "2026-07-28T01:00:00Z",
                    "number": 2,
                    "pull_request": {"id": 42},
                },
                {
                    "created_at": "2026-07-28T00:00:00Z",
                    "number": 1,
                    "pull_request": {"id": 42},
                },
            ],
            42,
        )
        self.assertEqual(latest["number"], 2)

    def test_active_build_states_prevent_duplicate_runs(self) -> None:
        self.assertTrue(is_active_build({"state": "scheduled"}))
        self.assertTrue(is_active_build({"state": "running"}))
        self.assertTrue(is_active_build({"state": "waiting"}))
        self.assertTrue(is_active_build({"blocked": True, "state": "passed"}))
        self.assertFalse(is_active_build({"state": "failed"}))

    def test_build_payload_preserves_pr_context(self) -> None:
        payload = create_build_payload(
            actor="reviewer",
            comment_id=99,
            pr=make_pr(labels=[{"name": "ready"}, {"name": "v1"}]),
        )
        self.assertEqual(
            payload,
            {
                "commit": "0123456789abcdef",
                "branch": "feature",
                "message": "PR #42 /ci run by @reviewer",
                "pull_request_id": 42,
                "pull_request_base_branch": "main",
                "pull_request_repository": ("https://github.com/contributor/vllm.git"),
                "pull_request_labels": ["ready", "v1"],
                "ignore_pipeline_branch_filters": True,
                "env": {
                    "VLLM_CI_GITHUB_COMMENT_ID": "99",
                    "VLLM_CI_TRIGGERED_BY": "reviewer",
                },
                "meta_data": {
                    "github-comment-id": "99",
                    "github-pr-number": "42",
                    "github-triggered-by": "reviewer",
                },
            },
        )

    def test_ci_run_dispatches_build_with_current_pr_metadata(self) -> None:
        github = FakeGitHub()
        buildkite = FakeBuildkite([[], []])
        run(make_event(COMMAND_RUN_CI), github, buildkite)

        self.assertEqual(len(buildkite.created_builds), 1)
        self.assertEqual(
            buildkite.created_builds[0]["message"],
            "PR #42 /ci run by @reviewer",
        )
        self.assertEqual(github.reactions, ["eyes", "rocket"])
        self.assertIn("Buildkite CI #123", github.comments[0])

    def test_unapproved_authors_are_denied_without_buildkite(self) -> None:
        github = FakeGitHub(
            permission="read",
            pr=make_pr(),
            review_decision="REVIEW_REQUIRED",
        )
        buildkite = FakeBuildkite()
        run(make_event(COMMAND_RUN_CI, "author"), github, buildkite)

        self.assertEqual(buildkite.list_calls, [])
        self.assertEqual(github.reactions, ["eyes", "-1"])
        self.assertIn("approve the PR", github.comments[0])

    def test_ci_retry_uses_latest_current_sha_build(self) -> None:
        github = FakeGitHub(
            permission="read",
            pr=make_pr(labels=[{"name": "ready"}]),
        )
        buildkite = FakeBuildkite(
            [
                [
                    {
                        "created_at": "2026-07-28T01:00:00Z",
                        "finished_at": "2026-07-28T02:00:00Z",
                        "number": 123,
                        "pull_request": {"id": 42},
                        "state": "failed",
                        "web_url": "https://buildkite.example/builds/123",
                    }
                ]
            ]
        )
        run(make_event(COMMAND_RETRY_FAILED, "author"), github, buildkite)

        self.assertEqual(buildkite.retry_calls, [(123, RETRY_STATES)])
        self.assertIn("Queued 3 failed job", github.comments[0])

    def test_buildkite_retry_uses_retry_failed_jobs_endpoint(self) -> None:
        transport = FakeTransport({"retried_jobs_count": 2})
        client = BuildkiteClient(
            "secret",
            "vllm",
            "ci",
            transport=transport,
        )
        client.retry_failed_jobs(123, RETRY_STATES)

        call = transport.calls[0]
        self.assertEqual(call["method"], "PUT")
        self.assertTrue(call["url"].endswith("/123/retry_failed_jobs"))
        self.assertEqual(call["body"], {"states": RETRY_STATES})


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