"""Tests for merge functionality.""" import unittest from dulwich.merge import MergeConflict, Merger, three_way_merge from dulwich.objects import Blob, Commit, Tree from dulwich.repo import MemoryRepo class MergeTests(unittest.TestCase): """Tests for merge functionality.""" def setUp(self): self.repo = MemoryRepo() 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 " base_commit.committer = b"Test Author " 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 " ours_commit.committer = b"Test Author " 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 " theirs_commit.committer = b"Test Author " 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))