retrigger_integrating_projects.py

#!/usr/bin/env python3
import argparse
import sys
from gitlab import Gitlab, GitlabJobRetryError

from get_integration_sources import get_integration_sources
from get_merge_requests import get_merge_requests
from retrigger_pipeline_jobs import retrigger_pipeline_jobs


def main():
    """
    Retrigger a given job in all open MRs of projects in the given group with a target
    branch that is configured for automatic integration into the given manifest project
    and branch.
    """
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--gitlab-url",
        help="""URL to the GitLab instance""",
        dest="gitlab_url",
        required=True,
    )
    parser.add_argument(
        "--token",
        help="""GitLab REST API private access token""",
        dest="token",
        required=True,
    )
    parser.add_argument(
        "--manifest-project",
        help="""name of the GitLab manifest project""",
        dest="manifest_project",
        required=True,
    )
    parser.add_argument(
        "--manifest-branch",
        help="""target integration branch""",
        dest="manifest_branch",
        required=True,
    )
    parser.add_argument(
        "--group",
        help="""group path or id to limit search scope to""",
        dest="group",
        required=True,
    )
    parser.add_argument(
        "--job",
        help="""job to retrigger""",
        dest="job",
        required=False,
        default="check",
    )

    args, _ = parser.parse_known_args()

    gitlab = Gitlab(args.gitlab_url, private_token=args.token)
    group = gitlab.groups.get(args.group, retry_transient_errors=True)

    integration_sources = get_integration_sources(
        args.manifest_project, args.manifest_branch, group
    )

    failed = 0

    for source in integration_sources:

        project = gitlab.projects.get(source["project"], retry_transient_errors=True)

        mrs = get_merge_requests(
            project,
            state="opened",
            target_branch=source["branch"],
        )

        for mr in mrs:
            # Get pipeline
            if not mr.pipeline:
                print("No pipeline in %s" % mr.web_url)
                continue
            pipeline = project.pipelines.get(
                mr.pipeline.get("id"),
                retry_transient_errors=True,
            )
            try:
                states = ["success", "running"]
                jobs = retrigger_pipeline_jobs(
                    project,
                    pipeline,
                    args.job,
                    states,
                    include_children=True,
                )
                if not jobs:
                    print(
                        "Could not find any jobs named '%s' with states %s in %s"
                        % (args.job, states, pipeline.web_url)
                    )
            except GitlabJobRetryError as e:
                print(
                    "ERROR: Could not retrigger job '%s' in %s: %s"
                    % (args.job, pipeline.web_url, e)
                )
                failed = failed + 1
                continue

    if failed > 0:
        sys.exit(1)


if __name__ == "__main__":
    main()