buildkite/bazelci_test.py

#!/usr/bin/env python3
#
# Copyright 2023 The Bazel Authors. All rights reserved.
#
# 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.

import os

os.environ["BUILDKITE_ORGANIZATION_SLUG"] = "bazel"
os.environ["BUILDKITE_PIPELINE_SLUG"] = "test"

import bazelci
import shlex
import tempfile
import unittest
from unittest import mock
import yaml


class CalculateFlags(unittest.TestCase):
    _CONFIGS = yaml.safe_load(
        """
.base_flags: &base_flags
  ? "--enable_a"
  ? "--enable_b"

tasks:
  basic:
    build_flags:
      - "--enable_x"
      - "--enable_y"
    build_targets:
      - "//..."
  json_profile:
    build_flags:
      - "--enable_x"
      - "--enable_y"
    build_targets:
      - "//..."
    include_json_profile:
      - build
  capture_corrupted:
    build_flags:
      - "--enable_x"
      - "--enable_y"
    build_targets:
      - "//..."
    capture_corrupted_outputs:
      - build
  no_flags:
    test_targets:
      - "//..."
  merge_flags:
    build_flags:
      <<: *base_flags
      ? "--enable_z"
      ? "--enable_w"
    build_targets:
      - "//..."
  flatten_flags:
    build_flags:
      - "--enable_z"
      - ["--enable_y", "--enable_x"]
      - "--enable_w"
    """
    )

    def test_basic_functionality(self):
        tasks = self._CONFIGS.get("tasks")
        flags, json_profile_out, capture_corrupted_outputs_dir = bazelci.calculate_flags(
            tasks.get("basic"), "build_flags", "build", "/tmp", ["HOME"]
        )
        self.assertEqual(flags, ["--enable_x", "--enable_y", "--test_env=HOME"])
        self.assertEqual(json_profile_out, None)
        self.assertEqual(capture_corrupted_outputs_dir, None)

    def test_json_profile(self):
        tasks = self._CONFIGS.get("tasks")
        flags, json_profile_out, capture_corrupted_outputs_dir = bazelci.calculate_flags(
            tasks.get("json_profile"), "build_flags", "build", "/tmp", ["HOME"]
        )
        expected_profile = os.path.join("/tmp", "build.profile.gz")
        self.assertEqual(
            flags,
            ["--enable_x", "--enable_y", "--profile={}".format(expected_profile), "--test_env=HOME"],
        )
        self.assertEqual(json_profile_out, expected_profile)

    def test_capture_corrupted(self):
        tasks = self._CONFIGS.get("tasks")
        flags, json_profile_out, capture_corrupted_outputs_dir = bazelci.calculate_flags(
            tasks.get("capture_corrupted"), "build_flags", "build", "/tmp", ["HOME"]
        )
        expected_corrupted_dir = os.path.join("/tmp", "build_corrupted_outputs")
        self.assertEqual(
            flags,
            [
                "--enable_x",
                "--enable_y",
                "--experimental_remote_capture_corrupted_outputs={}".format(expected_corrupted_dir),
                "--test_env=HOME",
            ],
        )
        self.assertEqual(capture_corrupted_outputs_dir, expected_corrupted_dir)

    def test_no_flags_in_config(self):
        tasks = self._CONFIGS.get("tasks")
        flags, json_profile_out, capture_corrupted_outputs_dir = bazelci.calculate_flags(
            tasks.get("no_flags"), "build_flags", "build", "/tmp", ["HOME"]
        )
        self.assertEqual(flags, ["--test_env=HOME"])

    def test_merge_flags(self):
        tasks = self._CONFIGS.get("tasks")
        flags, json_profile_out, capture_corrupted_outputs_dir = bazelci.calculate_flags(
            tasks.get("merge_flags"), "build_flags", "build", "/tmp", ["HOME"]
        )
        self.assertEqual(
            flags,
            ["--enable_a", "--enable_b", "--enable_z", "--enable_w", "--test_env=HOME"],
        )

    def test_flatten_flags(self):
        tasks = self._CONFIGS.get("tasks")
        flags, json_profile_out, capture_corrupted_outputs_dir = bazelci.calculate_flags(
            tasks.get("flatten_flags"), "build_flags", "build", "/tmp", ["HOME"]
        )
        self.assertEqual(
            flags,
            [
                "--enable_z",
                "--enable_y",
                "--enable_x",
                "--enable_w",
                "--test_env=HOME"],
        )


class CalculateTargets(unittest.TestCase):
    _CONFIGS = yaml.safe_load(
        """
.base_targets: &base_targets
  ? "//..."
  ? "-//experimental/..."

tasks:
  basic:
    build_targets:
      - "//..."
      - "-//bad/..."
  merge:
    build_targets:
      <<: *base_targets
      ? "//experimental/good/..."

  flatten:
    build_targets:
      - "//..."
      - ["-//bad/one", "-//bad/two"]
      - "-//bad/three"
    """
    )

    def test_basic_functionality(self):
        tasks = self._CONFIGS.get("tasks")
        build_targets, test_targets, coverage_targets, index_targets = bazelci.calculate_targets(
            tasks.get("basic"),
            "bazel",
            build_only=False,
            test_only=False,
            workspace_dir="/tmp",
            ws_setup_func=None,
            git_commit="abcd",
            test_flags=[],
        )
        self.assertEqual(build_targets, ["//...", "-//bad/..."])
        self.assertEqual(test_targets, [])
        self.assertEqual(coverage_targets, [])
        self.assertEqual(index_targets, [])

    def test_merge(self):
        tasks = self._CONFIGS.get("tasks")
        build_targets, test_targets, coverage_targets, index_targets = bazelci.calculate_targets(
            tasks.get("merge"),
            "bazel",
            build_only=False,
            test_only=False,
            workspace_dir="/tmp",
            ws_setup_func=None,
            git_commit="abcd",
            test_flags=[],
        )
        self.assertEqual(build_targets, ["//...", "-//experimental/...", "//experimental/good/..."])

    def test_flatten(self):
        tasks = self._CONFIGS.get("tasks")
        build_targets, test_targets, coverage_targets, index_targets = bazelci.calculate_targets(
            tasks.get("flatten"),
            "bazel",
            build_only=False,
            test_only=False,
            workspace_dir="/tmp",
            ws_setup_func=None,
            git_commit="abcd",
            test_flags=[],
        )
        self.assertEqual(build_targets, [
            "//...",
            "-//bad/one",
            "-//bad/two",
            "-//bad/three",
        ])

class MatrixExpansion(unittest.TestCase):
    _CONFIGS = yaml.safe_load(
        """
matrix:
  bazel: ["1.2.3", "2.3.4"]
  platform: ["pf1", "pf2"]
  exit_status: [1, 2]
tasks:
  basic:
    name: "Basic"
  unformatted:
    name: "Unformatted"
    bazel: ${{ bazel }}
  without_name:
    bazel: ${{ bazel }}
  formatted:
    name: "Formatted w/ Bazel v{bazel} on {platform}"
    bazel: ${{ bazel }}
    platform: ${{ platform }}
  exit_status:
    name: "Exit Status: {exit_status}"
    soft_fail:
      - exit_status: ${{exit_status}}
    """
    )

    def test_basic_functionality(self):
        import copy

        config = copy.deepcopy(self._CONFIGS)

        bazelci.expand_task_config(config)
        expanded_tasks = config["tasks"]

        self.assertEqual(
            list(expanded_tasks.values()),
            [
                # no matrix expansion
                dict(name="Basic"),
                dict(name="Unformatted", bazel="1.2.3"),
                dict(name="Unformatted", bazel="2.3.4"),
                # no name
                dict(bazel="1.2.3"),
                dict(bazel="2.3.4"),
                dict(name="Formatted w/ Bazel v1.2.3 on pf1", bazel="1.2.3", platform="pf1"),
                dict(name="Formatted w/ Bazel v1.2.3 on pf2", bazel="1.2.3", platform="pf2"),
                dict(name="Formatted w/ Bazel v2.3.4 on pf1", bazel="2.3.4", platform="pf1"),
                dict(name="Formatted w/ Bazel v2.3.4 on pf2", bazel="2.3.4", platform="pf2"),
                dict(name="Exit Status: 1", soft_fail=[dict(exit_status=1)]),
                dict(name="Exit Status: 2", soft_fail=[dict(exit_status=2)]),
            ],
        )

    def test_embedded_matrix_expansion(self):
        config = yaml.safe_load(
            """
matrix:
  bazel_version:
    7.x: 7.7.1
    8.x: 8.6.0
tasks:
  embedded:
    name: "Embedded {bazel_version}"
    build_flags:
      - --config=bazel-${{ bazel_version }}
      - --bazel_version=${{ bazel_version }}
      - prefix-${{ bazel_version }}-suffix
            """
        )

        bazelci.expand_task_config(config)
        expanded_tasks = list(config["tasks"].values())

        self.assertEqual(
            expanded_tasks,
            [
                dict(
                    name="Embedded 7.x",
                    build_flags=[
                        "--config=bazel-7.7.1",
                        "--bazel_version=7.7.1",
                        "prefix-7.7.1-suffix",
                    ],
                ),
                dict(
                    name="Embedded 8.x",
                    build_flags=[
                        "--config=bazel-8.6.0",
                        "--bazel_version=8.6.0",
                        "prefix-8.6.0-suffix",
                    ],
                ),
            ],
        )


class MatrixExclude(unittest.TestCase):
    _CONFIGS_SINGLE_EXCLUDE = yaml.safe_load(
        """
matrix:
  bazel: ["1.2.3", "2.3.4"]
  platform: ["pf1", "pf2"]
  exclude:
    - bazel: "1.2.3"
      platform: "pf2"
tasks:
  formatted:
    name: "Formatted w/ Bazel v{bazel} on {platform}"
    bazel: ${{ bazel }}
    platform: ${{ platform }}
        """
    )

    _CONFIGS_MULTIPLE_EXCLUDES = yaml.safe_load(
        """
matrix:
  bazel: ["1.2.3", "2.3.4"]
  platform: ["pf1", "pf2"]
  exclude:
    - bazel: "1.2.3"
      platform: "pf2"
    - bazel: "2.3.4"
      platform: "pf1"
tasks:
  formatted:
    name: "Formatted w/ Bazel v{bazel} on {platform}"
    bazel: ${{ bazel }}
    platform: ${{ platform }}
        """
    )

    _CONFIGS_PARTIAL_EXCLUDE = yaml.safe_load(
        """
matrix:
  bazel: ["1.2.3", "2.3.4"]
  platform: ["pf1", "pf2"]
  exclude:
    - platform: "pf2"
tasks:
  formatted:
    name: "Formatted w/ Bazel v{bazel} on {platform}"
    bazel: ${{ bazel }}
    platform: ${{ platform }}
        """
    )

    _CONFIGS_DICT_EXCLUDE = yaml.safe_load(
        """
matrix:
  flags:
    set-one: ["--one", "--two"]
    set-two: ["--three", "--four"]
  platform: ["pf1", "pf2"]
  exclude:
    - platform: "pf2"
      flags: set-one
tasks:
  formatted:
    name: "Formatted on {platform} with flags {flags}"
    platform: ${{ platform }}
    build_flags: ${{ flags }}
        """
    )

    def test_single_exclude(self):
        import copy
        config = copy.deepcopy(self._CONFIGS_SINGLE_EXCLUDE)

        bazelci.expand_task_config(config)
        expanded_tasks = config["tasks"]
        # Total combinations: 2 * 2 = 4, minus 1 excluded = 3
        self.assertEqual(len(expanded_tasks), 3)
        expanded_task_names = [task.get("name", None) for id, task in expanded_tasks.items()]
        self.assertEqual(expanded_task_names, [
            "Formatted w/ Bazel v1.2.3 on pf1",
            "Formatted w/ Bazel v2.3.4 on pf1",
            "Formatted w/ Bazel v2.3.4 on pf2",
        ])

    def test_multiple_excludes(self):
        import copy
        config = copy.deepcopy(self._CONFIGS_MULTIPLE_EXCLUDES)

        bazelci.expand_task_config(config)
        expanded_tasks = config["tasks"]
        # Total combinations: 2 * 2 = 4, minus 2 excluded = 2
        self.assertEqual(len(expanded_tasks), 2)
        expanded_task_names = [task.get("name", None) for id, task in expanded_tasks.items()]
        self.assertEqual(expanded_task_names, [
            "Formatted w/ Bazel v1.2.3 on pf1",
            "Formatted w/ Bazel v2.3.4 on pf2",
        ])

    def test_partial_attribute_exclude(self):
        import copy
        config = copy.deepcopy(self._CONFIGS_PARTIAL_EXCLUDE)

        bazelci.expand_task_config(config)
        expanded_tasks = config["tasks"]
        # Total combinations: 2 * 2 = 4, minus 2 excluded (all pf2) = 2
        self.assertEqual(len(expanded_tasks), 2)
        expanded_task_names = [task.get("name", None) for id, task in expanded_tasks.items()]
        self.assertEqual(expanded_task_names, [
            "Formatted w/ Bazel v1.2.3 on pf1",
            "Formatted w/ Bazel v2.3.4 on pf1",
        ])

    def test_dict_attribute_exclude(self):
        import copy
        config = copy.deepcopy(self._CONFIGS_DICT_EXCLUDE)

        bazelci.expand_task_config(config)
        expanded_tasks = config["tasks"]
        # Total combinations: 2 * 2 = 4, minus 1 excluded = 3
        self.assertEqual(len(expanded_tasks), 3)
        expanded_task_names = [task.get("name", None) for id, task in expanded_tasks.items()]
        # the name receives the friendly alias
        self.assertEqual(expanded_task_names, [
            "Formatted on pf1 with flags set-one",
            "Formatted on pf1 with flags set-two",
            "Formatted on pf2 with flags set-two",
        ])
        expanded_task_flags = [task.get("build_flags", None) for id, task in expanded_tasks.items()]
        # but the flags get the real value
        self.assertEqual(expanded_task_flags, [
            ["--one", "--two"],
            ["--three", "--four"],
            ["--three", "--four"],
        ])


class ShellQuoting(unittest.TestCase):
    """Regression tests: values interpolated into generated Buildkite step
    commands must be shell-quoted so that unusual task names (which come from
    user-controlled `.bazelci/presubmit.yml` dict keys) cannot break out of the
    --task argument and inject extra shell commands."""

    def _runner_command(self, step):
        # create_step stores the list of shell commands under "command".
        commands = step["command"]
        for cmd in commands:
            if "--task=" in cmd:
                return cmd
        self.fail("no command containing --task= was generated")

    def _sentinel_path(self):
        # A uniquely-named path that an unquoted injection would create as its
        # side effect. We never create it ourselves, so it must not exist after
        # the command is generated.
        path = tempfile.mktemp(prefix="pwned.")
        self.addCleanup(lambda: os.remove(path) if os.path.exists(path) else None)
        return path

    def test_runner_step_quotes_task_name(self):
        sentinel = self._sentinel_path()
        # If interpolated unquoted, the `;` would end the runner invocation and
        # `touch <sentinel>` would run as a separate command.
        task = "x; touch %s; #" % sentinel

        step = bazelci.runner_step(
            platform=bazelci.DEFAULT_PLATFORM,
            task=task,
            project_name="test",
        )
        cmd = self._runner_command(step)
        # Parse the command exactly as a shell would.
        tokens = shlex.split(cmd)
        # The whole task name must survive as a single --task= argument...
        self.assertIn("--task=" + task, tokens)
        # ...the injected `touch` must NOT appear as its own token...
        self.assertNotIn("touch", tokens)
        # ...and the injection's side-effect file must never be created.
        self.assertFalse(os.path.exists(sentinel))

    def test_bazel_build_step_quotes_task_name(self):
        sentinel = self._sentinel_path()
        task = "x; touch %s; #" % sentinel

        step = bazelci.bazel_build_step(
            task=task,
            platform=bazelci.DEFAULT_PLATFORM,
            project_name="test",
        )
        cmd = self._runner_command(step)
        tokens = shlex.split(cmd)
        self.assertIn("--task=" + task, tokens)
        self.assertNotIn("touch", tokens)
        self.assertFalse(os.path.exists(sentinel))

    def test_runner_step_quotes_config_arguments(self):
        sentinel = self._sentinel_path()

        step = bazelci.runner_step(
            platform=bazelci.DEFAULT_PLATFORM,
            task="ok",
            project_name="test",
            file_config="x; touch %s; #" % sentinel,
            git_commit="$(touch %s)" % sentinel,
        )
        cmd = self._runner_command(step)
        tokens = shlex.split(cmd)
        self.assertIn("--file_config=x; touch %s; #" % sentinel, tokens)
        self.assertIn("--git_commit=$(touch %s)" % sentinel, tokens)
        self.assertNotIn("touch", tokens)
        self.assertFalse(os.path.exists(sentinel))


class InitialSteps(unittest.TestCase):
    def test_blocks_config_changes_without_presubmit_auto_run_label(self):
        with mock.patch.dict(
            os.environ,
            {
                "BUILDKITE_BRANCH": "feature",
                "BUILDKITE_PIPELINE_DEFAULT_BRANCH": "main",
                "BUILDKITE_PULL_REQUEST_LABELS": "",
            },
        ), mock.patch.object(
            bazelci, "get_modified_files", return_value=[".bazelci/presubmit.yml"]
        ):
            steps = bazelci.create_initial_steps()

        self.assertEqual(len(steps), 1)
        self.assertIn("block", steps[0])

    def test_presubmit_auto_run_label_skips_config_change_block(self):
        for label in ("presubmit-auto-run", "CI:run"):
            with self.subTest(label=label):
                with mock.patch.dict(
                    os.environ,
                    {
                        "BUILDKITE_BRANCH": "feature",
                        "BUILDKITE_PIPELINE_DEFAULT_BRANCH": "main",
                        "BUILDKITE_PULL_REQUEST_LABELS": f"foo,{label},bar",
                    },
                ), mock.patch.object(bazelci, "get_modified_files") as get_modified_files:
                    steps = bazelci.create_initial_steps()
                self.assertEqual(steps, [])
                get_modified_files.assert_not_called()

    def test_has_presubmit_auto_run_label(self):
        with mock.patch.dict(os.environ, {"BUILDKITE_PULL_REQUEST_LABELS": ""}):
            self.assertFalse(bazelci.has_presubmit_auto_run_label())
        with mock.patch.dict(os.environ, {"BUILDKITE_PULL_REQUEST_LABELS": "foo,bar"}):
            self.assertFalse(bazelci.has_presubmit_auto_run_label())
        with mock.patch.dict(os.environ, {"BUILDKITE_PULL_REQUEST_LABELS": "presubmit-auto-run"}):
            self.assertTrue(bazelci.has_presubmit_auto_run_label())
        with mock.patch.dict(os.environ, {"BUILDKITE_PULL_REQUEST_LABELS": "foo,CI:run,bar"}):
            self.assertTrue(bazelci.has_presubmit_auto_run_label())

class FetchCiScripts(unittest.TestCase):
    def test_curl_download_command(self):
        cmd = bazelci.curl_download_command("https://example.com/foo.py", "foo.py")
        self.assertEqual(
            cmd,
            f"curl {bazelci.CURL_FLAGS_STR} https://example.com/foo.py -o foo.py",
        )

    def test_fetch_ci_scripts_command_returns_list_of_curl_commands(self):
        cmds = bazelci.fetch_ci_scripts_command()
        self.assertIsInstance(cmds, list)
        self.assertEqual(len(cmds), 2)
        self.assertTrue(cmds[0].startswith(f"curl {bazelci.CURL_FLAGS_STR}"))
        self.assertIn("bazelci.py", cmds[0])
        self.assertTrue(cmds[1].startswith(f"curl {bazelci.CURL_FLAGS_STR}"))
        self.assertIn("collect_metrics.py", cmds[1])
        # Verify no shell delimiter like ';' or '&&' is appended
        self.assertFalse(cmds[0].endswith(";"))
        self.assertFalse(cmds[0].endswith("&&"))

    def test_create_step_flattens_nested_commands(self):
        step = bazelci.create_step(
            label="test",
            commands=[
                bazelci.fetch_ci_scripts_command(),
                "echo hello",
            ],
            platform=bazelci.DEFAULT_PLATFORM,
        )
        self.assertEqual(len(step["command"]), 3)
        self.assertIn("bazelci.py", step["command"][0])
        self.assertIn("collect_metrics.py", step["command"][1])
        self.assertEqual(step["command"][2], "echo hello")

    def test_runner_step_includes_fetch_ci_scripts(self):
        step = bazelci.runner_step(
            platform=bazelci.DEFAULT_PLATFORM,
            task="basic",
            project_name="test",
        )
        commands = step["command"]
        self.assertEqual(len(commands), 3)
        self.assertIn("bazelci.py", commands[0])
        self.assertIn("collect_metrics.py", commands[1])
        self.assertIn("--task=basic", commands[2])


class GetCiScriptRefTest(unittest.TestCase):

    def test_production_returns_master(self):
        with mock.patch.object(bazelci, "THIS_IS_TESTING", False):
            self.assertEqual(bazelci.get_ci_script_ref(), "master")

    def test_ci_repo_with_commit(self):
        with mock.patch.object(bazelci, "THIS_IS_TESTING", True):
            with mock.patch.dict(
                os.environ,
                {
                    "BUILDKITE_REPO": "https://github.com/bazelbuild/continuous-integration.git",
                    "BUILDKITE_COMMIT": "deadbeef1234",
                },
                clear=True,
            ):
                self.assertEqual(bazelci.get_ci_script_ref(), "deadbeef1234")

    def test_ci_repo_with_head_commit_uses_branch(self):
        with mock.patch.object(bazelci, "THIS_IS_TESTING", True):
            with mock.patch.dict(
                os.environ,
                {
                    "BUILDKITE_REPO": "https://github.com/bazelbuild/continuous-integration.git",
                    "BUILDKITE_COMMIT": "HEAD",
                    "BUILDKITE_BRANCH": "pr-branch-name",
                },
                clear=True,
            ):
                self.assertEqual(bazelci.get_ci_script_ref(), "pr-branch-name")

    def test_other_repo_defaults_to_testing(self):
        with mock.patch.object(bazelci, "THIS_IS_TESTING", True):
            with mock.patch.dict(
                os.environ,
                {
                    "BUILDKITE_REPO": "https://github.com/bazelbuild/bazel.git",
                    "BUILDKITE_COMMIT": "some_bazel_commit",
                },
                clear=True,
            ):
                self.assertEqual(bazelci.get_ci_script_ref(), "testing")

    def test_fork_repo_defaults_to_testing(self):
        with mock.patch.object(bazelci, "THIS_IS_TESTING", True):
            with mock.patch.dict(
                os.environ,
                {
                    "BUILDKITE_REPO": "https://github.com/forkuser/continuous-integration.git",
                    "BUILDKITE_COMMIT": "fork_commit",
                },
                clear=True,
            ):
                self.assertEqual(bazelci.get_ci_script_ref(), "testing")


class GitBranchHandlingTest(unittest.TestCase):

    def test_upload_project_pipeline_step_with_git_commit(self):
        step = bazelci.upload_project_pipeline_step(
            project_name="Protobuf 35.x",
            git_repository="https://github.com/protocolbuffers/protobuf.git",
            http_config=None,
            file_config=".bazelci/presubmit.yml",
            git_commit="origin/35.x",
        )
        commands = step["command"]
        project_pipeline_cmd = [c for c in commands if "project_pipeline" in c][0]
        self.assertIn("--git_commit=origin/35.x", project_pipeline_cmd)
        self.assertIn("--project_name='Protobuf 35.x'", project_pipeline_cmd)
        self.assertIn('--file_config=.bazelci/presubmit.yml', project_pipeline_cmd)

    def test_print_bazel_downstream_pipeline_passes_origin_branch_as_git_commit(self):
        test_projects = {
            "Protobuf 35.x": {
                "git_repository": "https://github.com/protocolbuffers/protobuf.git",
                "git_branch": "35.x",
                "file_config": ".bazelci/presubmit.yml",
                "pipeline_slug": "protobuf",
            }
        }
        with mock.patch.dict(bazelci.DOWNSTREAM_PROJECTS, test_projects, clear=True), \
             mock.patch.object(bazelci, "upload_project_pipeline_step") as mock_upload, \
             mock.patch.object(bazelci, "print_pipeline_steps"), \
             mock.patch.object(bazelci, "bazel_build_step", return_value={"label": "Bazel"}), \
             mock.patch.object(bazelci, "get_platform_for_task", return_value="ubuntu2004"):
            bazelci.print_bazel_downstream_pipeline(
                task_configs={"basic": {}},
                http_config=None,
                file_config=None,
                test_disabled_projects=False,
                notify=False,
            )
            mock_upload.assert_called_once_with(
                project_name="Protobuf 35.x",
                git_repository="https://github.com/protocolbuffers/protobuf.git",
                http_config=None,
                file_config=".bazelci/presubmit.yml",
                git_commit="origin/35.x",
            )

    def test_get_last_green_commit_ignores_projects_with_git_branch(self):
        with mock.patch.dict(
            bazelci.DOWNSTREAM_PROJECTS,
            {
                "TestProject": {
                    "git_repository": "https://github.com/foo/bar.git",
                    "pipeline_slug": "bar",
                    "git_branch": "35.x",
                }
            },
            clear=True,
        ):
            self.assertIsNone(bazelci.get_last_green_commit("TestProject"))

    def test_clone_git_repository_resets_to_git_commit(self):
        executed_commands = []

        def fake_execute(args, print_output=True, suppress_stdout=False):
            executed_commands.append(args)

        with mock.patch.object(bazelci, "execute_command", side_effect=fake_execute), \
             mock.patch("os.path.exists", return_value=True), \
             mock.patch("os.chdir"):
            bazelci.clone_git_repository(
                "https://github.com/protocolbuffers/protobuf.git",
                git_commit="origin/35.x",
                suppress_stdout=True,
            )

        self.assertIn(["git", "reset", "origin/35.x", "--hard"], executed_commands)


class ExecuteCommandTimeout(unittest.TestCase):
    def test_execute_command_raises_on_timeout(self):
        with self.assertRaises(bazelci.subprocess.TimeoutExpired):
            bazelci.execute_command(["sleep", "5"], timeout=0.2)

    def test_upload_test_logs_from_bep_does_not_raise_on_timeout(self):
        with mock.patch.object(
            bazelci, "download_bazelci_agent", return_value="bazelci-agent"
        ), mock.patch.object(
            bazelci,
            "execute_command",
            side_effect=bazelci.subprocess.TimeoutExpired(cmd="bazelci-agent", timeout=7200),
        ):
            bazelci.upload_test_logs_from_bep("bep.json", "/tmp", monitor_flaky_tests=False)


class EolPlatforms(unittest.TestCase):
    _EOL_PLATFORM_FOR_TEST = sorted(bazelci.EOL_PLATFORMS)[0]
    _SUPPORTED_PLATFORM_FOR_TEST = sorted(set(bazelci.PLATFORMS.keys()) - bazelci.EOL_PLATFORMS)[0]
    _CONFIG_WITH_MULTIPLE_EOL_PLATFORMS = yaml.safe_load(f"""
tasks:
    foobar_eol:
        platform: {_EOL_PLATFORM_FOR_TEST}
    foobar_eol2:
        platform: {_EOL_PLATFORM_FOR_TEST}
""")
    _CONFIG_WITH_SUPPORTED_PLATFORM = yaml.safe_load(f"""
tasks:
    foobar_supported:
        platform: {_SUPPORTED_PLATFORM_FOR_TEST}
""")

    def get_steps_no_stdout(self, config):
        # Silence stdout to avoid polluting console when running unit test
        with mock.patch("sys.stdout") as _:
            steps = bazelci.print_project_pipeline(
                configs=config,
                project_name="foobar_project",
                http_config=None,
                file_config=None,
                git_repository=None,
                git_commit=None,
                monitor_flaky_tests=None,
                notify=False,
                print_shard_summary=False,
            )
            return steps

    def get_eol_warning_steps(self, steps):
        # Not every step has a label, e.g. "wait" steps or the "block" step that
        # is added to presubmit builds that modify config files.
        return [
            s
            for s in steps
            if isinstance(s, dict) and "WARNING: EOL platform detected" in s.get("label", "")
        ]

    def test_multiple_eol_platform_has_warning(self):
        steps = self.get_steps_no_stdout(self._CONFIG_WITH_MULTIPLE_EOL_PLATFORMS)
        # Find all steps that contain the EOL warning
        warning_steps = self.get_eol_warning_steps(steps)
        # There should only be 1 warning step.
        self.assertEqual(1, len(warning_steps))
        # The step should only have 1 command.
        self.assertEqual(1, len(warning_steps[0]["command"]))
        # The warning message should be specific to the EOL platform.
        warning_message = warning_steps[0]["command"][0]
        self.assertTrue(self._EOL_PLATFORM_FOR_TEST in warning_message)
        # The platform should only be listed once in the warning message.
        # The list of EOL platforms exists from the first ":" to the subsequent ".".
        platform_list = warning_message.split(":", 1)[1].split(".", 1)[0]
        platforms = [p.strip() for p in platform_list.split(",")]
        self.assertEqual([self._EOL_PLATFORM_FOR_TEST], platforms)

    def test_supported_platform_has_no_warning(self):
        steps = self.get_steps_no_stdout(self._CONFIG_WITH_SUPPORTED_PLATFORM)
        # Find all steps that contain the EOL warning (there should be none)
        warning_steps = self.get_eol_warning_steps(steps)
        # There should be no warning steps.
        self.assertEqual(0, len(warning_steps))


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