Fixes squash merge request for Single Job Mode

.github/workflows/test.yml

@@ -23,6 +23,6 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
       - name: Install requirements
-        run: pip install -r requirements_frozen.txt
+        run: pip install -c constraints.txt -r requirements_frozen.txt
       - name: Run tests
         run: pytest

constraints.txt

@@ -0,0 +1 @@
+importlib-metadata<5.0

marge/job.py

@@ -9,7 +9,7 @@
 from .branch import Branch
 from .interval import IntervalUnion
 from .merge_request import MergeRequestRebaseFailed
-from .project import Project
+from .project import Project, SquashOption
 from .user import User
 from .pipeline import Pipeline
 
@@ -47,7 +47,12 @@ def ensure_mergeable_mr(self, merge_request):
         if merge_request.work_in_progress:
             raise CannotMerge("Sorry, I can't merge requests marked as Work-In-Progress!")
 
-        if merge_request.squash and self._options.requests_commit_tagging:
+        auto_squash = (
+            self._project.squash_option is SquashOption.always or
+            merge_request.squash
+        )
+
+        if auto_squash and self._options.requests_commit_tagging:
             raise CannotMerge(
                 "Sorry, merging requests marked as auto-squash would ruin my commit tagging!"
             )

marge/merge_request.py

@@ -200,14 +200,17 @@ def rebase(self):
 
         raise TimeoutError('Waiting for merge request to be rebased by GitLab')
 
-    def accept(self, remove_branch=False, sha=None, merge_when_pipeline_succeeds=True):
+    def accept(self, remove_branch=False, sha=None, merge_when_pipeline_succeeds=True, auto_squash=None):
+        params = dict(
+            should_remove_source_branch=remove_branch,
+            merge_when_pipeline_succeeds=merge_when_pipeline_succeeds,
+            sha=sha or self.sha,  # if provided, ensures what is merged is what we want (or fails)
+        )
+        if auto_squash is not None:
+            params['squash'] = auto_squash
         return self._api.call(PUT(
             '/projects/{0.project_id}/merge_requests/{0.iid}/merge'.format(self),
-            dict(
-                should_remove_source_branch=remove_branch,
-                merge_when_pipeline_succeeds=merge_when_pipeline_succeeds,
-                sha=sha or self.sha,  # if provided, ensures what is merged is what we want (or fails)
-            ),
+            params
         ))
 
     def close(self):

marge/project.py

@@ -1,5 +1,5 @@
 import logging as log
-from enum import IntEnum, unique
+from enum import Enum, IntEnum, unique
 from functools import partial
 
 from . import gitlab
@@ -89,6 +89,10 @@ def http_url_to_repo(self):
     def merge_requests_enabled(self):
         return self.info['merge_requests_enabled']
 
+    @property
+    def squash_option(self):
+        return SquashOption(self.info['squash_option'])
+
     @property
     def only_allow_merge_if_pipeline_succeeds(self):
         return self.info['only_allow_merge_if_pipeline_succeeds']
@@ -124,3 +128,11 @@ class AccessLevel(IntEnum):
     developer = 30
     maintainer = 40
     owner = 50
+
+
+@unique
+class SquashOption(str, Enum):
+    always = "always"
+    default_off = "default_off"
+    default_on = "default_on"
+    never = "never"

marge/single_merge_job.py

@@ -6,6 +6,7 @@
 from . import git, gitlab
 from .commit import Commit
 from .job import CannotMerge, GitLabRebaseResultMismatch, MergeJob, SkipMerge
+from .project import SquashOption
 
 
 class SingleMergeJob(MergeJob):
@@ -92,11 +93,14 @@ def update_merge_request_and_accept(self, approvals):
 
             self.ensure_mergeable_mr(merge_request)
 
+            auto_squash = True if target_project.squash_option is SquashOption.always else None
+
             try:
                 ret = merge_request.accept(
                     remove_branch=merge_request.force_remove_source_branch,
                     sha=actual_sha,
                     merge_when_pipeline_succeeds=bool(target_project.only_allow_merge_if_pipeline_succeeds),
+                    auto_squash=auto_squash
                 )
                 log.info('merge_request.accept result: %s', ret)
             except gitlab.NotAcceptable as err:

tests/test_job.py

@@ -144,15 +144,45 @@ def test_ensure_mergeable_mr_unresolved_discussion(self):
 
         assert exc_info.value.reason == "Sorry, I can't merge requests which have unresolved discussions!"
 
-    def test_ensure_mergeable_mr_squash_and_trailers(self):
-        merge_job = self.get_merge_job(options=MergeJobOptions.default(add_reviewers=True))
+    @pytest.mark.parametrize('squash_option', marge.project.SquashOption)
+    def test_ensure_mergeable_mr_squash_wanted_and_trailers(self, squash_option):
+        merge_job = self.get_merge_job(
+            project=create_autospec(
+                marge.project.Project,
+                spec_set=True,
+                squash_option=squash_option,
+            ),
+            options=MergeJobOptions.default(add_reviewers=True)
+        )
         merge_request = self._mock_merge_request(
             assignee_ids=[merge_job._user.id],
             state='opened',
             work_in_progress=False,
             squash=True,
         )
-        merge_request.fetch_approvals.return_value.sufficient = True
+        with pytest.raises(CannotMerge) as exc_info:
+            merge_job.ensure_mergeable_mr(merge_request)
+
+        assert (
+            exc_info.value.reason == "Sorry, merging requests marked as auto-squash "
+                                     "would ruin my commit tagging!"
+        )
+
+    def test_ensure_mergeable_mr_squash_needed_and_trailers(self):
+        merge_job = self.get_merge_job(
+            project=create_autospec(
+                marge.project.Project,
+                spec_set=True,
+                squash_option=marge.project.SquashOption.always,
+            ),
+            options=MergeJobOptions.default(add_reviewers=True),
+        )
+        merge_request = self._mock_merge_request(
+            assignee_ids=[merge_job._user.id],
+            state='opened',
+            work_in_progress=False,
+            squash=False,
+        )
         with pytest.raises(CannotMerge) as exc_info:
             merge_job.ensure_mergeable_mr(merge_request)
 

tests/test_merge_request.py

@@ -161,6 +161,32 @@ def test_rebase_was_in_progress_no_error(self):
         self.merge_request.rebase()
         self.api.call.assert_has_calls([call(req) for (req, resp) in expected])
 
+    @pytest.mark.parametrize('squash_wanted', [True, False])
+    def test_accept_auto_squash_is_boolean(self, squash_wanted):
+        self._load(dict(INFO, sha='badc0de'))
+        self.merge_request.accept(auto_squash=squash_wanted)
+        self.api.call.assert_called_once_with(PUT(
+            '/projects/1234/merge_requests/54/merge',
+            dict(
+                merge_when_pipeline_succeeds=True,
+                should_remove_source_branch=False,
+                sha='badc0de',
+                squash=squash_wanted,
+            )
+        ))
+
+    def test_accept_auto_squash_is_none(self):
+        self._load(dict(INFO, sha='badc0de'))
+        self.merge_request.accept(auto_squash=None)
+        self.api.call.assert_called_once_with(PUT(
+            '/projects/1234/merge_requests/54/merge',
+            dict(
+                merge_when_pipeline_succeeds=True,
+                should_remove_source_branch=False,
+                sha='badc0de',
+            )
+        ))
+
     def test_accept_remove_branch(self):
         self._load(dict(INFO, sha='badc0de'))
 

tests/test_project.py

@@ -20,7 +20,8 @@
         'group_access': {
             'access_level': AccessLevel.developer.value,
         }
-    }
+    },
+    'squash_option': 'default_off',
 }
 
 GROUP_ACCESS = {
@@ -109,6 +110,7 @@ def test_properties(self):
         assert project.merge_requests_enabled is True
         assert project.only_allow_merge_if_pipeline_succeeds is True
         assert project.only_allow_merge_if_all_discussions_are_resolved is False
+        assert project.squash_option == 'default_off'
         assert project.access_level == AccessLevel.developer
 
     def test_group_access(self):