File indexing completed on 2024-12-01 08:16:20
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 + " 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)