recipe_modules/abandon_old_changes/api.py

# Copyright 2024 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.
"""Abandon old changes owned by the current service account."""

from __future__ import annotations

from recipe_engine import recipe_api

from PB.recipe_modules.pigweed.abandon_old_changes import (
    properties as properties_pb,
)


class AbandonOldChangesApi(recipe_api.RecipeApi):
    """Abandon old changes owned by the current service account."""

    def __init__(self, props: properties_pb.InputProperties, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._hosts = props.hosts
        self._age_days = props.age_days or 30
        self._max_to_abandon = props.max_to_abandon or 10

    def __call__(
        self,
        *,
        host: str | None = None,
        age_days: int | None = None,
        continue_on_failure: bool = True,
        max_to_abandon: int | None = None,
    ) -> None:
        if age_days is None:
            age_days = self._age_days
        if max_to_abandon is None:
            max_to_abandon = self._max_to_abandon

        hosts = []
        if host:
            hosts.append(host)
        hosts.extend(self._hosts)

        if not hosts:
            return  # pragma: no cover

        with self.m.step.nest('abandon old changes'):
            for host in hosts:
                with self.m.step.nest(self.m.gerrit.host_basename(host)):
                    try:
                        self._abandon_old_changes(
                            host=self.m.gerrit.host_from_remote_url(host),
                            age_days=age_days,
                            max_to_abandon=max_to_abandon,
                        )
                        self.m.step.empty('success')

                    except self.m.step.StepFailure:
                        self.m.step.empty('failure')
                        if not continue_on_failure:
                            raise  # pragma: no cover

    def _abandon_old_changes(
        self,
        *,
        host: str,
        age_days: int,
        max_to_abandon: int,
    ) -> None:
        query_string = (
            f'is:open '
            f'owner:{self.m.buildbucket.swarming_task_service_account} '
            f'age:{age_days}d '
        )

        changes = self.m.gerrit.change_query(
            name='get old changes',
            query_string=query_string,
            host=host,
            max_attempts=1,
            timeout=30,
            test_data=self.m.json.test_api.output(
                [
                    {'_number': 1001},
                    {'_number': 1002},
                ],
            ),
        ).json.output

        self.m.random.shuffle(changes)

        futures = []
        for change in changes[:max_to_abandon]:
            futures.append(
                self.m.futures.spawn(
                    self.m.gerrit.abandon,
                    f'abandon {change["_number"]}',
                    change_id=change['_number'],
                    message='Abandoning orphaned roll change.',
                    host=host,
                ),
            )

        self.m.futures.wait(futures)