recipe_modules/split_triggers/tests/full.py

# Copyright 2025 The Pigweed Authors
#
# 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
#
#     https://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.
"""Tests for split_triggers module."""

from __future__ import annotations

import urllib.parse
from collections.abc import Iterator

from recipe_engine import post_process, recipe_api, recipe_test_api

from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb
from PB.recipe_engine import result as result_pb

DEPS = [
    'pigweed/split_triggers',
    'recipe_engine/buildbucket',
    'recipe_engine/properties',
    'recipe_engine/step',
]


def _expand_commit(commit: str) -> str:
    return (commit * 40)[0:40]


def RunSteps(api: recipe_api.RecipeScriptApi) -> result_pb.RawResult | None:
    if result := api.split_triggers():
        return result

    api.step.empty('continuing')


def GenTests(
    api: recipe_test_api.RecipeTestApi,
) -> Iterator[recipe_test_api.TestData]:
    pigweed = 'https://pigweed.googlesource.com/pigweed/pigweed'

    def commit(
        id: str,
        *,
        url: str = pigweed,
        branch: str = 'main',
    ) -> common_pb.GitilesCommit:
        parsed = urllib.parse.urlparse(url)
        return common_pb.GitilesCommit(
            host=parsed.hostname,
            project=parsed.path.lstrip('/'),
            ref=f'refs/heads/{branch}',
            id=_expand_commit(id),
        )

    def properties(
        *commits: common_pb.GitilesCommit,
        enabled: bool = True,
        bb_trigger: bool = False,
    ) -> recipe_test_api.TestData:
        result = api.properties(
            **{'$pigweed/split_triggers': {'enabled': enabled}}
        )

        triggers = []
        for c in commits:
            remote = f'https://{c.host}/{c.project}'

            triggers.append(
                {
                    'gitiles': {
                        'ref': c.ref,
                        'repo': remote,
                        'revision': c.id,
                    },
                    'id': f'{remote}/+/{c.ref}@{c.id}',
                    'title': c.id,
                    'url': f'{remote}/+/{c.id}',
                },
            )

        if bb_trigger:
            triggers.append({'id': '12345', 'buildbucket': {}})

        result += api.properties(
            **{'$recipe_engine/scheduler': {'triggers': triggers}},
        )

        return result

    def assert_non_gitiles_trigger() -> recipe_test_api.TestData:
        return api.post_check(
            post_process.MustRun,
            'split triggers.processing triggers.one non-gitiles trigger',
        )

    def assert_trigger(
        commit_id: str,
        *,
        repo: str = pigweed,
        branch: str = 'main',
    ) -> recipe_test_api.TestData:
        commit_id = _expand_commit(commit_id)
        return api.post_check(
            post_process.MustRun,
            f'split triggers.processing triggers.one trigger for {repo} '
            f'refs/heads/{branch} '
            f'{commit_id}',
        )

    def assert_selected(
        commit_id: str,
        *,
        repo: str = pigweed,
        branch: str = 'main',
    ) -> recipe_test_api.TestData:
        commit_id = _expand_commit(commit_id)
        return api.post_check(
            post_process.MustRun,
            f'split triggers.filtering.selected {repo} '
            f'refs/heads/{branch} {commit_id}',
        )

    def assert_no_filtering() -> recipe_test_api.TestData:
        return api.post_check(
            post_process.DoesNotRun, 'split triggers.filtering'
        )

    def assert_no_split_triggers() -> recipe_test_api.TestData:
        return api.post_check(post_process.DoesNotRun, 'split triggers')

    def assert_no_branches() -> recipe_test_api.TestData:
        return api.post_check(
            post_process.MustRun, 'split triggers.no branches'
        )

    def assert_only_one_branch() -> recipe_test_api.TestData:
        return api.post_check(
            post_process.MustRun, 'split triggers.only one branch'
        )

    def assert_launching(n: int) -> recipe_test_api.TestData:
        return api.post_check(
            post_process.MustRun, f'split triggers.launching {n} builds'
        )

    def assert_continuing() -> recipe_test_api.TestData:
        return api.post_check(post_process.MustRun, 'continuing')

    def drop_expectations_must_be_last() -> recipe_test_api.TestData:
        # No need for expectation files, everything of note here is tested by
        # assertions. This must be the last thing added to the test.
        return api.post_process(post_process.DropExpectation)

    yield api.test(
        'manual',
        properties(),
        assert_no_branches(),
        assert_no_filtering(),
        assert_continuing(),
        drop_expectations_must_be_last(),
    )

    yield api.test(
        'single',
        properties(commit('abc')),
        assert_trigger(commit_id='abc'),
        assert_only_one_branch(),
        assert_no_filtering(),
        assert_continuing(),
        drop_expectations_must_be_last(),
    )

    yield api.test(
        'double',
        properties(
            commit('abc'),
            commit('def'),
        ),
        assert_trigger(commit_id='def'),
        assert_only_one_branch(),
        assert_no_filtering(),
        assert_continuing(),
        drop_expectations_must_be_last(),
    )

    yield api.test(
        'with-bb',
        properties(commit('abc'), bb_trigger=True),
        assert_trigger(commit_id='abc'),
        assert_only_one_branch(),
        assert_no_filtering(),
        assert_non_gitiles_trigger(),
        assert_continuing(),
        drop_expectations_must_be_last(),
    )

    yield api.test(
        'bb-only',
        properties(bb_trigger=True),
        assert_no_branches(),
        assert_no_filtering(),
        assert_non_gitiles_trigger(),
        assert_continuing(),
        drop_expectations_must_be_last(),
    )

    yield api.test(
        'two-branches',
        properties(
            commit('abc', branch='foo'),
            commit('def', branch='bar'),
        ),
        assert_trigger(commit_id='abc', branch='foo'),
        assert_trigger(commit_id='def', branch='bar'),
        assert_selected(commit_id='abc', branch='foo'),
        assert_selected(commit_id='def', branch='bar'),
        assert_launching(2),
        drop_expectations_must_be_last(),
    )

    yield api.test(
        'two-branches-multiple',
        properties(
            commit('abc1', branch='foo'),
            commit('abc2', branch='foo'),
            commit('abc3', branch='foo'),
            commit('abc4', branch='foo'),
            commit('abc5', branch='foo'),
            commit('def', branch='bar'),
            enabled=False,
        ),
        assert_trigger(commit_id='abc1', branch='foo'),
        assert_trigger(commit_id='abc2', branch='foo'),
        assert_trigger(commit_id='abc3', branch='foo'),
        assert_trigger(commit_id='abc4', branch='foo'),
        assert_trigger(commit_id='abc5', branch='foo'),
        assert_trigger(commit_id='def', branch='bar'),
        assert_selected(commit_id='abc5', branch='foo'),
        assert_selected(commit_id='def', branch='bar'),
        assert_launching(2),
        drop_expectations_must_be_last(),
    )

    yield api.test(
        'tryjob',
        api.buildbucket.try_build(project='pigweed'),
        properties(
            commit('abc', branch='foo'),
            commit('def', branch='bar'),
        ),
        assert_no_split_triggers(),
        assert_continuing(),
        drop_expectations_must_be_last(),
    )

    yield api.test(
        'recursion-prevention',
        properties(
            commit('abc', branch='foo'),
            commit('def', branch='bar'),
        ),
        api.buildbucket.ci_build(
            tags=api.buildbucket.tags(user_agent='split_triggers'),
        ),
        assert_no_split_triggers(),
        assert_continuing(),
        drop_expectations_must_be_last(),
    )