Add basic support for merge drivers (#1690)

NEWS

@@ -1,5 +1,8 @@
 0.23.3	UNRELEASED
 
+ * Add support for merge drivers.
+   (Jelmer Vernooij)
+
  * ``dulwich.porcelain.diff``: Support diffing two commits
    and diffing cached and working tree. (Jelmer Vernooij)
 

dulwich/merge.py

@@ -7,6 +7,9 @@ try:
 except ImportError:
     merge3 = None  # type: ignore
 
+from dulwich.attrs import GitAttributes
+from dulwich.config import Config
+from dulwich.merge_drivers import get_merge_driver_registry
 from dulwich.object_store import BaseObjectStore
 from dulwich.objects import S_ISGITLINK, Blob, Commit, Tree, is_blob, is_tree
 
@@ -91,6 +94,9 @@ def merge_blobs(
     base_blob: Optional[Blob],
     ours_blob: Optional[Blob],
     theirs_blob: Optional[Blob],
+    path: Optional[bytes] = None,
+    gitattributes: Optional[GitAttributes] = None,
+    config: Optional[Config] = None,
 ) -> tuple[bytes, bool]:
     """Perform three-way merge on blob contents.
 
@@ -98,10 +104,44 @@ def merge_blobs(
         base_blob: Common ancestor blob (can be None)
         ours_blob: Our version of the blob (can be None)
         theirs_blob: Their version of the blob (can be None)
+        path: Optional path of the file being merged
+        gitattributes: Optional GitAttributes object for checking merge drivers
+        config: Optional Config object for loading merge driver configuration
 
     Returns:
         Tuple of (merged_content, had_conflicts)
     """
+    # Check for merge driver
+    merge_driver_name = None
+    if path and gitattributes:
+        attrs = gitattributes.match_path(path)
+        merge_value = attrs.get(b"merge")
+        if merge_value and isinstance(merge_value, bytes) and merge_value != b"text":
+            merge_driver_name = merge_value.decode("utf-8", errors="replace")
+
+    # Use merge driver if found
+    if merge_driver_name:
+        registry = get_merge_driver_registry(config)
+        driver = registry.get_driver(merge_driver_name)
+        if driver:
+            # Get content from blobs
+            base_content = base_blob.data if base_blob else b""
+            ours_content = ours_blob.data if ours_blob else b""
+            theirs_content = theirs_blob.data if theirs_blob else b""
+
+            # Use merge driver
+            merged_content, success = driver.merge(
+                ancestor=base_content,
+                ours=ours_content,
+                theirs=theirs_content,
+                path=path.decode("utf-8", errors="replace") if path else None,
+                marker_size=7,
+            )
+            # Convert success (no conflicts) to had_conflicts (conflicts occurred)
+            had_conflicts = not success
+            return merged_content, had_conflicts
+
+    # Fall back to default merge behavior
     # Handle deletion cases
     if ours_blob is None and theirs_blob is None:
         return b"", False
@@ -183,19 +223,29 @@ def merge_blobs(
 class Merger:
     """Handles git merge operations."""
 
-    def __init__(self, object_store: BaseObjectStore) -> None:
+    def __init__(
+        self,
+        object_store: BaseObjectStore,
+        gitattributes: Optional[GitAttributes] = None,
+        config: Optional[Config] = None,
+    ) -> None:
         """Initialize merger.
 
         Args:
             object_store: Object store to read objects from
+            gitattributes: Optional GitAttributes object for checking merge drivers
+            config: Optional Config object for loading merge driver configuration
         """
         self.object_store = object_store
+        self.gitattributes = gitattributes
+        self.config = config
 
-    @staticmethod
     def merge_blobs(
+        self,
         base_blob: Optional[Blob],
         ours_blob: Optional[Blob],
         theirs_blob: Optional[Blob],
+        path: Optional[bytes] = None,
     ) -> tuple[bytes, bool]:
         """Perform three-way merge on blob contents.
 
@@ -203,11 +253,14 @@ class Merger:
             base_blob: Common ancestor blob (can be None)
             ours_blob: Our version of the blob (can be None)
             theirs_blob: Their version of the blob (can be None)
+            path: Optional path of the file being merged
 
         Returns:
             Tuple of (merged_content, had_conflicts)
         """
-        return merge_blobs(base_blob, ours_blob, theirs_blob)
+        return merge_blobs(
+            base_blob, ours_blob, theirs_blob, path, self.gitattributes, self.config
+        )
 
     def merge_trees(
         self, base_tree: Optional[Tree], ours_tree: Tree, theirs_tree: Tree
@@ -375,7 +428,7 @@ class Merger:
                     assert isinstance(ours_blob, Blob)
                     assert isinstance(theirs_blob, Blob)
                     merged_content, had_conflict = self.merge_blobs(
-                        base_blob, ours_blob, theirs_blob
+                        base_blob, ours_blob, theirs_blob, path
                     )
 
                     if had_conflict:
@@ -400,6 +453,8 @@ def three_way_merge(
     base_commit: Optional[Commit],
     ours_commit: Commit,
     theirs_commit: Commit,
+    gitattributes: Optional[GitAttributes] = None,
+    config: Optional[Config] = None,
 ) -> tuple[Tree, list[bytes]]:
     """Perform a three-way merge between commits.
 
@@ -408,11 +463,13 @@ def three_way_merge(
         base_commit: Common ancestor commit (None if no common ancestor)
         ours_commit: Our commit
         theirs_commit: Their commit
+        gitattributes: Optional GitAttributes object for checking merge drivers
+        config: Optional Config object for loading merge driver configuration
 
     Returns:
         tuple of (merged_tree, list_of_conflicted_paths)
     """
-    merger = Merger(object_store)
+    merger = Merger(object_store, gitattributes, config)
 
     base_tree = None
     if base_commit:

dulwich/merge_drivers.py

@@ -0,0 +1,245 @@
+# merge_drivers.py -- Merge driver support for dulwich
+# Copyright (C) 2025 Jelmer Vernooij <jelmer@jelmer.uk>
+#
+# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
+# General Public License as public by the Free Software Foundation; version 2.0
+# or (at your option) any later version. You can redistribute it and/or
+# modify it under the terms of either of these two licenses.
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# You should have received a copy of the licenses; if not, see
+# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
+# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
+# License, Version 2.0.
+#
+
+"""Merge driver support for dulwich."""
+
+import os
+import subprocess
+import tempfile
+from typing import Any, Optional, Protocol
+
+from .config import Config
+
+
+class MergeDriver(Protocol):
+    """Protocol for merge drivers."""
+
+    def merge(
+        self,
+        ancestor: bytes,
+        ours: bytes,
+        theirs: bytes,
+        path: Optional[str] = None,
+        marker_size: int = 7,
+    ) -> tuple[bytes, bool]:
+        """Perform a three-way merge.
+
+        Args:
+            ancestor: Content of the common ancestor version
+            ours: Content of our version
+            theirs: Content of their version
+            path: Optional path of the file being merged
+            marker_size: Size of conflict markers (default 7)
+
+        Returns:
+            Tuple of (merged content, success flag)
+            If success is False, the content may contain conflict markers
+        """
+        ...
+
+
+class ProcessMergeDriver:
+    """Merge driver that runs an external process."""
+
+    def __init__(self, command: str, name: str = "custom"):
+        """Initialize process merge driver.
+
+        Args:
+            command: Command to run for merging
+            name: Name of the merge driver
+        """
+        self.command = command
+        self.name = name
+
+    def merge(
+        self,
+        ancestor: bytes,
+        ours: bytes,
+        theirs: bytes,
+        path: Optional[str] = None,
+        marker_size: int = 7,
+    ) -> tuple[bytes, bool]:
+        """Perform merge using external process.
+
+        The command is executed with the following placeholders:
+        - %O: path to ancestor version (base)
+        - %A: path to our version
+        - %B: path to their version
+        - %L: conflict marker size
+        - %P: original path of the file
+
+        The command should write the merge result to the file at %A.
+        Exit code 0 means successful merge, non-zero means conflicts.
+        """
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Write temporary files
+            ancestor_path = os.path.join(tmpdir, "ancestor")
+            ours_path = os.path.join(tmpdir, "ours")
+            theirs_path = os.path.join(tmpdir, "theirs")
+
+            with open(ancestor_path, "wb") as f:
+                f.write(ancestor)
+            with open(ours_path, "wb") as f:
+                f.write(ours)
+            with open(theirs_path, "wb") as f:
+                f.write(theirs)
+
+            # Prepare command with placeholders
+            cmd = self.command
+            cmd = cmd.replace("%O", ancestor_path)
+            cmd = cmd.replace("%A", ours_path)
+            cmd = cmd.replace("%B", theirs_path)
+            cmd = cmd.replace("%L", str(marker_size))
+            if path:
+                cmd = cmd.replace("%P", path)
+
+            # Execute merge command
+            try:
+                result = subprocess.run(
+                    cmd,
+                    shell=True,
+                    capture_output=True,
+                    text=False,
+                )
+
+                # Read merged content from ours file
+                with open(ours_path, "rb") as f:
+                    merged_content = f.read()
+
+                # Exit code 0 means clean merge, non-zero means conflicts
+                success = result.returncode == 0
+
+                return merged_content, success
+
+            except subprocess.SubprocessError:
+                # If the command fails completely, return original with conflicts
+                return ours, False
+
+
+class MergeDriverRegistry:
+    """Registry for merge drivers."""
+
+    def __init__(self, config: Optional[Config] = None):
+        """Initialize merge driver registry.
+
+        Args:
+            config: Git configuration object
+        """
+        self._drivers: dict[str, MergeDriver] = {}
+        self._factories: dict[str, Any] = {}
+        self._config = config
+
+        # Register built-in drivers
+        self._register_builtin_drivers()
+
+    def _register_builtin_drivers(self):
+        """Register built-in merge drivers."""
+        # The "text" driver is the default three-way merge
+        # We don't register it here as it's handled by the default merge code
+
+    def register_driver(self, name: str, driver: MergeDriver):
+        """Register a merge driver instance.
+
+        Args:
+            name: Name of the merge driver
+            driver: Driver instance
+        """
+        self._drivers[name] = driver
+
+    def register_factory(self, name: str, factory):
+        """Register a factory function for creating merge drivers.
+
+        Args:
+            name: Name of the merge driver
+            factory: Factory function that returns a MergeDriver
+        """
+        self._factories[name] = factory
+
+    def get_driver(self, name: str) -> Optional[MergeDriver]:
+        """Get a merge driver by name.
+
+        Args:
+            name: Name of the merge driver
+
+        Returns:
+            MergeDriver instance or None if not found
+        """
+        # First check registered drivers
+        if name in self._drivers:
+            return self._drivers[name]
+
+        # Then check factories
+        if name in self._factories:
+            driver = self._factories[name]()
+            self._drivers[name] = driver
+            return driver
+
+        # Finally check configuration
+        if self._config:
+            driver = self._create_from_config(name)
+            if driver:
+                self._drivers[name] = driver
+                return driver
+
+        return None
+
+    def _create_from_config(self, name: str) -> Optional[MergeDriver]:
+        """Create a merge driver from git configuration.
+
+        Args:
+            name: Name of the merge driver
+
+        Returns:
+            MergeDriver instance or None if not configured
+        """
+        if not self._config:
+            return None
+
+        # Look for merge.<name>.driver configuration
+        try:
+            command = self._config.get(("merge", name), "driver")
+            if command:
+                return ProcessMergeDriver(command.decode(), name)
+        except KeyError:
+            pass
+
+        return None
+
+
+# Global registry instance
+_merge_driver_registry: Optional[MergeDriverRegistry] = None
+
+
+def get_merge_driver_registry(config: Optional[Config] = None) -> MergeDriverRegistry:
+    """Get the global merge driver registry.
+
+    Args:
+        config: Git configuration object
+
+    Returns:
+        MergeDriverRegistry instance
+    """
+    global _merge_driver_registry
+    if _merge_driver_registry is None:
+        _merge_driver_registry = MergeDriverRegistry(config)
+    elif config is not None:
+        # Update config if provided
+        _merge_driver_registry._config = config
+    return _merge_driver_registry

dulwich/porcelain.py

@@ -3762,8 +3762,10 @@ def _do_merge(
 
     # Perform three-way merge
     base_commit = r[base_commit_id]
+    gitattributes = r.get_gitattributes()
+    config = r.get_config()
     merged_tree, conflicts = three_way_merge(
-        r.object_store, base_commit, head_commit, merge_commit
+        r.object_store, base_commit, head_commit, merge_commit, gitattributes, config
     )
 
     # Add merged tree to object store
@@ -3917,7 +3919,9 @@ def merge_tree(
         theirs = parse_tree(r, their_tree)
 
         # Perform the merge
-        merger = Merger(r.object_store)
+        gitattributes = r.get_gitattributes()
+        config = r.get_config()
+        merger = Merger(r.object_store, gitattributes, config)
         merged_tree, conflicts = merger.merge_trees(base, ours, theirs)
 
         # Add the merged tree to the object store

examples/merge_driver.py

@@ -0,0 +1,124 @@
+#!/usr/bin/python3
+# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
+
+"""Simple example demonstrating merge driver usage in dulwich.
+
+This example:
+1. Creates a test repository with .gitattributes
+2. Implements a JSON merge driver
+3. Creates two branches with conflicting JSON changes
+4. Merges one commit into another using the custom driver
+"""
+
+import json
+import os
+from typing import Optional
+
+from dulwich import porcelain
+from dulwich.merge_drivers import get_merge_driver_registry
+from dulwich.repo import Repo
+
+
+class JSONMergeDriver:
+    """Simple merge driver for JSON files."""
+
+    def merge(
+        self,
+        ancestor: bytes,
+        ours: bytes,
+        theirs: bytes,
+        path: Optional[str] = None,
+        marker_size: int = 7,
+    ) -> tuple[bytes, bool]:
+        """Merge JSON files by combining objects."""
+        try:
+            # Parse JSON content
+            ancestor_data = json.loads(ancestor.decode()) if ancestor.strip() else {}
+            ours_data = json.loads(ours.decode()) if ours.strip() else {}
+            theirs_data = json.loads(theirs.decode()) if theirs.strip() else {}
+
+            # Simple merge: combine all fields
+            merged: dict = {}
+            merged.update(ancestor_data)
+            merged.update(ours_data)
+            merged.update(theirs_data)
+
+            # Convert back to JSON with nice formatting
+            result = json.dumps(merged, indent=2).encode()
+            return result, True
+
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            # Fall back to simple concatenation on parse error
+            result = ours + b"\n<<<<<<< MERGE CONFLICT >>>>>>>\n" + theirs
+            return result, False
+
+
+# Create temporary directory for test repo
+# Initialize repository
+repo = Repo.init("merge-driver", mkdir=True)
+
+# Create .gitattributes file
+gitattributes_path = os.path.join(repo.path, ".gitattributes")
+with open(gitattributes_path, "w") as f:
+    f.write("*.json merge=jsondriver\n")
+
+# Create initial JSON file
+config_path = os.path.join(repo.path, "config.json")
+initial_config = {"name": "test-project", "version": "1.0.0"}
+with open(config_path, "w") as f:
+    json.dump(initial_config, f, indent=2)
+
+# Add and commit initial files
+repo.stage([".gitattributes", "config.json"])
+initial_commit = repo.do_commit(b"Initial commit", committer=b"Test <test@example.com>")
+
+# Register our custom merge driver globally
+registry = get_merge_driver_registry()
+registry.register_driver("jsondriver", JSONMergeDriver())
+
+# Create and switch to feature branch
+porcelain.branch_create(repo, "feature")
+repo.refs[b"HEAD"] = repo.refs[b"refs/heads/feature"]
+
+# Make changes on feature branch
+feature_config = {
+    "name": "test-project",
+    "version": "1.0.0",
+    "author": "Alice",
+    "features": ["logging", "database"],
+}
+with open(config_path, "w") as f:
+    json.dump(feature_config, f, indent=2)
+
+repo.stage(["config.json"])
+feature_commit = repo.do_commit(
+    b"Add author and features", committer=b"Alice <alice@example.com>"
+)
+
+# Switch back to master
+repo.refs[b"HEAD"] = repo.refs[b"refs/heads/master"]
+
+# Make different changes on master
+master_config = {
+    "name": "test-project",
+    "version": "1.1.0",
+    "description": "A test project for merge drivers",
+    "license": "Apache-2.0",
+}
+with open(config_path, "w") as f:
+    json.dump(master_config, f, indent=2)
+
+repo.stage(["config.json"])
+master_commit = repo.do_commit(
+    b"Add description and license", committer=b"Bob <bob@example.com>"
+)
+
+# Perform the merge using porcelain.merge
+# The merge should use our custom JSON driver for config.json
+merge_result = porcelain.merge(repo, "feature")
+# Show the merged content
+with open(config_path) as f:
+    merged_content = f.read()
+
+print("\nMerged config.json content:")
+print(merged_content)

tests/__init__.py

@@ -146,6 +146,7 @@ def self_test_suite():
         "lru_cache",
         "mailmap",
         "merge",
+        "merge_drivers",
         "missing_obj_finder",
         "notes",
         "objects",

tests/test_merge_drivers.py

@@ -0,0 +1,359 @@
+# test_merge_drivers.py -- Tests for merge driver support
+# Copyright (C) 2025 Jelmer Vernooij <jelmer@jelmer.uk>
+#
+# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
+# General Public License as public by the Free Software Foundation; version 2.0
+# or (at your option) any later version. You can redistribute it and/or
+# modify it under the terms of either of these two licenses.
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# You should have received a copy of the licenses; if not, see
+# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
+# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
+# License, Version 2.0.
+#
+
+"""Tests for merge driver support."""
+
+import sys
+import unittest
+from typing import Optional
+
+from dulwich.attrs import GitAttributes, Pattern
+from dulwich.config import ConfigDict
+from dulwich.merge import merge_blobs
+from dulwich.merge_drivers import (
+    MergeDriverRegistry,
+    ProcessMergeDriver,
+    get_merge_driver_registry,
+)
+from dulwich.objects import Blob
+
+
+class _TestMergeDriver:
+    """Test merge driver implementation."""
+
+    def __init__(self, name: str = "test"):
+        self.name = name
+        self.called = False
+        self.last_args = None
+
+    def merge(
+        self,
+        ancestor: bytes,
+        ours: bytes,
+        theirs: bytes,
+        path: Optional[str] = None,
+        marker_size: int = 7,
+    ) -> tuple[bytes, bool]:
+        """Test merge implementation."""
+        self.called = True
+        self.last_args = {
+            "ancestor": ancestor,
+            "ours": ours,
+            "theirs": theirs,
+            "path": path,
+            "marker_size": marker_size,
+        }
+
+        # Simple test merge: combine all three versions
+        result = b"TEST MERGE OUTPUT\n"
+        result += b"Ancestor: " + ancestor + b"\n"
+        result += b"Ours: " + ours + b"\n"
+        result += b"Theirs: " + theirs + b"\n"
+
+        # Return success if all three are different
+        success = ancestor != ours and ancestor != theirs
+        return result, success
+
+
+class MergeDriverRegistryTests(unittest.TestCase):
+    """Tests for MergeDriverRegistry."""
+
+    def test_register_driver(self):
+        """Test registering a merge driver."""
+        registry = MergeDriverRegistry()
+        driver = _TestMergeDriver("mydriver")
+
+        registry.register_driver("mydriver", driver)
+        retrieved = registry.get_driver("mydriver")
+
+        self.assertIs(retrieved, driver)
+
+    def test_register_factory(self):
+        """Test registering a merge driver factory."""
+        registry = MergeDriverRegistry()
+
+        def create_driver():
+            return _TestMergeDriver("factory_driver")
+
+        registry.register_factory("factory", create_driver)
+        driver = registry.get_driver("factory")
+
+        self.assertIsInstance(driver, _TestMergeDriver)
+        self.assertEqual(driver.name, "factory_driver")
+
+        # Second call should return the same instance
+        driver2 = registry.get_driver("factory")
+        self.assertIs(driver2, driver)
+
+    def test_get_nonexistent_driver(self):
+        """Test getting a non-existent driver returns None."""
+        registry = MergeDriverRegistry()
+        driver = registry.get_driver("nonexistent")
+        self.assertIsNone(driver)
+
+    def test_create_from_config(self):
+        """Test creating a merge driver from configuration."""
+        config = ConfigDict()
+        config.set((b"merge", b"xmlmerge"), b"driver", b"xmlmerge %O %A %B")
+
+        registry = MergeDriverRegistry(config)
+        driver = registry.get_driver("xmlmerge")
+
+        self.assertIsInstance(driver, ProcessMergeDriver)
+        self.assertEqual(driver.name, "xmlmerge")
+        self.assertEqual(driver.command, "xmlmerge %O %A %B")
+
+
+class ProcessMergeDriverTests(unittest.TestCase):
+    """Tests for ProcessMergeDriver."""
+
+    def test_merge_with_echo(self):
+        """Test merge driver using echo command."""
+        # Use a simple echo command that writes to the output file
+        command = "echo merged > %A"
+        driver = ProcessMergeDriver(command, "echo_driver")
+
+        ancestor = b"ancestor content"
+        ours = b"our content"
+        theirs = b"their content"
+
+        result, success = driver.merge(ancestor, ours, theirs, "test.txt", 7)
+
+        # Expect different line endings on Windows vs Unix
+        if sys.platform == "win32":
+            expected = b"merged \r\n"
+        else:
+            expected = b"merged\n"
+        self.assertEqual(result, expected)
+        self.assertTrue(success)  # echo returns 0
+
+    def test_merge_with_cat(self):
+        """Test merge driver using cat command."""
+        # Cat all three files together
+        command = "cat %O %B >> %A"
+        driver = ProcessMergeDriver(command, "cat_driver")
+
+        ancestor = b"ancestor\n"
+        ours = b"ours\n"
+        theirs = b"theirs\n"
+
+        result, success = driver.merge(ancestor, ours, theirs)
+
+        self.assertEqual(result, b"ours\nancestor\ntheirs\n")
+        self.assertTrue(success)
+
+    def test_merge_with_failure(self):
+        """Test merge driver that fails."""
+        # Use false command which always returns 1
+        command = "false"
+        driver = ProcessMergeDriver(command, "fail_driver")
+
+        result, success = driver.merge(b"a", b"b", b"c")
+
+        # Should return original content on failure
+        self.assertEqual(result, b"b")
+        self.assertFalse(success)
+
+    def test_merge_with_markers(self):
+        """Test merge driver with conflict marker size."""
+        # Echo the marker size
+        command = "echo marker size: %L > %A"
+        driver = ProcessMergeDriver(command, "marker_driver")
+
+        result, success = driver.merge(b"a", b"b", b"c", marker_size=15)
+
+        # Expect different line endings on Windows vs Unix
+        if sys.platform == "win32":
+            expected = b"marker size: 15 \r\n"
+        else:
+            expected = b"marker size: 15\n"
+        self.assertEqual(result, expected)
+        self.assertTrue(success)
+
+    def test_merge_with_path(self):
+        """Test merge driver with file path."""
+        # Echo the path
+        command = "echo path: %P > %A"
+        driver = ProcessMergeDriver(command, "path_driver")
+
+        result, success = driver.merge(b"a", b"b", b"c", path="dir/file.xml")
+
+        # Expect different line endings on Windows vs Unix
+        if sys.platform == "win32":
+            expected = b"path: dir/file.xml \r\n"
+        else:
+            expected = b"path: dir/file.xml\n"
+        self.assertEqual(result, expected)
+        self.assertTrue(success)
+
+
+class MergeBlobsWithDriversTests(unittest.TestCase):
+    """Tests for merge_blobs with merge drivers."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        # Reset global registry
+        global _merge_driver_registry
+        from dulwich import merge_drivers
+
+        merge_drivers._merge_driver_registry = None
+
+    def test_merge_blobs_without_driver(self):
+        """Test merge_blobs without any merge driver."""
+        base = Blob.from_string(b"base\ncontent\n")
+        ours = Blob.from_string(b"base\nour change\n")
+        theirs = Blob.from_string(b"base\ntheir change\n")
+
+        result, has_conflicts = merge_blobs(base, ours, theirs)
+
+        # Should use default merge and have conflicts
+        self.assertTrue(has_conflicts)
+        self.assertIn(b"<<<<<<< ours", result)
+        self.assertIn(b">>>>>>> theirs", result)
+
+    def test_merge_blobs_with_text_driver(self):
+        """Test merge_blobs with 'text' merge driver (default)."""
+        base = Blob.from_string(b"base\ncontent\n")
+        ours = Blob.from_string(b"base\nour change\n")
+        theirs = Blob.from_string(b"base\ntheir change\n")
+
+        # Set up gitattributes
+        patterns = [(Pattern(b"*.txt"), {b"merge": b"text"})]
+        gitattributes = GitAttributes(patterns)
+
+        result, has_conflicts = merge_blobs(
+            base, ours, theirs, b"file.txt", gitattributes
+        )
+
+        # Should use default merge (text is the default)
+        self.assertTrue(has_conflicts)
+        self.assertIn(b"<<<<<<< ours", result)
+
+    def test_merge_blobs_with_custom_driver(self):
+        """Test merge_blobs with custom merge driver."""
+        # Register a test driver
+        registry = get_merge_driver_registry()
+        test_driver = _TestMergeDriver("custom")
+        registry.register_driver("custom", test_driver)
+
+        base = Blob.from_string(b"base content")
+        ours = Blob.from_string(b"our content")
+        theirs = Blob.from_string(b"their content")
+
+        # Set up gitattributes
+        patterns = [(Pattern(b"*.xml"), {b"merge": b"custom"})]
+        gitattributes = GitAttributes(patterns)
+
+        result, has_conflicts = merge_blobs(
+            base, ours, theirs, b"file.xml", gitattributes
+        )
+
+        # Check that our test driver was called
+        self.assertTrue(test_driver.called)
+        self.assertEqual(test_driver.last_args["ancestor"], b"base content")
+        self.assertEqual(test_driver.last_args["ours"], b"our content")
+        self.assertEqual(test_driver.last_args["theirs"], b"their content")
+        self.assertEqual(test_driver.last_args["path"], "file.xml")
+
+        # Check result
+        self.assertIn(b"TEST MERGE OUTPUT", result)
+        self.assertFalse(
+            has_conflicts
+        )  # Our test driver returns success=True when all differ, so had_conflicts=False
+
+    def test_merge_blobs_with_process_driver(self):
+        """Test merge_blobs with process-based merge driver."""
+        # Set up config with merge driver
+        config = ConfigDict()
+        config.set((b"merge", b"union"), b"driver", b"echo process merge worked > %A")
+
+        base = Blob.from_string(b"base")
+        ours = Blob.from_string(b"ours")
+        theirs = Blob.from_string(b"theirs")
+
+        # Set up gitattributes
+        patterns = [(Pattern(b"*.list"), {b"merge": b"union"})]
+        gitattributes = GitAttributes(patterns)
+
+        result, has_conflicts = merge_blobs(
+            base, ours, theirs, b"file.list", gitattributes, config
+        )
+
+        # Check that the process driver was executed
+        # Expect different line endings on Windows vs Unix
+        if sys.platform == "win32":
+            expected = b"process merge worked \r\n"
+        else:
+            expected = b"process merge worked\n"
+        self.assertEqual(result, expected)
+        self.assertFalse(has_conflicts)  # echo returns 0
+
+    def test_merge_blobs_driver_not_found(self):
+        """Test merge_blobs when specified driver is not found."""
+        base = Blob.from_string(b"base")
+        ours = Blob.from_string(b"ours")
+        theirs = Blob.from_string(b"theirs")
+
+        # Set up gitattributes with non-existent driver
+        patterns = [(Pattern(b"*.dat"), {b"merge": b"nonexistent"})]
+        gitattributes = GitAttributes(patterns)
+
+        result, has_conflicts = merge_blobs(
+            base, ours, theirs, b"file.dat", gitattributes
+        )
+
+        # Should fall back to default merge
+        self.assertTrue(has_conflicts)
+        self.assertIn(b"<<<<<<< ours", result)
+
+
+class GlobalRegistryTests(unittest.TestCase):
+    """Tests for global merge driver registry."""
+
+    def setUp(self):
+        """Reset global registry before each test."""
+        global _merge_driver_registry
+        from dulwich import merge_drivers
+
+        merge_drivers._merge_driver_registry = None
+
+    def test_get_merge_driver_registry_singleton(self):
+        """Test that get_merge_driver_registry returns singleton."""
+        registry1 = get_merge_driver_registry()
+        registry2 = get_merge_driver_registry()
+
+        self.assertIs(registry1, registry2)
+
+    def test_get_merge_driver_registry_with_config(self):
+        """Test updating config on existing registry."""
+        # Get registry without config
+        registry = get_merge_driver_registry()
+        self.assertIsNone(registry._config)
+
+        # Get with config
+        config = ConfigDict()
+        registry2 = get_merge_driver_registry(config)
+
+        self.assertIs(registry2, registry)
+        self.assertIsNotNone(registry._config)
+
+
+if __name__ == "__main__":
+    unittest.main()