tests: Move write_pack_data to utils.build_pack.

Change-Id: I53aab57e17c00bb5d8457d1423b687bc37e1c688

NEWS

@@ -47,6 +47,9 @@
   * If setuptools is installed, "python setup.py test" will now run the testsuite.
     (Jelmer Vernooij)
 
+  * Add a new build_pack test utility for building packs from a simple spec.
+    (Dave Borowitz)
+
 0.7.1	2011-04-12
 
  BUG FIXES

dulwich/tests/test_pack.py

@@ -49,10 +49,8 @@ from dulwich.objects import (
 from dulwich.pack import (
     OFS_DELTA,
     REF_DELTA,
-    DELTA_TYPES,
     MemoryPackIndex,
     Pack,
-    obj_sha,
     PackData,
     ThinPackData,
     apply_delta,
@@ -74,6 +72,7 @@ from dulwich.tests import (
     )
 from utils import (
     make_object,
+    build_pack,
     )
 
 pack1_sha = 'bc63ddad95e7321ee734ea11a7a62d314e0d7481'
@@ -621,6 +620,7 @@ class TestPackIterator(DeltaChainIterator):
         self._unpacked = set()
 
     def _result(self, offset, type_num, chunks, sha, crc32):
+        """Return entries in the same format as build_pack."""
         return offset, type_num, ''.join(chunks), sha, crc32
 
     def _resolve_object(self, offset, base_type_num, base_chunks):
@@ -646,75 +646,6 @@ class DeltaChainIteratorTests(TestCase):
             self.store.add_object(blob)
         return blobs
 
-    def write_pack_data(self, objects_spec):
-        """Write test pack data from a concise spec.
-
-        :param objects_spec: A list of (type_num, obj). For non-delta types,
-            obj is the string of that object's data.
-            For delta types, obj is a tuple of (base_index, data), where
-            base_index is the index in objects_spec of the base for that delta,
-            and data is the full, non-deltified data for that object.
-            (Offsets/refs and deltas are computed within this function.)
-        :return: A tuple of (f, entries), where f is a file-like object
-            pointed at the beginning of a pack with the requested data, and
-            entries is a list of tuples of (offset, type num, data, sha, CRC32)
-            These tuples match the result format from TestPackIterator, and are
-            returned in the order specified by objects_spec.
-        """
-        f = StringIO()
-        sf = SHA1Writer(f)
-        num_objects = len(objects_spec)
-        write_pack_header(sf, num_objects)
-
-        full_objects = {}
-        offsets = {}
-        crc32s = {}
-
-        while len(full_objects) < num_objects:
-            for i, (type_num, data) in enumerate(objects_spec):
-                if type_num not in DELTA_TYPES:
-                    full_objects[i] = (type_num, data,
-                                       obj_sha(type_num, [data]))
-                    continue
-                base, data = data
-                if isinstance(base, int):
-                    if base not in full_objects:
-                        continue
-                    base_type_num, _, _ = full_objects[base]
-                else:
-                    base_type_num, _ = self.store.get_raw(base)
-                full_objects[i] = (base_type_num, data,
-                                   obj_sha(base_type_num, [data]))
-
-        for i, (type_num, obj) in enumerate(objects_spec):
-            offset = f.tell()
-            if type_num == OFS_DELTA:
-                base_index, data = obj
-                base = offset - offsets[base_index]
-                _, base_data, _ = full_objects[base_index]
-                obj = (base, create_delta(base_data, data))
-            elif type_num == REF_DELTA:
-                base_ref, data = obj
-                if isinstance(base_ref, int):
-                    _, base_data, base = full_objects[base_ref]
-                else:
-                    base_type_num, base_data = self.store.get_raw(base_ref)
-                    base = obj_sha(base_type_num, base_data)
-                obj = (base, create_delta(base_data, data))
-
-            crc32 = write_pack_object(sf, type_num, obj)
-            offsets[i] = offset
-            crc32s[i] = crc32
-
-        expected = []
-        for i in xrange(num_objects):
-            type_num, data, sha = full_objects[i]
-            expected.append((offsets[i], type_num, data, sha, crc32s[i]))
-
-        sf.write_sha()
-        f.seek(0)
-        return f, expected
-
     def get_raw_no_repeat(self, bin_sha):
         """Wrapper around store.get_raw that doesn't allow repeat lookups."""
         hex_sha = sha_to_hex(bin_sha)
@@ -737,7 +668,7 @@ class DeltaChainIteratorTests(TestCase):
         self.assertEqual(expected, list(pack_iter._walk_all_chains()))
 
     def test_no_deltas(self):
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (Commit.type_num, 'commit'),
           (Blob.type_num, 'blob'),
           (Tree.type_num, 'tree'),
@@ -745,7 +676,7 @@ class DeltaChainIteratorTests(TestCase):
         self.assertEntriesMatch([0, 1, 2], entries, self.make_pack_iter(f))
 
     def test_ofs_deltas(self):
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (Blob.type_num, 'blob'),
           (OFS_DELTA, (0, 'blob1')),
           (OFS_DELTA, (0, 'blob2')),
@@ -753,7 +684,7 @@ class DeltaChainIteratorTests(TestCase):
         self.assertEntriesMatch([0, 1, 2], entries, self.make_pack_iter(f))
 
     def test_ofs_deltas_chain(self):
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (Blob.type_num, 'blob'),
           (OFS_DELTA, (0, 'blob1')),
           (OFS_DELTA, (1, 'blob2')),
@@ -761,7 +692,7 @@ class DeltaChainIteratorTests(TestCase):
         self.assertEntriesMatch([0, 1, 2], entries, self.make_pack_iter(f))
 
     def test_ref_deltas(self):
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (REF_DELTA, (1, 'blob1')),
           (Blob.type_num, ('blob')),
           (REF_DELTA, (1, 'blob2')),
@@ -769,7 +700,7 @@ class DeltaChainIteratorTests(TestCase):
         self.assertEntriesMatch([1, 0, 2], entries, self.make_pack_iter(f))
 
     def test_ref_deltas_chain(self):
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (REF_DELTA, (2, 'blob1')),
           (Blob.type_num, ('blob')),
           (REF_DELTA, (1, 'blob2')),
@@ -779,7 +710,7 @@ class DeltaChainIteratorTests(TestCase):
     def test_ofs_and_ref_deltas(self):
         # Deltas pending on this offset are popped before deltas depending on
         # this ref.
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (REF_DELTA, (1, 'blob1')),
           (Blob.type_num, ('blob')),
           (OFS_DELTA, (1, 'blob2')),
@@ -787,7 +718,7 @@ class DeltaChainIteratorTests(TestCase):
         self.assertEntriesMatch([1, 2, 0], entries, self.make_pack_iter(f))
 
     def test_mixed_chain(self):
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (Blob.type_num, 'blob'),
           (REF_DELTA, (2, 'blob2')),
           (OFS_DELTA, (0, 'blob1')),
@@ -802,7 +733,7 @@ class DeltaChainIteratorTests(TestCase):
         objects_spec = [(Blob.type_num, 'blob')]
         for i in xrange(n):
             objects_spec.append((OFS_DELTA, (i, 'blob%i' % i)))
-        f, entries = self.write_pack_data(objects_spec)
+        f, entries = build_pack(objects_spec)
         self.assertEntriesMatch(xrange(n + 1), entries, self.make_pack_iter(f))
 
     def test_branchy_chain(self):
@@ -810,42 +741,43 @@ class DeltaChainIteratorTests(TestCase):
         objects_spec = [(Blob.type_num, 'blob')]
         for i in xrange(n):
             objects_spec.append((OFS_DELTA, (0, 'blob%i' % i)))
-        f, entries = self.write_pack_data(objects_spec)
+        f, entries = build_pack(objects_spec)
         self.assertEntriesMatch(xrange(n + 1), entries, self.make_pack_iter(f))
 
     def test_ext_ref(self):
         blob, = self.store_blobs(['blob'])
-        f, entries = self.write_pack_data([(REF_DELTA, (blob.id, 'blob1'))])
+        f, entries = build_pack([(REF_DELTA, (blob.id, 'blob1'))],
+                                store=self.store)
         pack_iter = self.make_pack_iter(f)
         self.assertEntriesMatch([0], entries, pack_iter)
         self.assertEqual([hex_to_sha(blob.id)], pack_iter.ext_refs())
 
     def test_ext_ref_chain(self):
         blob, = self.store_blobs(['blob'])
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (REF_DELTA, (1, 'blob2')),
           (REF_DELTA, (blob.id, 'blob1')),
-          ])
+          ], store=self.store)
         pack_iter = self.make_pack_iter(f)
         self.assertEntriesMatch([1, 0], entries, pack_iter)
         self.assertEqual([hex_to_sha(blob.id)], pack_iter.ext_refs())
 
     def test_ext_ref_multiple_times(self):
         blob, = self.store_blobs(['blob'])
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (REF_DELTA, (blob.id, 'blob1')),
           (REF_DELTA, (blob.id, 'blob2')),
-          ])
+          ], store=self.store)
         pack_iter = self.make_pack_iter(f)
         self.assertEntriesMatch([0, 1], entries, pack_iter)
         self.assertEqual([hex_to_sha(blob.id)], pack_iter.ext_refs())
 
     def test_multiple_ext_refs(self):
         b1, b2 = self.store_blobs(['foo', 'bar'])
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (REF_DELTA, (b1.id, 'foo1')),
           (REF_DELTA, (b2.id, 'bar2')),
-          ])
+          ], store=self.store)
         pack_iter = self.make_pack_iter(f)
         self.assertEntriesMatch([0, 1], entries, pack_iter)
         self.assertEqual([hex_to_sha(b1.id), hex_to_sha(b2.id)],
@@ -853,7 +785,8 @@ class DeltaChainIteratorTests(TestCase):
 
     def test_bad_ext_ref_non_thin_pack(self):
         blob, = self.store_blobs(['blob'])
-        f, entries = self.write_pack_data([(REF_DELTA, (blob.id, 'blob1'))])
+        f, entries = build_pack([(REF_DELTA, (blob.id, 'blob1'))],
+                                store=self.store)
         pack_iter = self.make_pack_iter(f, thin=False)
         try:
             list(pack_iter._walk_all_chains())
@@ -863,12 +796,12 @@ class DeltaChainIteratorTests(TestCase):
 
     def test_bad_ext_ref_thin_pack(self):
         b1, b2, b3 = self.store_blobs(['foo', 'bar', 'baz'])
-        f, entries = self.write_pack_data([
+        f, entries = build_pack([
           (REF_DELTA, (1, 'foo99')),
           (REF_DELTA, (b1.id, 'foo1')),
           (REF_DELTA, (b2.id, 'bar2')),
           (REF_DELTA, (b3.id, 'baz3')),
-          ])
+          ], store=self.store)
         del self.store[b2.id]
         del self.store[b3.id]
         pack_iter = self.make_pack_iter(f)

dulwich/tests/utils.py

@@ -20,6 +20,7 @@
 """Utility functions common to Dulwich tests."""
 
 
+from cStringIO import StringIO
 import datetime
 import os
 import shutil
@@ -31,6 +32,16 @@ from dulwich.objects import (
     FixedSha,
     Commit,
     )
+from dulwich.pack import (
+    OFS_DELTA,
+    REF_DELTA,
+    DELTA_TYPES,
+    obj_sha,
+    SHA1Writer,
+    write_pack_header,
+    write_pack_object,
+    create_delta,
+    )
 from dulwich.repo import Repo
 from dulwich.tests import (
     TestSkipped,
@@ -147,3 +158,77 @@ def ext_functest_builder(method, func):
         method(self, func)
 
     return do_test
+
+
+def build_pack(objects_spec, store=None):
+    """Write test pack data from a concise spec.
+
+    :param objects_spec: A list of (type_num, obj). For non-delta types, obj
+        is the string of that object's data.
+
+        For delta types, obj is a tuple of (base, data), where:
+          base can be either an index in objects_spec of the base for that
+          delta; or for a ref delta, a SHA, in which case the resulting pack
+          will be thin and the base will be an external ref.
+          data is a string of the full, non-deltified data for that object.
+        Note that offsets/refs and deltas are computed within this function.
+    :param store: An optional ObjectStore for looking up external refs.
+
+    :return: A tuple of (f, entries), where f is a file-like object pointed
+        at the beginning of a pack with the requested data, and entries is a
+        list of tuples in the order specified by objects_spec:
+          (offset, type num, data, sha, CRC32)
+    """
+    f = StringIO()
+    sf = SHA1Writer(f)
+    num_objects = len(objects_spec)
+    write_pack_header(sf, num_objects)
+
+    full_objects = {}
+    offsets = {}
+    crc32s = {}
+
+    while len(full_objects) < num_objects:
+        for i, (type_num, data) in enumerate(objects_spec):
+            if type_num not in DELTA_TYPES:
+                full_objects[i] = (type_num, data,
+                                   obj_sha(type_num, [data]))
+                continue
+            base, data = data
+            if isinstance(base, int):
+                if base not in full_objects:
+                    continue
+                base_type_num, _, _ = full_objects[base]
+            else:
+                base_type_num, _ = store.get_raw(base)
+            full_objects[i] = (base_type_num, data,
+                               obj_sha(base_type_num, [data]))
+
+    for i, (type_num, obj) in enumerate(objects_spec):
+        offset = f.tell()
+        if type_num == OFS_DELTA:
+            base_index, data = obj
+            base = offset - offsets[base_index]
+            _, base_data, _ = full_objects[base_index]
+            obj = (base, create_delta(base_data, data))
+        elif type_num == REF_DELTA:
+            base_ref, data = obj
+            if isinstance(base_ref, int):
+                _, base_data, base = full_objects[base_ref]
+            else:
+                base_type_num, base_data = store.get_raw(base_ref)
+                base = obj_sha(base_type_num, base_data)
+            obj = (base, create_delta(base_data, data))
+
+        crc32 = write_pack_object(sf, type_num, obj)
+        offsets[i] = offset
+        crc32s[i] = crc32
+
+    expected = []
+    for i in xrange(num_objects):
+        type_num, data, sha = full_objects[i]
+        expected.append((offsets[i], type_num, data, sha, crc32s[i]))
+
+    sf.write_sha()
+    f.seek(0)
+    return f, expected