feat: prevent overriding commits in a branch that were not made by rcmt (#439)

Author: wndhydrnt

rcmt/git.py

@@ -15,6 +15,11 @@
 log: structlog.stdlib.BoundLogger = structlog.get_logger(package="git")
 
 
+class BranchModifiedError(RuntimeError):
+    def __init__(self, checksums: list[str]):
+        self.checksums = checksums
+
+
 class Git:
     def __init__(
         self,
@@ -38,6 +43,22 @@ def commit_changes(self, repo_dir: str, msg: str):
         git_repo.git.add(all=True)
         git_repo.index.commit(msg)
 
+    def _detect_modified_branch(
+        self,
+        merge_base: str,
+        repo: git.Repo,
+    ) -> None:
+        foreign_commits: list[str] = []
+        for commit in repo.iter_commits():
+            if commit.hexsha == merge_base:
+                break
+
+            if commit.author.email != self.user_email:
+                foreign_commits.append(commit.hexsha)
+
+        if len(foreign_commits) > 0:
+            raise BranchModifiedError(foreign_commits)
+
     @staticmethod
     def has_changes_origin(branch: str, repo_dir: str) -> bool:
         git_repo = git.Repo(path=repo_dir)
@@ -54,7 +75,9 @@ def has_changes_local(repo_dir: str) -> bool:
         git_repo = git.Repo(path=repo_dir)
         return len(git_repo.index.diff(None)) > 0 or len(git_repo.untracked_files) > 0
 
-    def prepare(self, repo: source.Repository, iteration: int = 0) -> Tuple[str, bool]:
+    def prepare(
+        self, repo: source.Repository, force_rebase: bool, iteration: int = 0
+    ) -> Tuple[str, bool]:
         """
         1. Clone repository
         2. Checkout base branch
@@ -94,7 +117,7 @@ def prepare(self, repo: source.Repository, iteration: int = 0) -> Tuple[str, boo
                 branch=repo.base_branch,
             )
             shutil.rmtree(checkout_dir)
-            return self.prepare(repo, iteration=1)
+            return self.prepare(force_rebase=force_rebase, iteration=1, repo=repo)
 
         hash_before_pull = str(git_repo.head.commit)
         log.debug(
@@ -146,7 +169,18 @@ def prepare(self, repo: source.Repository, iteration: int = 0) -> Tuple[str, boo
 
         log.debug("Checking out work branch", branch=self.branch_name)
         git_repo.heads[self.branch_name].checkout()
+        if remote_branch is not None:
+            log.debug("Pulling changes into work branch", branch=self.branch_name)
+            # `rebase=True` to end up with a clean history.
+            # `strategy_option="theirs"` to always prefer changes from the remote.
+            # Commits by someone else will be preserved with this strategy and there
+            # will be no conflict.
+            git_repo.remotes["origin"].pull(rebase=True, strategy_option="theirs")
+
         merge_base = git_repo.git.merge_base(repo.base_branch, self.branch_name)
+        if force_rebase is False:
+            self._detect_modified_branch(merge_base=merge_base, repo=git_repo)
+
         log.debug("Resetting to merge base", branch=self.branch_name)
         git_repo.git.reset(merge_base, hard=True)
         log.debug("Rebasing onto work branch", branch=self.branch_name)

rcmt/rcmt.py

@@ -5,8 +5,9 @@
 import datetime
 import shutil
 from enum import Enum
-from typing import Iterator, Optional
+from typing import Any, Iterator, Optional
 
+import jinja2
 import structlog
 from git.exc import GitCommandError
 
@@ -16,6 +17,24 @@
 log: structlog.stdlib.BoundLogger = structlog.get_logger()
 
 
+TEMPLATE_BRANCH_MODIFIED = jinja2.Template(
+    """:warning: **This pull request has been modified.**
+
+This is a safety mechanism to prevent rcmt from accidentally overriding custom commits.
+
+rcmt will not be able to resolve merge conflicts with `{{ default_branch }}` automatically.
+It will not update this pull request or auto-merge it.
+
+Check the box in the description of this PR to force a rebase. This will remove all commits not made by rcmt.
+
+The commit(s) that modified the pull request:
+{% for checksum in checksums %}
+- {{ checksum }}
+{% endfor %}
+"""
+)
+
+
 class Options:
     def __init__(self, cfg: config.Config):
         self.config = cfg
@@ -75,8 +94,28 @@ def execute(
             )
             return RunResult.NO_CHANGES
 
+        force_rebase = self._has_rebase_checked(pr=pr_identifier, repo=repo)
         try:
-            work_dir, has_conflict = self.git.prepare(repo)
+            work_dir, has_conflict = self.git.prepare(
+                force_rebase=force_rebase, repo=repo
+            )
+            if force_rebase is True:
+                # It is likely that the branch was modified previously and a comment was
+                # created to notify the user. Delete that comment.
+                repo.delete_pr_comment_with_identifier(
+                    identifier="branch-modified", pr=pr_identifier
+                )
+
+        except git.BranchModifiedError as e:
+            log.warn("Branch contains commits not made by rcmt")
+            ctx.repo.create_pr_comment_with_identifier(
+                body=TEMPLATE_BRANCH_MODIFIED.render(
+                    checksums=e.checksums, default_branch=repo.base_branch
+                ),
+                identifier="branch-modified",
+                pr=pr_identifier,
+            )
+            return RunResult.NO_CHANGES
         except GitCommandError as e:
             # Catch any error raised by the git client, delete the repository and
             # initialize it again
@@ -86,7 +125,9 @@ def execute(
             )
             checkout_dir = self.git.checkout_dir(repo)
             shutil.rmtree(checkout_dir)
-            work_dir, has_conflict = self.git.prepare(repo)
+            work_dir, has_conflict = self.git.prepare(
+                force_rebase=force_rebase, repo=repo
+            )
 
         apply_actions(ctx=ctx, task_=matcher, work_dir=work_dir)
         has_local_changes = self.git.has_changes_local(work_dir)
@@ -202,6 +243,14 @@ def execute(
 
         return RunResult.NO_CHANGES
 
+    @staticmethod
+    def _has_rebase_checked(pr: Any, repo: source.Repository) -> bool:
+        if pr is None:
+            return False
+
+        desc = repo.get_pr_body(pr)
+        return "[x] If you want to rebase this PR" in desc
+
 
 def apply_actions(
     ctx: context.Context,

rcmt/source/github.py

@@ -14,7 +14,13 @@
 import structlog
 
 from ..log import SECRET_MASKER
-from .source import Base, PullRequest, Repository, add_credentials_to_url
+from .source import (
+    Base,
+    PullRequest,
+    PullRequestComment,
+    Repository,
+    add_credentials_to_url,
+)
 
 log = structlog.get_logger(source="github")
 
@@ -57,6 +63,9 @@ def close_pull_request(
         pr.create_issue_comment(body=message)
         pr.edit(state="closed")
 
+    def create_pr_comment(self, body: str, pr: github.PullRequest.PullRequest) -> None:
+        pr.create_issue_comment(body=body)
+
     def create_pull_request(self, branch: str, pr: PullRequest):
         log.debug(
             "Creating pull request", base=self.base_branch, head=branch, repo=str(self)
@@ -76,6 +85,11 @@ def delete_branch(self, identifier: github.PullRequest.PullRequest) -> None:
         if self.repo.delete_branch_on_merge is False:
             self.repo.get_git_ref(ref=f"heads/{identifier.head.ref}").delete()
 
+    def delete_pr_comment(
+        self, comment: PullRequestComment, pr: github.PullRequest.PullRequest
+    ) -> None:
+        pr.get_issue_comment(comment.id).delete()
+
     def find_pull_request(self, branch: str) -> Union[Any, None]:
         log.debug("Listing pull requests", repo=str(self))
         for pr in self.repo.get_pulls(state="all"):
@@ -98,6 +112,9 @@ def get_file(self, path: str) -> TextIO:
 
         return io.StringIO(file.decoded_content.decode("utf-8"))
 
+    def get_pr_body(self, pr: github.PullRequest.PullRequest) -> str:
+        return pr.body
+
     def has_file(self, path: str) -> bool:
         try:
             tree = self.repo.get_git_tree(self.base_branch, True)
@@ -136,6 +153,15 @@ def is_pr_merged(self, pr: github.PullRequest.PullRequest) -> bool:
     def is_pr_open(self, pr: github.PullRequest.PullRequest) -> bool:
         return pr.state == "open"
 
+    def list_pr_comments(
+        self, pr: github.PullRequest.PullRequest
+    ) -> Iterator[PullRequestComment]:
+        if pr is None:
+            return []
+
+        for issue in pr.get_issue_comments():
+            yield PullRequestComment(body=issue.body, id=issue.id)
+
     def merge_pull_request(self, pr: github.PullRequest.PullRequest) -> None:
         log.debug("Merging pull request", repo=str(self))
         pr.merge(commit_title="Auto-merge by rcmt")

rcmt/source/gitlab.py

@@ -18,7 +18,13 @@
 from gitlab.v4.objects.merge_requests import ProjectMergeRequest as GitlabMergeRequest
 
 from ..log import SECRET_MASKER
-from .source import Base, PullRequest, Repository, add_credentials_to_url
+from .source import (
+    Base,
+    PullRequest,
+    PullRequestComment,
+    Repository,
+    add_credentials_to_url,
+)
 
 log: structlog.stdlib.BoundLogger = structlog.get_logger(source="gitlab")
 
@@ -48,6 +54,9 @@ def close_pull_request(self, message: str, pr: GitlabMergeRequest) -> None:
         pr.state_event = "close"
         pr.save()
 
+    def create_pr_comment(self, body: str, pr: GitlabMergeRequest) -> None:
+        pr.notes.create({"body": body})
+
     def create_pull_request(self, branch: str, pr: PullRequest) -> None:
         log.debug(
             "Creating merge request", base=self.base_branch, head=branch, repo=str(self)
@@ -67,6 +76,11 @@ def delete_branch(self, identifier: GitlabMergeRequest) -> None:
         if identifier.should_remove_source_branch is not True:
             self._project.branches.get(id=identifier.source_branch, lazy=True).delete()
 
+    def delete_pr_comment(
+        self, comment: PullRequestComment, pr: GitlabMergeRequest
+    ) -> None:
+        pr.notes.delete(comment.id)
+
     def find_pull_request(self, branch: str) -> Union[Any, None]:
         log.debug("Listing merge requests", repo=str(self))
         mrs = self._project.mergerequests.list(
@@ -98,6 +112,9 @@ def get_file(self, path: str) -> TextIO:
         except GitlabGetError:
             raise FileNotFoundError("file does not exist in repository")
 
+    def get_pr_body(self, mr: GitlabMergeRequest) -> str:
+        return mr.description
+
     def has_file(self, path: str) -> bool:
         directory = os.path.dirname(path)
         try:
@@ -156,6 +173,13 @@ def is_pr_merged(self, mr: GitlabMergeRequest) -> bool:
     def is_pr_open(self, mr: GitlabMergeRequest) -> bool:
         return mr.state == "opened"
 
+    def list_pr_comments(self, mr: GitlabMergeRequest) -> Iterator[PullRequestComment]:
+        if mr is None:
+            return []
+
+        for note in mr.notes.list(iterator=True):
+            yield PullRequestComment(body=note.body, id=note.id)
+
     def merge_pull_request(self, identifier: GitlabMergeRequest):
         log.debug("Merging merge request", repo=str(self), id=identifier.get_id())
         identifier.merge()

rcmt/source/source.py

@@ -4,6 +4,7 @@
 
 import datetime
 import urllib.parse
+from dataclasses import dataclass
 from typing import Any, Generator, Iterator, Optional, TextIO, Union
 
 import humanize
@@ -100,6 +101,8 @@ def body(self) -> str:
         else:
             body += "**Ignore:** This PR will be recreated if closed.  \n"
 
+        body += "\n---\n- [ ] If you want to rebase this PR, check this box"
+
         body += """
 ---
 
@@ -116,6 +119,12 @@ def title(self) -> str:
         return f"{self.title_prefix} {self.title_body} {self.title_suffix}".strip()
 
 
+@dataclass
+class PullRequestComment:
+    body: str
+    id: Any
+
+
 class Repository:
     """
     Repository provides all methods needed to interact with a single repository of a
@@ -169,6 +178,25 @@ def close_pull_request(self, message: str, pr: Any) -> None:
             "class does not implement Repository.close_pull_request()"
         )
 
+    def create_pr_comment(self, body: str, pr: Any) -> None:
+        raise NotImplementedError(
+            "class does not implement Repository.create_pr_comment()"
+        )
+
+    def create_pr_comment_with_identifier(
+        self, body: str, identifier: str, pr: Any
+    ) -> None:
+        if identifier == "":
+            raise RuntimeError("identifier cannot be empty")
+
+        prefix = f"<!-- rcmt::{identifier} -->"
+        for comment in self.list_pr_comments(pr):
+            if comment.body.startswith(prefix):
+                return None
+
+        body = f"{prefix}\n{body}"
+        self.create_pr_comment(body=body, pr=pr)
+
     def create_pull_request(self, branch: str, pr: PullRequest) -> None:
         """
         Creates a pull request for the given branch.
@@ -181,6 +209,21 @@ def create_pull_request(self, branch: str, pr: PullRequest) -> None:
             "class does not implement Repository.create_pull_request()"
         )
 
+    def delete_pr_comment(self, comment: PullRequestComment, pr: Any) -> None:
+        raise NotImplementedError(
+            "class does not implement Repository.delete_comment()"
+        )
+
+    def delete_pr_comment_with_identifier(self, identifier: str, pr: Any) -> None:
+        if identifier == "":
+            raise RuntimeError("identifier cannot be empty")
+
+        prefix = f"<!-- rcmt::{identifier} -->"
+        for comment in self.list_pr_comments(pr):
+            if comment.body.startswith(prefix):
+                self.delete_pr_comment(comment=comment, pr=pr)
+                return
+
     def delete_branch(self, identifier: Any) -> None:
         raise NotImplementedError(
             "class does not implement Repository.delete_pull_request()"
@@ -210,6 +253,11 @@ def full_name(self) -> str:
     def get_file(self, path: str) -> TextIO:
         raise NotImplementedError("class does not implement Repository.has_file()")
 
+    def get_pr_body(self, pr_identifier: Any) -> str:
+        raise NotImplementedError(
+            "class does not implement Repository.get_pr_description()"
+        )
+
     def has_file(self, path: str) -> bool:
         """
         Checks if a file exists in a repository.
@@ -272,6 +320,11 @@ def is_pr_open(self, identifier: Any) -> bool:
         """
         raise NotImplementedError("class does not implement Repository.is_pr_open()")
 
+    def list_pr_comments(self, pr: Any) -> Iterator[PullRequestComment]:
+        raise NotImplementedError(
+            "class does not implement Repository.list_pr_comments()"
+        )
+
     def merge_pull_request(self, identifier: Any) -> None:
         """
         Merges a pull request.