community.general/plugins/modules/gitlab_merge_request.py

#!/usr/bin/python

# Copyright (c) 2023, Ondrej Zvara (ozvara1@gmail.com)
# Based on code:
# Copyright (c) 2021, Lennert Mertens (lennert@nubera.be)
# Copyright (c) 2021, Werner Dijkerman (ikben@werner-dijkerman.nl)
# Copyright (c) 2015, Werner Dijkerman (ikben@werner-dijkerman.nl)
# Copyright (c) 2019, Guillaume Martinez (lunik@tiwabbit.fr)
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import annotations

DOCUMENTATION = r"""
module: gitlab_merge_request
short_description: Create, update, or delete GitLab merge requests
version_added: 7.1.0
description:
  - Creates a merge request if it does not exist.
  - When a single merge request does exist, it is updated if the provided parameters are different.
  - When a single merge request does exist and O(state=absent), the merge request is deleted.
  - When multiple merge requests are detected, the task fails.
  - Existing merge requests are matched based on O(title), O(source_branch), O(target_branch), and O(state_filter) filters.
author:
  - zvaraondrej (@zvaraondrej)
requirements:
  - python-gitlab >= 2.3.0
extends_documentation_fragment:
  - community.general.auth_basic
  - community.general.gitlab
  - community.general.attributes

attributes:
  check_mode:
    support: full
  diff_mode:
    support: none

options:
  state:
    description:
      - Create or delete merge request.
    default: present
    type: str
    choices: ["present", "absent"]
  project:
    description:
      - The path or name of the project.
    required: true
    type: str
  source_branch:
    description:
      - Merge request's source branch.
      - Ignored while updating existing merge request.
    required: true
    type: str
  target_branch:
    description:
      - Merge request's target branch.
    required: true
    type: str
  title:
    description:
      - A title for the merge request.
    type: str
    required: true
  description:
    description:
      - A description for the merge request.
      - Gets overridden by a content of file specified at O(description_path), if found.
    type: str
  description_path:
    description:
      - A path of file containing merge request's description.
      - Accepts MarkDown formatted files.
    type: path
  labels:
    description:
      - Comma separated list of label names.
    type: str
    default: ""
  remove_source_branch:
    description:
      - Flag indicating if a merge request should remove the source branch when merging.
    type: bool
    default: false
  state_filter:
    description:
      - Filter specifying state of merge requests while searching.
    type: str
    choices: ["opened", "closed", "locked", "merged"]
    default: opened
  assignee_ids:
    description:
      - Comma separated list of assignees usernames omitting V(@) character.
      - Set to empty string to unassign all assignees.
    type: str
  reviewer_ids:
    description:
      - Comma separated list of reviewers usernames omitting V(@) character.
      - Set to empty string to unassign all reviewers.
    type: str
"""


EXAMPLES = r"""
- name: Create Merge Request from branch1 to branch2
  community.general.gitlab_merge_request:
    api_url: https://gitlab.com
    api_token: secret_access_token
    project: "group1/project1"
    source_branch: branch1
    target_branch: branch2
    title: "Ansible demo MR"
    description: "Demo MR description"
    labels: "Ansible,Demo"
    state_filter: "opened"
    remove_source_branch: true
    state: present

- name: Delete Merge Request from branch1 to branch2
  community.general.gitlab_merge_request:
    api_url: https://gitlab.com
    api_token: secret_access_token
    project: "group1/project1"
    source_branch: branch1
    target_branch: branch2
    title: "Ansible demo MR"
    state_filter: "opened"
    state: absent
"""

RETURN = r"""
msg:
  description: Success or failure message.
  returned: always
  type: str
  sample: "Success"

mr:
  description: API object.
  returned: success
  type: dict
"""

from ansible.module_utils.api import basic_auth_argument_spec
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.text.converters import to_text

from ansible_collections.community.general.plugins.module_utils.gitlab import (
    auth_argument_spec,
    find_project,
    gitlab,
    gitlab_authentication,
)
from ansible_collections.community.general.plugins.module_utils.version import LooseVersion


class GitlabMergeRequest:
    def __init__(self, module, project, gitlab_instance):
        self._gitlab = gitlab_instance
        self._module = module
        self.project = project

    """
    @param branch Name of the branch
    """

    def get_branch(self, branch):
        try:
            return self.project.branches.get(branch)
        except gitlab.exceptions.GitlabGetError as e:
            self._module.fail_json(msg=f"Failed to get the branch: {e}")

    """
    @param title Title of the Merge Request
    @param source_branch Merge Request's source branch
    @param target_branch Merge Request's target branch
    @param state_filter Merge Request's state to filter on
    """

    def get_mr(self, title, source_branch, target_branch, state_filter):
        mrs = []
        try:
            mrs = self.project.mergerequests.list(
                search=title, source_branch=source_branch, target_branch=target_branch, state=state_filter
            )
        except gitlab.exceptions.GitlabGetError as e:
            self._module.fail_json(msg=f"Failed to list the Merge Request: {e}")

        if len(mrs) > 1:
            self._module.fail_json(msg="Multiple Merge Requests matched search criteria.")
        if len(mrs) == 1:
            try:
                return self.project.mergerequests.get(id=mrs[0].iid)
            except gitlab.exceptions.GitlabGetError as e:
                self._module.fail_json(msg=f"Failed to get the Merge Request: {e}")

    """
    @param username Name of the user
    """

    def get_user(self, username):
        users = []
        try:
            users = [user for user in self.project.users.list(username=username, all=True) if user.username == username]
        except gitlab.exceptions.GitlabGetError as e:
            self._module.fail_json(msg=f"Failed to list the users: {e}")

        if len(users) > 1:
            self._module.fail_json(msg="Multiple Users matched search criteria.")
        elif len(users) < 1:
            self._module.fail_json(msg="No User matched search criteria.")
        else:
            return users[0]

    """
    @param users List of usernames
    """

    def get_user_ids(self, users):
        return [self.get_user(user).id for user in users]

    """
    @param options Options of the Merge Request
    """

    def create_mr(self, options):
        if self._module.check_mode:
            self._module.exit_json(changed=True, msg=f"Successfully created the Merge Request {options['title']}")

        try:
            return self.project.mergerequests.create(options)
        except gitlab.exceptions.GitlabCreateError as e:
            self._module.fail_json(msg=f"Failed to create Merge Request: {e}")

    """
    @param mr Merge Request object to delete
    """

    def delete_mr(self, mr):
        if self._module.check_mode:
            self._module.exit_json(changed=True, msg=f"Successfully deleted the Merge Request {mr['title']}")

        try:
            return mr.delete()
        except gitlab.exceptions.GitlabDeleteError as e:
            self._module.fail_json(msg=f"Failed to delete Merge Request: {e}")

    """
    @param mr Merge Request object to update
    """

    def update_mr(self, mr, options):
        if self._module.check_mode:
            self._module.exit_json(changed=True, msg=f"Successfully updated the Merge Request {mr['title']}")

        try:
            return self.project.mergerequests.update(mr.iid, options)
        except gitlab.exceptions.GitlabUpdateError as e:
            self._module.fail_json(msg=f"Failed to update Merge Request: {e}")

    """
    @param mr Merge Request object to evaluate
    @param options New options to update MR with
    """

    def mr_has_changed(self, mr, options):
        for key, value in options.items():
            if value is not None:
                # see https://gitlab.com/gitlab-org/gitlab-foss/-/issues/27355
                if key == "remove_source_branch":
                    key = "force_remove_source_branch"

                if key == "assignee_ids":
                    if value != sorted([user["id"] for user in mr.assignees]):
                        return True

                elif key == "reviewer_ids":
                    if value != sorted([user["id"] for user in mr.reviewers]):
                        return True

                elif key == "labels":
                    if value != sorted(getattr(mr, key)):
                        return True

                elif getattr(mr, key) != value:
                    return True

        return False


def main():
    argument_spec = basic_auth_argument_spec()
    argument_spec.update(auth_argument_spec())
    argument_spec.update(
        project=dict(type="str", required=True),
        source_branch=dict(type="str", required=True),
        target_branch=dict(type="str", required=True),
        title=dict(type="str", required=True),
        description=dict(type="str"),
        labels=dict(type="str", default=""),
        description_path=dict(type="path"),
        remove_source_branch=dict(type="bool", default=False),
        state_filter=dict(type="str", default="opened", choices=["opened", "closed", "locked", "merged"]),
        assignee_ids=dict(type="str"),
        reviewer_ids=dict(type="str"),
        state=dict(type="str", default="present", choices=["absent", "present"]),
    )

    module = AnsibleModule(
        argument_spec=argument_spec,
        mutually_exclusive=[
            ["api_username", "api_token"],
            ["api_username", "api_oauth_token"],
            ["api_username", "api_job_token"],
            ["api_token", "api_oauth_token"],
            ["api_token", "api_job_token"],
            ["description", "description_path"],
        ],
        required_together=[
            ["api_username", "api_password"],
        ],
        required_one_of=[["api_username", "api_token", "api_oauth_token", "api_job_token"]],
        required_if=[
            ["state", "present", ["source_branch", "target_branch", "title"], True],
            ["state", "absent", ["source_branch", "target_branch", "title"], True],
        ],
        supports_check_mode=True,
    )

    # check prerequisites and connect to gitlab server
    gitlab_instance = gitlab_authentication(module)

    project = module.params["project"]
    source_branch = module.params["source_branch"]
    target_branch = module.params["target_branch"]
    title = module.params["title"]
    description = module.params["description"]
    labels = module.params["labels"]
    description_path = module.params["description_path"]
    remove_source_branch = module.params["remove_source_branch"]
    state_filter = module.params["state_filter"]
    assignee_ids = module.params["assignee_ids"]
    reviewer_ids = module.params["reviewer_ids"]
    state = module.params["state"]

    gitlab_version = gitlab.__version__
    if LooseVersion(gitlab_version) < LooseVersion("2.3.0"):
        module.fail_json(
            msg=f"community.general.gitlab_merge_request requires python-gitlab Python module >= 2.3.0 (installed version: [{gitlab_version}])."
            " Please upgrade python-gitlab to version 2.3.0 or above."
        )

    this_project = find_project(gitlab_instance, project)
    if this_project is None:
        module.fail_json(msg=f"Failed to get the project: {project}")

    this_gitlab = GitlabMergeRequest(module=module, project=this_project, gitlab_instance=gitlab_instance)

    r_source_branch = this_gitlab.get_branch(source_branch)
    if not r_source_branch:
        module.fail_json(msg=f"Source branch {r_source_branch} not exist.")

    r_target_branch = this_gitlab.get_branch(target_branch)
    if not r_target_branch:
        module.fail_json(msg=f"Destination branch {r_target_branch} not exist.")

    this_mr = this_gitlab.get_mr(title, source_branch, target_branch, state_filter)

    if state == "present":
        if description_path:
            try:
                with open(description_path, "rb") as f:
                    description = to_text(f.read(), errors="surrogate_or_strict")
            except OSError as e:
                module.fail_json(msg=f"Cannot open {description_path}: {e}")

        # sorting necessary in order to properly detect changes, as we don't want to get false positive
        # results due to differences in ids ordering; see `mr_has_changed()`
        assignee_ids = sorted(this_gitlab.get_user_ids(assignee_ids.split(","))) if assignee_ids else []
        reviewer_ids = sorted(this_gitlab.get_user_ids(reviewer_ids.split(","))) if reviewer_ids else []
        labels = sorted(labels.split(",")) if labels else []

        options = {
            "target_branch": target_branch,
            "title": title,
            "description": description,
            "labels": labels,
            "remove_source_branch": remove_source_branch,
            "reviewer_ids": reviewer_ids,
            "assignee_ids": assignee_ids,
        }

        if not this_mr:
            options["source_branch"] = source_branch

            mr = this_gitlab.create_mr(options)
            module.exit_json(
                changed=True,
                msg=f"Created the Merge Request {title} from branch {source_branch} to branch {target_branch}.",
                mr=mr.asdict(),
            )
        else:
            if this_gitlab.mr_has_changed(this_mr, options):
                mr = this_gitlab.update_mr(this_mr, options)
                module.exit_json(
                    changed=True,
                    msg=f"Merge Request {title} from branch {source_branch} to branch {target_branch} updated.",
                    mr=mr,
                )
            else:
                module.exit_json(
                    changed=False,
                    msg=f"Merge Request {title} from branch {source_branch} to branch {target_branch} already exist",
                    mr=this_mr.asdict(),
                )
    elif this_mr and state == "absent":
        mr = this_gitlab.delete_mr(this_mr)
        module.exit_json(
            changed=True,
            msg=f"Merge Request {title} from branch {source_branch} to branch {target_branch} deleted.",
            mr=mr,
        )
    else:
        module.exit_json(changed=False, msg="No changes are needed.", mr=this_mr.asdict())


if __name__ == "__main__":
    main()