recipe_modules/initialization/tests/full.py

# Copyright 2026 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.

from __future__ import annotations

from collections.abc import Iterator

from recipe_engine import post_process, recipe_api, recipe_test_api

from PB.recipe_engine import result as result_pb

# Import the generated proto.
from PB.recipe_modules.pigweed.initialization.tests.full import InputProperties

from RECIPE_MODULES.pigweed.checkout import api as checkout_api

DEPS = [
    'fuchsia/builder_status',
    'pigweed/checkout',
    'pigweed/initialization',
    'recipe_engine/buildbucket',
    'recipe_engine/cv',
    'recipe_engine/json',
    'recipe_engine/properties',
    'recipe_engine/step',
]

INLINE_PROPERTIES_PROTO = """
import "recipe_modules/pigweed/checkout/options.proto";

message InputProperties {
  recipe_modules.pigweed.checkout.Options checkout_options = 1;
}
"""

PROPERTIES = InputProperties


def RunSteps(
    api: recipe_api.RecipeScriptApi,
    props: InputProperties,
) -> result_pb.RawResult | None:
    exit_early, checkout = api.initialization(props.checkout_options)
    if exit_early:
        return exit_early

    api.step.empty('success')
    return None


def GenTests(
    api: recipe_test_api.RecipeTestApi,
) -> Iterator[recipe_test_api.TestData]:
    def properties() -> recipe_test_api.TestData:
        props = InputProperties()
        props.checkout_options.CopyFrom(api.checkout.git_options())
        return api.properties(props)

    def recipe_testing_enabled():
        return api.properties(**{'$fuchsia/recipe_testing': {'enabled': True}})

    def enable_roll_tryjob_reuse():
        return api.properties(
            **{'$pigweed/roll_tryjob_reuse': {'enabled': True}}
        )

    def drop_expectations_must_be_last():
        return api.post_process(post_process.DropExpectation)

    # 1. Success path
    yield api.test(
        'basic',
        properties(),
        api.checkout.ci_test_data(),
        api.post_check(post_process.MustRun, 'success'),
        drop_expectations_must_be_last(),
    )

    # 2. ci_status exit early
    yield api.test(
        'ci_status_exit_early',
        properties(),
        api.checkout.try_test_data(),
        recipe_testing_enabled(),
        api.buildbucket.simulated_search_results(
            [api.builder_status.failure(), api.builder_status.failure()],
            step_name='checking CI status.buildbucket.search',
        ),
        api.post_check(post_process.DoesNotRun, 'success'),
        drop_expectations_must_be_last(),
    )

    # 3. split_triggers exit early (multiple branches)
    triggers = [
        {
            'gitiles': {
                'ref': 'refs/heads/foo',
                'repo': 'https://pigweed.googlesource.com/pigweed/pigweed',
                'revision': 'abc',
            },
            'id': 'foo',
        },
        {
            'gitiles': {
                'ref': 'refs/heads/bar',
                'repo': 'https://pigweed.googlesource.com/pigweed/pigweed',
                'revision': 'def',
            },
            'id': 'bar',
        },
    ]
    yield api.test(
        'split_triggers_exit_early',
        properties(),
        api.properties(
            **{'$pigweed/split_triggers': {'enabled': True}},
        ),
        api.properties(
            **{'$recipe_engine/scheduler': {'triggers': triggers}},
        ),
        api.post_check(
            post_process.MustRun,
            'split triggers.launching 2 builds.buildbucket.schedule',
        ),
        api.post_check(post_process.DoesNotRun, 'success'),
        drop_expectations_must_be_last(),
    )

    # 4. roll_tryjob_reuse exit early
    yield api.test(
        'roll_tryjob_reuse_exit_early',
        properties(),
        enable_roll_tryjob_reuse(),
        api.checkout.try_test_data(),
        api.checkout.cl_branch_parents(
            uploader='roller@pigweed-service-accounts.iam.gserviceaccount.com',
            message='Original-Reviewed-on: https://pigweed-review.googlesource.com/123456',
        ),
        api.step_data(
            'checking roll tryjob status.123456',
            api.json.output(
                {
                    'submitter': {
                        'email': 'pigweed-scoped@luci-project-accounts.iam.gserviceaccount.com'
                    },
                    'current_revision_number': 2,
                }
            ),
        ),
        api.buildbucket.simulated_search_results(
            [api.buildbucket.ci_build_message()],
            step_name='checking roll tryjob status.buildbucket.search',
        ),
        api.post_check(post_process.DoesNotRun, 'success'),
        drop_expectations_must_be_last(),
    )

    # 5. limited_cq exit early
    yield api.test(
        'limited_cq_exit_early',
        api.buildbucket.try_build(
            project='pigweed-internal',
            git_repo=checkout_api.PIGWEED_REMOTE,
        ),
        api.cv(run_mode=api.cv.DRY_RUN),
        properties(),
        api.checkout.cl_branch_parents(uploader='noreply@example.com'),
        api.post_check(post_process.DoesNotRun, 'success'),
        drop_expectations_must_be_last(),
    )