common.py

#!/usr/bin/env python3

import requests
import sys
import time
from git import Actor, GitCommandError
from git.repo.base import Repo
from gitlab import GitlabAuthenticationError, GitlabGetError, GitlabMRRebaseError
from gitlab.v4.objects import Project
from gitlab.v4.objects import MergeRequest


manifest_file = "default.xml"
pending_states = ["created", "waiting_for_resource", "preparing", "pending", "running"]


def integration_branch_name(project_name, branch_name):
    """Get integration branch name"""
    return "integrate/" + project_name + "/" + branch_name


def get_project(gitlab, project_name):
    """Get a GitLab project by its name (including or excluding namespace)"""
    project = None
    try:
        # First try direct access, assuming name contains full path including namespace
        try:
            project = gitlab.projects.get(project_name, retry_transient_errors=True)
        except GitlabGetError:
            pass
        # If not found, try searching for project, assuming only name given
        if not project:
            for p in gitlab.projects.list(
                search=project_name, retry_transient_errors=True
            ):
                if p.name == project_name:
                    project = p
        if not project:
            sys.exit("ERROR: project '%s' not found" % project_name)
    except requests.ConnectionError:
        sys.exit("ERROR: could not connect to GitLab server")
    except GitlabAuthenticationError:
        sys.exit("ERROR: authentication failed")
    return project


def get_latest_commit(project, branch_name):
    """Get latest commit on a given project branch"""
    try:
        branch = project.branches.get(branch_name, retry_transient_errors=True)
    except GitlabGetError as e:
        sys.exit(
            "ERROR: could not get branch '%s' for project '%s': %s"
            % (branch_name, project.name, e)
        )
    if not branch:
        sys.exit(
            "ERROR: branch '%s' not found in project %s" % (branch_name, project.name)
        )
    return branch.commit


def rebase_merge_request(project, merge_request):
    """Attempt to rebase a merge request and return the updated merge request object"""
    # Rebasing takes more than one API call, see:
    # https://docs.gitlab.com/ce/api/merge_requests.html#rebase-a-merge-request
    try:
        merge_request.rebase()
    except GitlabMRRebaseError as e:
        merge_request.merge_error = "Could not rebase merge request: %s" % e
        return merge_request
    rebase_in_progress = True
    while rebase_in_progress:
        time.sleep(1)
        try:
            updated_merge_request = project.mergerequests.get(
                id=merge_request.iid,
                query_parameters={"include_rebase_in_progress": "True"},
                retry_transient_errors=True,
            )
        except GitlabGetError as e:
            merge_request.merge_error = "Could not get updated merge request: %s" % e
            return merge_request
        rebase_in_progress = updated_merge_request.rebase_in_progress
    return updated_merge_request


def crosslink_merge_requests(source_mr: MergeRequest, integration_mr: MergeRequest):
    """Insert cross-links in merge requests"""
    integration_mr.notes.create(
        {"body": "Source merge request: %s" % source_mr.web_url}
    )
    source_mr.notes.create(
        {"body": "Integration merge request: %s" % integration_mr.web_url}
    )


def wait_until_merge_status_is_set(project: Project, mr: MergeRequest):
    """Periodically query MR until GitLab has checked its merge status"""
    print("Waiting until merge status has been checked", end="", flush=True)
    unchecked_states = ["unchecked", "checking", "cannot_be_merged_recheck"]
    while mr.merge_status in unchecked_states:
        print(".", end="", flush=True)
        time.sleep(1)
        mr = project.mergerequests.get(mr.iid, retry_transient_errors=True)
    print(" -> %s" % mr.merge_status)


def list_commits(commits):
    """Create a list of commits along with the commit messages"""
    commit_list = ""
    for commit in commits:
        commit_list += "\n--\n\nCommit: %s\n\n%s" % (commit.web_url, commit.message)
    return commit_list


def commit_and_push(project: Project, repo: Repo, branch, message, name, email):
    """Commit and push to a repo branch"""
    author = Actor(name, email)
    repo.index.commit(message, author=author, committer=author)

    # Push commit
    try:
        origin = repo.remote("origin")
        origin.push(branch, force=True)
    except GitCommandError as e:
        sys.exit("ERROR: could not commit changes\n" + str(e))

    # Print commit information
    revision = repo.head.commit.hexsha
    print("Pushed new commit:")
    print(project.web_url + "/-/commit/" + revision)
    print(repo.git.show("--summary", "--decorate"))

    return revision


def get_submodule(repo: Repo, submodule_name):
    """Find a submodule in a Git repository by its name"""
    submodule = None
    for sm in repo.submodules:
        if sm.name == submodule_name:
            submodule = sm
    if submodule is None:
        sys.exit("ERROR: submodule '%s' not found" % submodule_name)
    return submodule


def extract_message_body(msg):
    """Extract message body out of a commit message"""
    # Remove headline
    msg = msg.split("\n", 1)[-1]

    # Remove all newlines, whitespaces and hyphens from the beginning
    while msg[0] in ["\n", " ", "-"]:
        msg = msg[1:]

    return msg