Move tests for merge base out into unittest classes.

dulwich/merge.py

@@ -33,6 +33,7 @@ from .diff_tree import (
     CHANGE_RENAME,
     CHANGE_UNCHANGED,
     )
+from .merge_base import find_merge_base
 from .objects import Blob
 
 
@@ -42,21 +43,6 @@ class MergeConflict(namedtuple(
     """A merge conflict."""
 
 
-def find_merge_base(repo, commit_ids):
-    """Find a reasonable merge base.
-
-    Args:
-      repo: Repository object
-      commit_ids: List of commit ids
-    """
-    # TODO(jelmer): replace with a pure-python implementation
-    import subprocess
-    return subprocess.check_output(
-        ['git', 'merge-base'] +
-        [x.decode('ascii') for x in commit_ids],
-        cwd=repo.path).rstrip(b'\n')
-
-
 def _merge_entry(new_path, object_store, this_entry, other_entry, base_entry,
                  file_merger):
     """Run a per entry-merge."""
@@ -190,7 +176,7 @@ def merge(repo, commit_ids, rename_detector=None, file_merger=None):
     """Perform a merge.
     """
     conflicts = []
-    merge_base = find_merge_base(repo, [repo.head()] + commit_ids)
+    [merge_base] = find_merge_base(repo, [repo.head()] + commit_ids)
     [other_id] = commit_ids
     index = repo.open_index()
     this_tree = index.commit(repo.object_store)

dulwich/merge_base.py

@@ -8,7 +8,6 @@ Implementation of merge-base following the approach of git
 #
 # Available under the MIT License
 
-import sys
 from collections import deque
 
 
@@ -68,17 +67,17 @@ def _find_lcas(lookup_parents, c1, c2s):
     return results
 
 
-def find_merge_base(r, commit_ids):
-    """ find lowest common ancestors of commit_ids[0] and *any* of commits_ids[1:]
-       ARGS:
-          r: Repo object
-          commit_ids:  list of commit ids
-       Returns
-          list of lowest common ancestor commit_ids
-    """
+def find_merge_base(object_store, commit_ids):
+    """Find lowest common ancestors of commit_ids[0] and *any* of commits_ids[1:]
 
+    Args:
+      object_store: object store
+      commit_ids:  list of commit ids
+    Returns:
+      list of lowest common ancestor commit_ids
+    """
     def lookup_parents(commit_id):
-        return r.object_store[commit_id].parents
+        return object_store[commit_id].parents
 
     if not commit_ids:
         return []
@@ -91,22 +90,23 @@ def find_merge_base(r, commit_ids):
     return _find_lcas(lookup_parents, c1, c2s)
 
 
-def find_octopus_base(r, commit_ids):
-    """ find lowest common ancestors of *all* provided commit_ids
-       ARGS:
-          r: Repo object
-          commit_ids:  list of commit ids
-       Returns
-          list of lowest common ancestor commit_ids
+def find_octopus_base(object_store, commit_ids):
+    """Find lowest common ancestors of *all* provided commit_ids
+
+    Args:
+      object_store: Object store
+      commit_ids:  list of commit ids
+    Returns:
+      list of lowest common ancestor commit_ids
     """
 
     def lookup_parents(commit_id):
-        return r.object_store[commit_id].parents
+        return object_store[commit_id].parents
 
     if not commit_ids:
         return []
     if len(commit_ids) <= 2:
-        return find_merge_base(r, commit_ids)
+        return find_merge_base(object_store, commit_ids)
     lcas = [commit_ids[0]]
     others = commit_ids[1:]
     for cmt in others:
@@ -116,153 +116,3 @@ def find_octopus_base(r, commit_ids):
             next_lcas.extend(res)
         lcas = next_lcas[:]
     return lcas
-
-
-def test():
-
-    all_tests_passed = True
-
-    parents = {}
-
-    def lookup_parents(commit_id):
-        return parents.get(commit_id, [])
-
-    def run_test(dag, inputs, expected):
-        nonlocal parents
-        parents = dag
-        c1 = inputs[0]
-        c2s = inputs[1:]
-        res = _find_lcas(lookup_parents, c1, c2s)
-        return set(res) == expected
-
-    # two lowest common ancestors
-    test1 = {
-        '5': ['1', '2'],
-        '4': ['3', '1'],
-        '3': ['2'],
-        '2': ['0'],
-        '1': [],
-        '0': []
-    }
-    test_passed = run_test(test1, ['4', '5'], set(['1', '2']))
-    print('Test 1: Multiple LCA ', test_passed)
-    all_tests_passed = all_tests_passed and test_passed
-
-    # no common ancestor
-    test2 = {
-        '4': ['2'],
-        '3': ['1'],
-        '2': [],
-        '1': ['0'],
-        '0': [],
-    }
-    test_passed = run_test(test2, ['4', '3'], set([]))
-    print('Test 2: No Common Ancestor ', test_passed)
-    all_tests_passed = all_tests_passed and test_passed
-
-    # ancestor
-    test3 = {
-        'G': ['D', 'F'],
-        'F': ['E'],
-        'D': ['C'],
-        'C': ['B'],
-        'E': ['B'],
-        'B': ['A'],
-        'A': []
-    }
-    test_passed = run_test(test3, ['D', 'C'], set(['C']))
-    print('Test 3: Ancestor ', test_passed)
-    all_tests_passed = all_tests_passed and test_passed
-
-    # parent
-    test4 = {
-        'G': ['D', 'F'],
-        'F': ['E'],
-        'D': ['C'],
-        'C': ['B'],
-        'E': ['B'],
-        'B': ['A'],
-        'A': []
-    }
-    test_passed = run_test(test4, ['G', 'D'], set(['D']))
-    print('Test 4: Direct Parent ', test_passed)
-    all_tests_passed = all_tests_passed and test_passed
-
-    # Another cross over
-    test5 = {
-        'G': ['D', 'F'],
-        'F': ['E', 'C'],
-        'D': ['C', 'E'],
-        'C': ['B'],
-        'E': ['B'],
-        'B': ['A'],
-        'A': []
-    }
-    test_passed = run_test(test5, ['D', 'F'], set(['E', 'C']))
-    print('Test 5: Cross Over ', test_passed)
-    all_tests_passed = all_tests_passed and test_passed
-
-    # three way merge commit straight from git docs
-    test6 = {
-        'C': ['C1'],
-        'C1': ['C2'],
-        'C2': ['C3'],
-        'C3': ['C4'],
-        'C4': ['2'],
-        'B': ['B1'],
-        'B1': ['B2'],
-        'B2': ['B3'],
-        'B3': ['1'],
-        'A': ['A1'],
-        'A1': ['A2'],
-        'A2': ['A3'],
-        'A3': ['1'],
-        '1': ['2'],
-        '2': [],
-    }
-    # assumes a theoretical merge M exists that merges B and C first
-    # which actually means find the first LCA from either of B OR C with A
-    test_passed = run_test(test6, ['A', 'B', 'C'], set(['1']))
-    all_tests_passed = all_tests_passed and test_passed
-    print('Test 6: LCA of 3 commits ', test_passed)
-
-    # octopus algorithm test
-    # test straight from git docs of A, B, and C
-    # but this time use octopus to find lcas of A, B, and C simultaneously
-    test7 = {
-        'C': ['C1'],
-        'C1': ['C2'],
-        'C2': ['C3'],
-        'C3': ['C4'],
-        'C4': ['2'],
-        'B': ['B1'],
-        'B1': ['B2'],
-        'B2': ['B3'],
-        'B3': ['1'],
-        'A': ['A1'],
-        'A1': ['A2'],
-        'A2': ['A3'],
-        'A3': ['1'],
-        '1': ['2'],
-        '2': [],
-    }
-    parents = test7
-    lcas = ['A']
-    others = ['B', 'C']
-    for cmt in others:
-        next_lcas = []
-        for ca in lcas:
-            res = _find_lcas(lookup_parents, cmt, [ca])
-            next_lcas.extend(res)
-        lcas = next_lcas[:]
-    test_passed = set(lcas) == set(['2'])
-    all_tests_passed = all_tests_passed and test_passed
-    print('Test 7: Octopus LCA of 3 commits ', test_passed)
-
-    if all_tests_passed:
-        print('All Tests Succesfful')
-    return 0
-
-
-if __name__ == '__main__':
-    sys.exit(test())

dulwich/tests/__init__.py

@@ -122,6 +122,7 @@ def self_test_suite():
         "line_ending",
         "lru_cache",
         "mailmap",
+        'merge_base',
         "merge",
         "objects",
         "objectspec",

dulwich/tests/test_merge.py

@@ -76,10 +76,10 @@ class FindMergeBaseTests(TestCase):
         c1, c2, c3 = build_commit_graph(
                 self.repo.object_store, [[1], [2, 1], [3, 1]])
         self.assertEqual(
-            c1.id, find_merge_base(self.repo.object_store, [c2.id, c3.id]))
+            [c1.id], find_merge_base(self.repo.object_store, [c2.id, c3.id]))
 
     def test_fast_forward(self):
         c1, c2, c3 = build_commit_graph(
                 self.repo.object_store, [[1], [2, 1], [3, 2]])
         self.assertEqual(
-            c2.id, find_merge_base(self.repo.object_store, [c2.id, c3.id]))
+            [c2.id], find_merge_base(self.repo.object_store, [c2.id, c3.id]))

dulwich/tests/test_merge_base.py

@@ -0,0 +1,141 @@
+# -*- coding: utf-8 -*-
+# test_index.py -- Tests for merge
+# encoding: utf-8
+# Copyright (c) 2020 Kevin B. Hendricks, Stratford Ontario Canada
+# Available under the MIT License
+
+"""Tests for merge_base."""
+
+from dulwich.tests import TestCase
+
+from dulwich.merge_base import _find_lcas
+
+
+class FindMergeBaseTests(TestCase):
+
+    @staticmethod
+    def run_test(dag, inputs):
+        def lookup_parents(commit_id):
+            return dag[commit_id]
+        c1 = inputs[0]
+        c2s = inputs[1:]
+        return set(_find_lcas(lookup_parents, c1, c2s))
+
+    def test_multiple_lca(self):
+        # two lowest common ancestors
+        graph = {
+            '5': ['1', '2'],
+            '4': ['3', '1'],
+            '3': ['2'],
+            '2': ['0'],
+            '1': [],
+            '0': []
+        }
+        self.assertEqual(self.run_test(graph, ['4', '5']), set(['1', '2']))
+
+    def test_no_common_ancestor(self):
+        # no common ancestor
+        graph = {
+            '4': ['2'],
+            '3': ['1'],
+            '2': [],
+            '1': ['0'],
+            '0': [],
+        }
+        self.assertEqual(self.run_test(graph, ['4', '3']), set([]))
+
+    def test_ancestor(self):
+        # ancestor
+        graph = {
+            'G': ['D', 'F'],
+            'F': ['E'],
+            'D': ['C'],
+            'C': ['B'],
+            'E': ['B'],
+            'B': ['A'],
+            'A': []
+        }
+        self.assertEqual(self.run_test(graph, ['D', 'C']), set(['C']))
+
+    def test_direct_parent(self):
+        # parent
+        graph = {
+            'G': ['D', 'F'],
+            'F': ['E'],
+            'D': ['C'],
+            'C': ['B'],
+            'E': ['B'],
+            'B': ['A'],
+            'A': []
+        }
+        self.assertEqual(self.run_test(graph, ['G', 'D']), set(['D']))
+
+    def test_another_crossover(self):
+        # Another cross over
+        graph = {
+            'G': ['D', 'F'],
+            'F': ['E', 'C'],
+            'D': ['C', 'E'],
+            'C': ['B'],
+            'E': ['B'],
+            'B': ['A'],
+            'A': []
+        }
+        self.assertEqual(self.run_test(graph, ['D', 'F']), set(['E', 'C']))
+
+    def test_three_way_merge_lca(self):
+        # three way merge commit straight from git docs
+        graph = {
+            'C': ['C1'],
+            'C1': ['C2'],
+            'C2': ['C3'],
+            'C3': ['C4'],
+            'C4': ['2'],
+            'B': ['B1'],
+            'B1': ['B2'],
+            'B2': ['B3'],
+            'B3': ['1'],
+            'A': ['A1'],
+            'A1': ['A2'],
+            'A2': ['A3'],
+            'A3': ['1'],
+            '1': ['2'],
+            '2': [],
+        }
+        # assumes a theoretical merge M exists that merges B and C first
+        # which actually means find the first LCA from either of B OR C with A
+        self.assertEqual(self.run_test(graph, ['A', 'B', 'C']), set(['1']))
+
+    def test_octopus(self):
+        # octopus algorithm test
+        # test straight from git docs of A, B, and C
+        # but this time use octopus to find lcas of A, B, and C simultaneously
+        graph = {
+            'C': ['C1'],
+            'C1': ['C2'],
+            'C2': ['C3'],
+            'C3': ['C4'],
+            'C4': ['2'],
+            'B': ['B1'],
+            'B1': ['B2'],
+            'B2': ['B3'],
+            'B3': ['1'],
+            'A': ['A1'],
+            'A1': ['A2'],
+            'A2': ['A3'],
+            'A3': ['1'],
+            '1': ['2'],
+            '2': [],
+        }
+
+        def lookup_parents(cid):
+            return graph[cid]
+        lcas = ['A']
+        others = ['B', 'C']
+        for cmt in others:
+            next_lcas = []
+            for ca in lcas:
+                res = _find_lcas(lookup_parents, cmt, [ca])
+                next_lcas.extend(res)
+            lcas = next_lcas[:]
+        self.assertEqual(set(lcas), set(['2']))