File indexing completed on 2024-05-12 05:43:36

0001 """
0002 Module containing classes for creating merge requests
0003 """
0004 
0005 # SPDX-FileCopyrightText: 2020 Jonah BrĂ¼chert <jbb@kaidan.im>
0006 #
0007 # SPDX-License-Identifier: GPL-2.0-or-later
0008 
0009 import argparse
0010 import re
0011 import os
0012 import subprocess
0013 import sys
0014 import time
0015 
0016 from typing import List, Any
0017 
0018 from git import Remote, IndexFile, PushInfo
0019 
0020 from gitlab.v4.objects import Project, ProjectMergeRequest
0021 from gitlab.exceptions import GitlabCreateError, GitlabGetError
0022 
0023 from lab.repositoryconnection import RepositoryConnection
0024 from lab.config import RepositoryConfig, Workflow
0025 from lab.utils import Utils, LogType
0026 from lab.editorinput import EditorInput
0027 
0028 
0029 def parser(
0030     subparsers: argparse._SubParsersAction,  # pylint: disable=protected-access
0031 ) -> argparse.ArgumentParser:
0032     """
0033     Subparser for merge request creation command
0034     :param subparsers: subparsers object from global parser
0035     :return: merge request creation subparser
0036     """
0037     create_parser: argparse.ArgumentParser = subparsers.add_parser(
0038         "mr", help="Create a new merge request for the current branch", aliases=["diff"]
0039     )
0040     create_parser.add_argument(
0041         "--target-branch",
0042         help="Use different target branch than master",
0043         default=Utils.get_default_branch(Utils.get_cwd_repo()),
0044     )
0045     create_parser.add_argument(
0046         "--noninteractive", help="Don't ask any interactive questions", action="store_true"
0047     )
0048     return create_parser
0049 
0050 
0051 def run(args: argparse.Namespace) -> None:
0052     """
0053     run merge request creation command
0054     :param args: parsed arguments
0055     """
0056     # To fork or not to fork
0057     fork: bool = RepositoryConfig().workflow() == Workflow.FORK
0058     creator: MergeRequestCreator = MergeRequestCreator(args.target_branch, fork)
0059     creator.check()
0060     creator.commit()
0061 
0062     if fork:
0063         creator.fork()
0064 
0065     creator.push()
0066     creator.create_mr(args.noninteractive)
0067 
0068 
0069 class MergeRequestCreator(RepositoryConnection):
0070     """
0071     Class for creating a merge request,
0072     including forking the remote repository and pushing to the fork.
0073     """
0074 
0075     # private
0076     __remote_fork: Project
0077     __target_branch: str
0078     __fork: bool
0079 
0080     def __init__(self, target_branch: str, fork: bool) -> None:
0081         RepositoryConnection.__init__(self)
0082         self.__target_branch = target_branch
0083         self.__fork = fork
0084 
0085     def check(self) -> None:
0086         """
0087         Run some sanity checks and warn the user if necessary
0088         """
0089         if self._local_repo.active_branch.name == "master":
0090             Utils.log(
0091                 LogType.WARNING,
0092                 "Creating merge requests from master is a bad idea.",
0093                 "Please check out a new a branch before creating a merge request.",
0094                 "To cancel, please press Ctrl + C.",
0095             )
0096             Utils.log(LogType.INFO, "Waiting for 10 seconds before continuing")
0097             time.sleep(10)
0098 
0099         if (
0100             not self._local_repo.active_branch.name.startswith("work/")
0101             and not self.__fork
0102             and "invent.kde.org" in self._connection.url
0103         ):
0104             Utils.log(
0105                 LogType.WARNING,
0106                 "Pushing to the upstream repository, but the branch name doesn't start with work/.",
0107             )
0108             print(
0109                 "This is not recommended on KDE infrastructure,",
0110                 "as it doesn't allow to rebase or force-push the branch.",
0111                 "To cancel, please press Ctrl + C.",
0112             )
0113             Utils.log(LogType.INFO, "Waiting for 10 seconds before continuing")
0114             time.sleep(10)
0115 
0116     def commit(self) -> None:
0117         """
0118         Determine whether there are uncommitted changes, and ask the user what to do about them
0119         """
0120 
0121         index: IndexFile = self._local_repo.index
0122         if len(index.diff("HEAD")) > 0:
0123             Utils.log(LogType.INFO, "You have staged but uncommited changes.")
0124             create_commit: bool = Utils.ask_bool("do you want to create a new commit?")
0125 
0126             if create_commit:
0127                 # We can't use self.local_repo().git.commit() here, as it would
0128                 # start the editor in the background
0129                 try:
0130                     subprocess.check_call(["git", "commit"])
0131                 except subprocess.CalledProcessError:
0132                     Utils.log(LogType.ERROR, "git exited with an error code")
0133                     sys.exit(1)
0134 
0135     def fork(self) -> None:
0136         """
0137         Try to create a fork of the remote repository.
0138         If the fork already exists, no new fork will be created.
0139         """
0140 
0141         if "fork" in self._local_repo.remotes:
0142             # Fork already exists
0143             fork_str_id: str = Utils.str_id_for_url(self._local_repo.remotes.fork.url)
0144 
0145             # Try to retrieve the remote project object, if it doesn't exist on the server,
0146             # go on with the logic to create a new fork.
0147             try:
0148                 self.__remote_fork = self._connection.projects.get(fork_str_id)
0149                 return
0150             except GitlabGetError:
0151                 pass
0152 
0153         try:
0154             self.__remote_fork = self._remote_project.forks.create({})
0155 
0156             # WORKAROUND: the return of create() is unreliable,
0157             # and sometimes doesn't allow to create merge requests,
0158             # so request a fresh project object.
0159             self.__remote_fork = self._connection.projects.get(self.__remote_fork.id)
0160 
0161             self._local_repo.create_remote("fork", url=self.__remote_fork.ssh_url_to_repo)
0162         except GitlabCreateError:
0163             Utils.log(
0164                 LogType.INFO,
0165                 "Fork exists, but no fork remote exists locally, trying to guess the url",
0166             )
0167             # Detect ssh url
0168             url = Utils.ssh_url_from_http(
0169                 self._connection.user.web_url + "/" + self._remote_project.path
0170             )
0171 
0172             self._local_repo.create_remote("fork", url=url)
0173 
0174             str_id: str = Utils.str_id_for_url(self._local_repo.remotes.fork.url)
0175             self.__remote_fork = self._connection.projects.get(str_id)
0176 
0177     def push(self) -> None:
0178         """
0179         pushes the local repository to the fork remote
0180         """
0181         remote: Remote
0182         info: PushInfo
0183         if self.__fork:
0184             remote = self._local_repo.remotes.fork
0185             info = remote.push(force=True)[0]
0186         else:
0187             remote = self._local_repo.remotes.origin
0188             info = remote.push(refspec=self._local_repo.head, force=True)[0]
0189 
0190         self._local_repo.active_branch.set_tracking_branch(info.remote_ref)
0191 
0192         if info.old_commit:
0193             print(info.local_ref, "was at", info.old_commit)
0194 
0195     def __upload_assets(self, text: str) -> str:
0196         """
0197         Scans the text for local file paths, uploads the files and returns
0198         the text modified to load the files from the uploaded urls
0199         """
0200         find_expr = re.compile(r"!\[[^\[\(]*\]\([^\[\(]*\)")
0201         extract_expr = re.compile(r"(?<=\().+?(?=\))")
0202 
0203         matches: List[Any] = find_expr.findall(text)
0204 
0205         output_text: str = text
0206 
0207         for match in matches:
0208             image = extract_expr.findall(match)[0]
0209 
0210             if not image.startswith("http"):
0211                 Utils.log(LogType.INFO, "Uploading", image)
0212 
0213                 filename: str = os.path.basename(image)
0214                 try:
0215                     uploaded_file = self._remote_project.upload(filename, filepath=image)
0216                     output_text = output_text.replace(image, uploaded_file["url"])
0217                 except FileNotFoundError:
0218                     Utils.log(LogType.WARNING, "Failed to upload image", image)
0219                     print("The file does not exist.")
0220 
0221         return output_text
0222 
0223     def create_mr(self, noninteractive: bool) -> None:
0224         """
0225         Creates a merge request with the changes from the current branch
0226         """
0227 
0228         mrs: List[ProjectMergeRequest] = self._remote_project.mergerequests.list(
0229             source_branch=self._local_repo.active_branch.name,
0230             target_branch=self.__target_branch,
0231             target_project_id=self._remote_project.id,
0232         )
0233 
0234         if len(mrs) > 0:
0235             merge_request = mrs[0]
0236             Utils.log(
0237                 LogType.INFO,
0238                 'Updating existing merge request "{}" at: {}'.format(
0239                     merge_request.title, merge_request.web_url
0240                 ),
0241             )
0242             return
0243 
0244         title: str = self._local_repo.head.commit.summary
0245         body: str = self._local_repo.head.commit.message.split("\n", 1)[1].strip()
0246 
0247         if not noninteractive:
0248             e_input = EditorInput(
0249                 placeholder_title=self._local_repo.head.commit.summary,
0250                 placeholder_body=self._local_repo.head.commit.message.split("\n", 1)[1].strip(),
0251                 extra_text="The markdown syntax for embedding images "
0252                 + "![description](/path/to/file) can be used to upload images.",
0253             )
0254             title = e_input.title
0255             body = self.__upload_assets(e_input.body)
0256 
0257         project: Project = self.__remote_fork if self.__fork else self._remote_project
0258 
0259         merge_request = project.mergerequests.create(
0260             {
0261                 "source_branch": self._local_repo.active_branch.name,
0262                 "target_branch": self.__target_branch,
0263                 "title": title,
0264                 "description": body,
0265                 "target_project_id": self._remote_project.id,
0266                 "allow_maintainer_to_push": True,
0267                 "remove_source_branch": True,
0268             }
0269         )
0270         Utils.log(LogType.INFO, "Created merge request at", merge_request.web_url)