dulwich/tests/test_merge.py

"""Tests for merge functionality."""

import importlib.util
import unittest

from dulwich.merge import MergeConflict, Merger, three_way_merge
from dulwich.objects import Blob, Commit, Tree
from dulwich.repo import MemoryRepo

from . import DependencyMissing


class MergeTests(unittest.TestCase):
    """Tests for merge functionality."""

    def setUp(self):
        self.repo = MemoryRepo()
        # Check if merge3 module is available
        if importlib.util.find_spec("merge3") is None:
            raise DependencyMissing("merge3")
        self.merger = Merger(self.repo.object_store)

    def test_merge_blobs_no_conflict(self):
        """Test merging blobs without conflicts."""
        # Create base blob
        base_blob = Blob.from_string(b"line1\nline2\nline3\n")

        # Create modified versions - currently our algorithm treats changes to different line groups as conflicts
        # This is a simple implementation - Git's merge is more sophisticated
        ours_blob = Blob.from_string(b"line1\nmodified line2\nline3\n")
        theirs_blob = Blob.from_string(b"line1\nline2\nmodified line3\n")

        # Add blobs to object store
        self.repo.object_store.add_object(base_blob)
        self.repo.object_store.add_object(ours_blob)
        self.repo.object_store.add_object(theirs_blob)

        # Merge - this will result in a conflict with our simple algorithm
        result, has_conflicts = self.merger.merge_blobs(
            base_blob, ours_blob, theirs_blob
        )

        # For now, expect conflicts since both sides changed (even different lines)
        self.assertTrue(has_conflicts)
        self.assertIn(b"<<<<<<< ours", result)
        self.assertIn(b">>>>>>> theirs", result)

    def test_merge_blobs_clean_merge(self):
        """Test merging blobs with a clean merge (one side unchanged)."""
        # Create base blob
        base_blob = Blob.from_string(b"line1\nline2\nline3\n")

        # Only ours modifies
        ours_blob = Blob.from_string(b"line1\nmodified line2\nline3\n")
        theirs_blob = base_blob  # unchanged

        # Add blobs to object store
        self.repo.object_store.add_object(base_blob)
        self.repo.object_store.add_object(ours_blob)

        # Merge
        result, has_conflicts = self.merger.merge_blobs(
            base_blob, ours_blob, theirs_blob
        )

        self.assertFalse(has_conflicts)
        self.assertEqual(result, b"line1\nmodified line2\nline3\n")

    def test_merge_blobs_with_conflict(self):
        """Test merging blobs with conflicts."""
        # Create base blob
        base_blob = Blob.from_string(b"line1\nline2\nline3\n")

        # Create conflicting modifications
        ours_blob = Blob.from_string(b"line1\nours line2\nline3\n")
        theirs_blob = Blob.from_string(b"line1\ntheirs line2\nline3\n")

        # Add blobs to object store
        self.repo.object_store.add_object(base_blob)
        self.repo.object_store.add_object(ours_blob)
        self.repo.object_store.add_object(theirs_blob)

        # Merge
        result, has_conflicts = self.merger.merge_blobs(
            base_blob, ours_blob, theirs_blob
        )

        self.assertTrue(has_conflicts)
        self.assertIn(b"<<<<<<< ours", result)
        self.assertIn(b"=======", result)
        self.assertIn(b">>>>>>> theirs", result)

    def test_merge_blobs_identical(self):
        """Test merging identical blobs."""
        blob = Blob.from_string(b"same content\n")
        self.repo.object_store.add_object(blob)

        result, has_conflicts = self.merger.merge_blobs(blob, blob, blob)

        self.assertFalse(has_conflicts)
        self.assertEqual(result, b"same content\n")

    def test_merge_blobs_one_side_unchanged(self):
        """Test merging when one side is unchanged."""
        base_blob = Blob.from_string(b"original\n")
        modified_blob = Blob.from_string(b"modified\n")

        self.repo.object_store.add_object(base_blob)
        self.repo.object_store.add_object(modified_blob)

        # Test ours unchanged, theirs modified
        result, has_conflicts = self.merger.merge_blobs(
            base_blob, base_blob, modified_blob
        )
        self.assertFalse(has_conflicts)
        self.assertEqual(result, b"modified\n")

        # Test theirs unchanged, ours modified
        result, has_conflicts = self.merger.merge_blobs(
            base_blob, modified_blob, base_blob
        )
        self.assertFalse(has_conflicts)
        self.assertEqual(result, b"modified\n")

    def test_merge_blobs_deletion_no_conflict(self):
        """Test merging with deletion where no conflict occurs."""
        base_blob = Blob.from_string(b"content\n")
        self.repo.object_store.add_object(base_blob)

        # Both delete
        result, has_conflicts = self.merger.merge_blobs(base_blob, None, None)
        self.assertFalse(has_conflicts)
        self.assertEqual(result, b"")

        # One deletes, other unchanged
        result, has_conflicts = self.merger.merge_blobs(base_blob, None, base_blob)
        self.assertFalse(has_conflicts)
        self.assertEqual(result, b"")

    def test_merge_blobs_deletion_with_conflict(self):
        """Test merging with deletion that causes conflict."""
        base_blob = Blob.from_string(b"content\n")
        modified_blob = Blob.from_string(b"modified content\n")

        self.repo.object_store.add_object(base_blob)
        self.repo.object_store.add_object(modified_blob)

        # We delete, they modify
        _result, has_conflicts = self.merger.merge_blobs(base_blob, None, modified_blob)
        self.assertTrue(has_conflicts)

    def test_merge_blobs_no_base(self):
        """Test merging blobs with no common ancestor."""
        blob1 = Blob.from_string(b"content1\n")
        blob2 = Blob.from_string(b"content2\n")

        self.repo.object_store.add_object(blob1)
        self.repo.object_store.add_object(blob2)

        # Different content added in both - conflict
        result, has_conflicts = self.merger.merge_blobs(None, blob1, blob2)
        self.assertTrue(has_conflicts)

        # Same content added in both - no conflict
        result, has_conflicts = self.merger.merge_blobs(None, blob1, blob1)
        self.assertFalse(has_conflicts)
        self.assertEqual(result, b"content1\n")

    def test_merge_trees_simple(self):
        """Test simple tree merge."""
        # Create base tree
        base_tree = Tree()
        blob1 = Blob.from_string(b"file1 content\n")
        blob2 = Blob.from_string(b"file2 content\n")
        self.repo.object_store.add_object(blob1)
        self.repo.object_store.add_object(blob2)
        base_tree.add(b"file1.txt", 0o100644, blob1.id)
        base_tree.add(b"file2.txt", 0o100644, blob2.id)
        self.repo.object_store.add_object(base_tree)

        # Create ours tree (modify file1)
        ours_tree = Tree()
        ours_blob1 = Blob.from_string(b"file1 modified by ours\n")
        self.repo.object_store.add_object(ours_blob1)
        ours_tree.add(b"file1.txt", 0o100644, ours_blob1.id)
        ours_tree.add(b"file2.txt", 0o100644, blob2.id)
        self.repo.object_store.add_object(ours_tree)

        # Create theirs tree (modify file2)
        theirs_tree = Tree()
        theirs_blob2 = Blob.from_string(b"file2 modified by theirs\n")
        self.repo.object_store.add_object(theirs_blob2)
        theirs_tree.add(b"file1.txt", 0o100644, blob1.id)
        theirs_tree.add(b"file2.txt", 0o100644, theirs_blob2.id)
        self.repo.object_store.add_object(theirs_tree)

        # Merge
        merged_tree, conflicts = self.merger.merge_trees(
            base_tree, ours_tree, theirs_tree
        )

        self.assertEqual(len(conflicts), 0)
        self.assertIn(b"file1.txt", [item.path for item in merged_tree.items()])
        self.assertIn(b"file2.txt", [item.path for item in merged_tree.items()])

    def test_merge_trees_with_conflict(self):
        """Test tree merge with conflicting changes."""
        # Create base tree
        base_tree = Tree()
        blob1 = Blob.from_string(b"original content\n")
        self.repo.object_store.add_object(blob1)
        base_tree.add(b"conflict.txt", 0o100644, blob1.id)
        self.repo.object_store.add_object(base_tree)

        # Create ours tree
        ours_tree = Tree()
        ours_blob = Blob.from_string(b"ours content\n")
        self.repo.object_store.add_object(ours_blob)
        ours_tree.add(b"conflict.txt", 0o100644, ours_blob.id)
        self.repo.object_store.add_object(ours_tree)

        # Create theirs tree
        theirs_tree = Tree()
        theirs_blob = Blob.from_string(b"theirs content\n")
        self.repo.object_store.add_object(theirs_blob)
        theirs_tree.add(b"conflict.txt", 0o100644, theirs_blob.id)
        self.repo.object_store.add_object(theirs_tree)

        # Merge
        _merged_tree, conflicts = self.merger.merge_trees(
            base_tree, ours_tree, theirs_tree
        )

        self.assertEqual(len(conflicts), 1)
        self.assertEqual(conflicts[0], b"conflict.txt")

    def test_three_way_merge(self):
        """Test three-way merge between commits."""
        # Create base commit
        base_tree = Tree()
        blob = Blob.from_string(b"base content\n")
        self.repo.object_store.add_object(blob)
        base_tree.add(b"file.txt", 0o100644, blob.id)
        self.repo.object_store.add_object(base_tree)

        base_commit = Commit()
        base_commit.tree = base_tree.id
        base_commit.author = b"Test Author <test@example.com>"
        base_commit.committer = b"Test Author <test@example.com>"
        base_commit.message = b"Base commit"
        base_commit.commit_time = base_commit.author_time = 12345
        base_commit.commit_timezone = base_commit.author_timezone = 0
        self.repo.object_store.add_object(base_commit)

        # Create ours commit
        ours_tree = Tree()
        ours_blob = Blob.from_string(b"ours content\n")
        self.repo.object_store.add_object(ours_blob)
        ours_tree.add(b"file.txt", 0o100644, ours_blob.id)
        self.repo.object_store.add_object(ours_tree)

        ours_commit = Commit()
        ours_commit.tree = ours_tree.id
        ours_commit.parents = [base_commit.id]
        ours_commit.author = b"Test Author <test@example.com>"
        ours_commit.committer = b"Test Author <test@example.com>"
        ours_commit.message = b"Ours commit"
        ours_commit.commit_time = ours_commit.author_time = 12346
        ours_commit.commit_timezone = ours_commit.author_timezone = 0
        self.repo.object_store.add_object(ours_commit)

        # Create theirs commit
        theirs_tree = Tree()
        theirs_blob = Blob.from_string(b"theirs content\n")
        self.repo.object_store.add_object(theirs_blob)
        theirs_tree.add(b"file.txt", 0o100644, theirs_blob.id)
        self.repo.object_store.add_object(theirs_tree)

        theirs_commit = Commit()
        theirs_commit.tree = theirs_tree.id
        theirs_commit.parents = [base_commit.id]
        theirs_commit.author = b"Test Author <test@example.com>"
        theirs_commit.committer = b"Test Author <test@example.com>"
        theirs_commit.message = b"Theirs commit"
        theirs_commit.commit_time = theirs_commit.author_time = 12347
        theirs_commit.commit_timezone = theirs_commit.author_timezone = 0
        self.repo.object_store.add_object(theirs_commit)

        # Perform three-way merge
        _merged_tree, conflicts = three_way_merge(
            self.repo.object_store, base_commit, ours_commit, theirs_commit
        )

        # Should have conflict since both modified the same file differently
        self.assertEqual(len(conflicts), 1)
        self.assertEqual(conflicts[0], b"file.txt")

    def test_merge_exception(self):
        """Test MergeConflict exception."""
        exc = MergeConflict(b"test/path", "test message")
        self.assertEqual(exc.path, b"test/path")
        self.assertIn("test/path", str(exc))
        self.assertIn("test message", str(exc))


class OctopusMergeTests(unittest.TestCase):
    """Tests for octopus merge functionality."""

    def setUp(self):
        self.repo = MemoryRepo()
        # Check if merge3 module is available
        if importlib.util.find_spec("merge3") is None:
            raise DependencyMissing("merge3")

    def test_octopus_merge_three_branches(self):
        """Test octopus merge with three branches."""
        from dulwich.merge import octopus_merge

        # Create base commit
        base_tree = Tree()
        blob1 = Blob.from_string(b"file1 content\n")
        blob2 = Blob.from_string(b"file2 content\n")
        blob3 = Blob.from_string(b"file3 content\n")
        self.repo.object_store.add_object(blob1)
        self.repo.object_store.add_object(blob2)
        self.repo.object_store.add_object(blob3)
        base_tree.add(b"file1.txt", 0o100644, blob1.id)
        base_tree.add(b"file2.txt", 0o100644, blob2.id)
        base_tree.add(b"file3.txt", 0o100644, blob3.id)
        self.repo.object_store.add_object(base_tree)

        base_commit = Commit()
        base_commit.tree = base_tree.id
        base_commit.author = b"Test <test@example.com>"
        base_commit.committer = b"Test <test@example.com>"
        base_commit.message = b"Base commit"
        base_commit.commit_time = base_commit.author_time = 12345
        base_commit.commit_timezone = base_commit.author_timezone = 0
        self.repo.object_store.add_object(base_commit)

        # Create HEAD commit (modifies file1)
        head_tree = Tree()
        head_blob1 = Blob.from_string(b"file1 modified by head\n")
        self.repo.object_store.add_object(head_blob1)
        head_tree.add(b"file1.txt", 0o100644, head_blob1.id)
        head_tree.add(b"file2.txt", 0o100644, blob2.id)
        head_tree.add(b"file3.txt", 0o100644, blob3.id)
        self.repo.object_store.add_object(head_tree)

        head_commit = Commit()
        head_commit.tree = head_tree.id
        head_commit.parents = [base_commit.id]
        head_commit.author = b"Test <test@example.com>"
        head_commit.committer = b"Test <test@example.com>"
        head_commit.message = b"Head commit"
        head_commit.commit_time = head_commit.author_time = 12346
        head_commit.commit_timezone = head_commit.author_timezone = 0
        self.repo.object_store.add_object(head_commit)

        # Create branch1 commit (modifies file2)
        branch1_tree = Tree()
        branch1_blob2 = Blob.from_string(b"file2 modified by branch1\n")
        self.repo.object_store.add_object(branch1_blob2)
        branch1_tree.add(b"file1.txt", 0o100644, blob1.id)
        branch1_tree.add(b"file2.txt", 0o100644, branch1_blob2.id)
        branch1_tree.add(b"file3.txt", 0o100644, blob3.id)
        self.repo.object_store.add_object(branch1_tree)

        branch1_commit = Commit()
        branch1_commit.tree = branch1_tree.id
        branch1_commit.parents = [base_commit.id]
        branch1_commit.author = b"Test <test@example.com>"
        branch1_commit.committer = b"Test <test@example.com>"
        branch1_commit.message = b"Branch1 commit"
        branch1_commit.commit_time = branch1_commit.author_time = 12347
        branch1_commit.commit_timezone = branch1_commit.author_timezone = 0
        self.repo.object_store.add_object(branch1_commit)

        # Create branch2 commit (modifies file3)
        branch2_tree = Tree()
        branch2_blob3 = Blob.from_string(b"file3 modified by branch2\n")
        self.repo.object_store.add_object(branch2_blob3)
        branch2_tree.add(b"file1.txt", 0o100644, blob1.id)
        branch2_tree.add(b"file2.txt", 0o100644, blob2.id)
        branch2_tree.add(b"file3.txt", 0o100644, branch2_blob3.id)
        self.repo.object_store.add_object(branch2_tree)

        branch2_commit = Commit()
        branch2_commit.tree = branch2_tree.id
        branch2_commit.parents = [base_commit.id]
        branch2_commit.author = b"Test <test@example.com>"
        branch2_commit.committer = b"Test <test@example.com>"
        branch2_commit.message = b"Branch2 commit"
        branch2_commit.commit_time = branch2_commit.author_time = 12348
        branch2_commit.commit_timezone = branch2_commit.author_timezone = 0
        self.repo.object_store.add_object(branch2_commit)

        # Perform octopus merge
        merged_tree, conflicts = octopus_merge(
            self.repo.object_store,
            [base_commit.id],
            head_commit,
            [branch1_commit, branch2_commit],
        )

        # Should have no conflicts since each branch modified different files
        self.assertEqual(len(conflicts), 0)

        # Check that all three modifications are in the merged tree
        self.assertIn(b"file1.txt", [item.path for item in merged_tree.items()])
        self.assertIn(b"file2.txt", [item.path for item in merged_tree.items()])
        self.assertIn(b"file3.txt", [item.path for item in merged_tree.items()])

    def test_octopus_merge_with_conflict(self):
        """Test that octopus merge refuses to proceed with conflicts."""
        from dulwich.merge import octopus_merge

        # Create base commit
        base_tree = Tree()
        blob1 = Blob.from_string(b"original content\n")
        self.repo.object_store.add_object(blob1)
        base_tree.add(b"file.txt", 0o100644, blob1.id)
        self.repo.object_store.add_object(base_tree)

        base_commit = Commit()
        base_commit.tree = base_tree.id
        base_commit.author = b"Test <test@example.com>"
        base_commit.committer = b"Test <test@example.com>"
        base_commit.message = b"Base commit"
        base_commit.commit_time = base_commit.author_time = 12345
        base_commit.commit_timezone = base_commit.author_timezone = 0
        self.repo.object_store.add_object(base_commit)

        # Create HEAD commit
        head_tree = Tree()
        head_blob = Blob.from_string(b"head content\n")
        self.repo.object_store.add_object(head_blob)
        head_tree.add(b"file.txt", 0o100644, head_blob.id)
        self.repo.object_store.add_object(head_tree)

        head_commit = Commit()
        head_commit.tree = head_tree.id
        head_commit.parents = [base_commit.id]
        head_commit.author = b"Test <test@example.com>"
        head_commit.committer = b"Test <test@example.com>"
        head_commit.message = b"Head commit"
        head_commit.commit_time = head_commit.author_time = 12346
        head_commit.commit_timezone = head_commit.author_timezone = 0
        self.repo.object_store.add_object(head_commit)

        # Create branch1 commit (conflicts with head)
        branch1_tree = Tree()
        branch1_blob = Blob.from_string(b"branch1 content\n")
        self.repo.object_store.add_object(branch1_blob)
        branch1_tree.add(b"file.txt", 0o100644, branch1_blob.id)
        self.repo.object_store.add_object(branch1_tree)

        branch1_commit = Commit()
        branch1_commit.tree = branch1_tree.id
        branch1_commit.parents = [base_commit.id]
        branch1_commit.author = b"Test <test@example.com>"
        branch1_commit.committer = b"Test <test@example.com>"
        branch1_commit.message = b"Branch1 commit"
        branch1_commit.commit_time = branch1_commit.author_time = 12347
        branch1_commit.commit_timezone = branch1_commit.author_timezone = 0
        self.repo.object_store.add_object(branch1_commit)

        # Perform octopus merge
        _merged_tree, conflicts = octopus_merge(
            self.repo.object_store,
            [base_commit.id],
            head_commit,
            [branch1_commit],
        )

        # Should have conflicts and refuse to merge
        self.assertEqual(len(conflicts), 1)
        self.assertEqual(conflicts[0], b"file.txt")

    def test_octopus_merge_no_commits(self):
        """Test that octopus merge raises error with no commits to merge."""
        from dulwich.merge import octopus_merge

        # Create a simple commit
        tree = Tree()
        blob = Blob.from_string(b"content\n")
        self.repo.object_store.add_object(blob)
        tree.add(b"file.txt", 0o100644, blob.id)
        self.repo.object_store.add_object(tree)

        commit = Commit()
        commit.tree = tree.id
        commit.author = b"Test <test@example.com>"
        commit.committer = b"Test <test@example.com>"
        commit.message = b"Commit"
        commit.commit_time = commit.author_time = 12345
        commit.commit_timezone = commit.author_timezone = 0
        self.repo.object_store.add_object(commit)

        # Try to do octopus merge with no commits
        with self.assertRaises(ValueError):
            octopus_merge(
                self.repo.object_store,
                [commit.id],
                commit,
                [],
            )