path: root/gitlab/v4/objects/merge_requests.py

from gitlab import cli, types
from gitlab import exceptions as exc
from gitlab.base import *  # noqa
from gitlab.mixins import *  # noqa
from .commits import ProjectCommit, ProjectCommitManager
from .issues import ProjectIssue, ProjectIssueManager
from .merge_request_approvals import (
    ProjectMergeRequestApprovalManager,
    ProjectMergeRequestApprovalRuleManager,
)
from .award_emojis import ProjectMergeRequestAwardEmojiManager
from .discussions import ProjectMergeRequestDiscussionManager
from .notes import ProjectMergeRequestNoteManager
from .events import (
    ProjectMergeRequestResourceLabelEventManager,
    ProjectMergeRequestResourceMilestoneEventManager,
)


__all__ = [
    "MergeRequest",
    "MergeRequestManager",
    "GroupMergeRequest",
    "GroupMergeRequestManager",
    "ProjectMergeRequest",
    "ProjectMergeRequestManager",
    "ProjectMergeRequestDiff",
    "ProjectMergeRequestDiffManager",
]


class MergeRequest(RESTObject):
    pass


class MergeRequestManager(ListMixin, RESTManager):
    _path = "/merge_requests"
    _obj_cls = MergeRequest
    _from_parent_attrs = {"group_id": "id"}
    _list_filters = (
        "state",
        "order_by",
        "sort",
        "milestone",
        "view",
        "labels",
        "created_after",
        "created_before",
        "updated_after",
        "updated_before",
        "scope",
        "author_id",
        "assignee_id",
        "my_reaction_emoji",
        "source_branch",
        "target_branch",
        "search",
        "wip",
    )
    _types = {"labels": types.ListAttribute}


class GroupMergeRequest(RESTObject):
    pass


class GroupMergeRequestManager(ListMixin, RESTManager):
    _path = "/groups/%(group_id)s/merge_requests"
    _obj_cls = GroupMergeRequest
    _from_parent_attrs = {"group_id": "id"}
    _list_filters = (
        "state",
        "order_by",
        "sort",
        "milestone",
        "view",
        "labels",
        "created_after",
        "created_before",
        "updated_after",
        "updated_before",
        "scope",
        "author_id",
        "assignee_id",
        "my_reaction_emoji",
        "source_branch",
        "target_branch",
        "search",
        "wip",
    )
    _types = {"labels": types.ListAttribute}


class ProjectMergeRequest(
    SubscribableMixin,
    TodoMixin,
    TimeTrackingMixin,
    ParticipantsMixin,
    SaveMixin,
    ObjectDeleteMixin,
    RESTObject,
):
    _id_attr = "iid"

    _managers = (
        ("approvals", "ProjectMergeRequestApprovalManager"),
        ("approval_rules", "ProjectMergeRequestApprovalRuleManager"),
        ("awardemojis", "ProjectMergeRequestAwardEmojiManager"),
        ("diffs", "ProjectMergeRequestDiffManager"),
        ("discussions", "ProjectMergeRequestDiscussionManager"),
        ("notes", "ProjectMergeRequestNoteManager"),
        ("resourcelabelevents", "ProjectMergeRequestResourceLabelEventManager"),
        ("resourcemilestoneevents", "ProjectMergeRequestResourceMilestoneEventManager"),
    )

    @cli.register_custom_action("ProjectMergeRequest")
    @exc.on_http_error(exc.GitlabMROnBuildSuccessError)
    def cancel_merge_when_pipeline_succeeds(self, **kwargs):
        """Cancel merge when the pipeline succeeds.

        Args:
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabMROnBuildSuccessError: If the server could not handle the
                request
        """

        path = "%s/%s/cancel_merge_when_pipeline_succeeds" % (
            self.manager.path,
            self.get_id(),
        )
        server_data = self.manager.gitlab.http_put(path, **kwargs)
        self._update_attrs(server_data)

    @cli.register_custom_action("ProjectMergeRequest")
    @exc.on_http_error(exc.GitlabListError)
    def closes_issues(self, **kwargs):
        """List issues that will close on merge."

        Args:
            all (bool): If True, return all the items, without pagination
            per_page (int): Number of items to retrieve per request
            page (int): ID of the page to return (starts with page 1)
            as_list (bool): If set to False and no pagination option is
                defined, return a generator instead of a list
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabListError: If the list could not be retrieved

        Returns:
            RESTObjectList: List of issues
        """
        path = "%s/%s/closes_issues" % (self.manager.path, self.get_id())
        data_list = self.manager.gitlab.http_list(path, as_list=False, **kwargs)
        manager = ProjectIssueManager(self.manager.gitlab, parent=self.manager._parent)
        return RESTObjectList(manager, ProjectIssue, data_list)

    @cli.register_custom_action("ProjectMergeRequest")
    @exc.on_http_error(exc.GitlabListError)
    def commits(self, **kwargs):
        """List the merge request commits.

        Args:
            all (bool): If True, return all the items, without pagination
            per_page (int): Number of items to retrieve per request
            page (int): ID of the page to return (starts with page 1)
            as_list (bool): If set to False and no pagination option is
                defined, return a generator instead of a list
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabListError: If the list could not be retrieved

        Returns:
            RESTObjectList: The list of commits
        """

        path = "%s/%s/commits" % (self.manager.path, self.get_id())
        data_list = self.manager.gitlab.http_list(path, as_list=False, **kwargs)
        manager = ProjectCommitManager(self.manager.gitlab, parent=self.manager._parent)
        return RESTObjectList(manager, ProjectCommit, data_list)

    @cli.register_custom_action("ProjectMergeRequest")
    @exc.on_http_error(exc.GitlabListError)
    def changes(self, **kwargs):
        """List the merge request changes.

        Args:
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabListError: If the list could not be retrieved

        Returns:
            RESTObjectList: List of changes
        """
        path = "%s/%s/changes" % (self.manager.path, self.get_id())
        return self.manager.gitlab.http_get(path, **kwargs)

    @cli.register_custom_action("ProjectMergeRequest")
    @exc.on_http_error(exc.GitlabListError)
    def pipelines(self, **kwargs):
        """List the merge request pipelines.

        Args:
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabListError: If the list could not be retrieved

        Returns:
            RESTObjectList: List of changes
        """

        path = "%s/%s/pipelines" % (self.manager.path, self.get_id())
        return self.manager.gitlab.http_get(path, **kwargs)

    @cli.register_custom_action("ProjectMergeRequest", tuple(), ("sha"))
    @exc.on_http_error(exc.GitlabMRApprovalError)
    def approve(self, sha=None, **kwargs):
        """Approve the merge request.

        Args:
            sha (str): Head SHA of MR
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabMRApprovalError: If the approval failed
        """
        path = "%s/%s/approve" % (self.manager.path, self.get_id())
        data = {}
        if sha:
            data["sha"] = sha

        server_data = self.manager.gitlab.http_post(path, post_data=data, **kwargs)
        self._update_attrs(server_data)

    @cli.register_custom_action("ProjectMergeRequest")
    @exc.on_http_error(exc.GitlabMRApprovalError)
    def unapprove(self, **kwargs):
        """Unapprove the merge request.

        Args:
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabMRApprovalError: If the unapproval failed
        """
        path = "%s/%s/unapprove" % (self.manager.path, self.get_id())
        data = {}

        server_data = self.manager.gitlab.http_post(path, post_data=data, **kwargs)
        self._update_attrs(server_data)

    @cli.register_custom_action("ProjectMergeRequest")
    @exc.on_http_error(exc.GitlabMRRebaseError)
    def rebase(self, **kwargs):
        """Attempt to rebase the source branch onto the target branch

        Args:
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabMRRebaseError: If rebasing failed
        """
        path = "%s/%s/rebase" % (self.manager.path, self.get_id())
        data = {}
        return self.manager.gitlab.http_put(path, post_data=data, **kwargs)

    @cli.register_custom_action(
        "ProjectMergeRequest",
        tuple(),
        (
            "merge_commit_message",
            "should_remove_source_branch",
            "merge_when_pipeline_succeeds",
        ),
    )
    @exc.on_http_error(exc.GitlabMRClosedError)
    def merge(
        self,
        merge_commit_message=None,
        should_remove_source_branch=False,
        merge_when_pipeline_succeeds=False,
        **kwargs
    ):
        """Accept the merge request.

        Args:
            merge_commit_message (bool): Commit message
            should_remove_source_branch (bool): If True, removes the source
                                                branch
            merge_when_pipeline_succeeds (bool): Wait for the build to succeed,
                                                 then merge
            **kwargs: Extra options to send to the server (e.g. sudo)

        Raises:
            GitlabAuthenticationError: If authentication is not correct
            GitlabMRClosedError: If the merge failed
        """
        path = "%s/%s/merge" % (self.manager.path, self.get_id())
        data = {}
        if merge_commit_message:
            data["merge_commit_message"] = merge_commit_message
        if should_remove_source_branch:
            data["should_remove_source_branch"] = True
        if merge_when_pipeline_succeeds:
            data["merge_when_pipeline_succeeds"] = True

        server_data = self.manager.gitlab.http_put(path, query_data=data, **kwargs)
        self._update_attrs(server_data)


class ProjectMergeRequestManager(CRUDMixin, RESTManager):
    _path = "/projects/%(project_id)s/merge_requests"
    _obj_cls = ProjectMergeRequest
    _from_parent_attrs = {"project_id": "id"}
    _create_attrs = (
        ("source_branch", "target_branch", "title"),
        (
            "assignee_id",
            "description",
            "target_project_id",
            "labels",
            "milestone_id",
            "remove_source_branch",
            "allow_maintainer_to_push",
            "squash",
        ),
    )
    _update_attrs = (
        tuple(),
        (
            "target_branch",
            "assignee_id",
            "title",
            "description",
            "state_event",
            "labels",
            "milestone_id",
            "remove_source_branch",
            "discussion_locked",
            "allow_maintainer_to_push",
            "squash",
        ),
    )
    _list_filters = (
        "state",
        "order_by",
        "sort",
        "milestone",
        "view",
        "labels",
        "created_after",
        "created_before",
        "updated_after",
        "updated_before",
        "scope",
        "author_id",
        "assignee_id",
        "my_reaction_emoji",
        "source_branch",
        "target_branch",
        "search",
        "wip",
    )
    _types = {"labels": types.ListAttribute}


class ProjectMergeRequestDiff(RESTObject):
    pass


class ProjectMergeRequestDiffManager(RetrieveMixin, RESTManager):
    _path = "/projects/%(project_id)s/merge_requests/%(mr_iid)s/versions"
    _obj_cls = ProjectMergeRequestDiff
    _from_parent_attrs = {"project_id": "project_id", "mr_iid": "iid"}