Prechádzať zdrojové kódy

Add support for remote.<name>.mirror configuration (#1652)

This change implements Git's mirror mode for push operations. When
remote.<name>.mirror is set to true, push will:

- Push all local refs to the remote with the same names
- Delete remote refs that don't exist locally

This is useful for maintaining exact mirrors of repositories. The
implementation:

- Checks the remote.<name>.mirror configuration before determining
refspecs
- In mirror mode, automatically includes all refs and handles deletions
Jelmer Vernooij 1 mesiac pred
rodič
commit
6fc829a574
2 zmenil súbory, kde vykonal 135 pridanie a 2 odobranie
  1. 27 2
      dulwich/porcelain.py
  2. 108 0
      tests/test_porcelain.py

+ 27 - 2
dulwich/porcelain.py

@@ -1784,9 +1784,25 @@ def push(
     """
     # Open the repo
     with open_repo_closing(repo) as r:
-        if refspecs is None:
-            refspecs = [active_branch(r)]
         (remote_name, remote_location) = get_remote_repo(r, remote_location)
+        # Check if mirror mode is enabled
+        mirror_mode = False
+        if remote_name:
+            try:
+                mirror_mode = r.get_config_stack().get_boolean(
+                    (b"remote", remote_name.encode()), b"mirror"
+                )
+            except KeyError:
+                pass
+
+        if mirror_mode:
+            # Mirror mode: push all refs and delete non-existent ones
+            refspecs = []
+            for ref in r.refs.keys():
+                # Push all refs to the same name on remote
+                refspecs.append(ref + b":" + ref)
+        elif refspecs is None:
+            refspecs = [active_branch(r)]
 
         # Get the client and path
         client, path = get_transport_and_path(
@@ -1799,6 +1815,15 @@ def push(
         def update_refs(refs):
             selected_refs.extend(parse_reftuples(r.refs, refs, refspecs, force=force))
             new_refs = {}
+
+            # In mirror mode, delete remote refs that don't exist locally
+            if mirror_mode:
+                local_refs = set(r.refs.keys())
+                for remote_ref in refs.keys():
+                    if remote_ref not in local_refs:
+                        new_refs[remote_ref] = ZERO_SHA
+                        remote_changed_refs[remote_ref] = None
+
             # TODO: Handle selected_refs == {None: None}
             for lh, rh, force_ref in selected_refs:
                 if lh is None:

+ 108 - 0
tests/test_porcelain.py

@@ -3998,6 +3998,114 @@ class PushTests(PorcelainTestCase):
         if result.ref_status:
             self.assertIsNone(result.ref_status.get(b"refs/heads/new-branch"))
 
+    def test_mirror_mode(self) -> None:
+        """Test push with remote.<name>.mirror configuration."""
+        outstream = BytesIO()
+        errstream = BytesIO()
+
+        # Create initial commit
+        porcelain.commit(
+            repo=self.repo.path,
+            message=b"init",
+            author=b"author <email>",
+            committer=b"committer <email>",
+        )
+
+        # Setup target repo cloned from temp test repo
+        clone_path = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, clone_path)
+        target_repo = porcelain.clone(
+            self.repo.path, target=clone_path, errstream=errstream
+        )
+        target_repo.close()
+
+        # Create multiple refs in the clone
+        with Repo(clone_path) as r_clone:
+            # Create a new branch
+            r_clone.refs[b"refs/heads/feature"] = r_clone[b"HEAD"].id
+            # Create a tag
+            r_clone.refs[b"refs/tags/v1.0"] = r_clone[b"HEAD"].id
+            # Create a remote tracking branch
+            r_clone.refs[b"refs/remotes/upstream/main"] = r_clone[b"HEAD"].id
+
+        # Create a branch in the remote that doesn't exist in clone
+        self.repo.refs[b"refs/heads/to-be-deleted"] = self.repo[b"HEAD"].id
+
+        # Configure mirror mode
+        with Repo(clone_path) as r_clone:
+            config = r_clone.get_config()
+            config.set((b"remote", b"origin"), b"mirror", True)
+            config.write_to_path()
+
+        # Push with mirror mode
+        porcelain.push(
+            clone_path,
+            "origin",
+            outstream=outstream,
+            errstream=errstream,
+        )
+
+        # Verify refs were properly mirrored
+        with Repo(clone_path) as r_clone:
+            # All local branches should be pushed
+            self.assertEqual(
+                r_clone.refs[b"refs/heads/feature"],
+                self.repo.refs[b"refs/heads/feature"],
+            )
+            # All tags should be pushed
+            self.assertEqual(
+                r_clone.refs[b"refs/tags/v1.0"], self.repo.refs[b"refs/tags/v1.0"]
+            )
+            # Remote tracking branches should be pushed
+            self.assertEqual(
+                r_clone.refs[b"refs/remotes/upstream/main"],
+                self.repo.refs[b"refs/remotes/upstream/main"],
+            )
+
+        # Verify the extra branch was deleted
+        self.assertNotIn(b"refs/heads/to-be-deleted", self.repo.refs)
+
+    def test_mirror_mode_disabled(self) -> None:
+        """Test that mirror mode is properly disabled when set to false."""
+        outstream = BytesIO()
+        errstream = BytesIO()
+
+        # Create initial commit
+        porcelain.commit(
+            repo=self.repo.path,
+            message=b"init",
+            author=b"author <email>",
+            committer=b"committer <email>",
+        )
+
+        # Setup target repo cloned from temp test repo
+        clone_path = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, clone_path)
+        target_repo = porcelain.clone(
+            self.repo.path, target=clone_path, errstream=errstream
+        )
+        target_repo.close()
+
+        # Create a branch in the remote that doesn't exist in clone
+        self.repo.refs[b"refs/heads/should-not-be-deleted"] = self.repo[b"HEAD"].id
+
+        # Explicitly set mirror mode to false
+        with Repo(clone_path) as r_clone:
+            config = r_clone.get_config()
+            config.set((b"remote", b"origin"), b"mirror", False)
+            config.write_to_path()
+
+        # Push normally (not mirror mode)
+        porcelain.push(
+            clone_path,
+            "origin",
+            outstream=outstream,
+            errstream=errstream,
+        )
+
+        # Verify the extra branch was NOT deleted
+        self.assertIn(b"refs/heads/should-not-be-deleted", self.repo.refs)
+
 
 class PullTests(PorcelainTestCase):
     def setUp(self) -> None: