Add InfoRefsContainer.

dulwich/repo.py

@@ -388,6 +388,37 @@ class DictRefsContainer(RefsContainer):
         self._peeled.update(peeled)
 
 
+class InfoRefsContainer(RefsContainer):
+    """Refs container that reads refs from a info/refs file."""
+
+    def __init__(self, f):
+        self._refs = {}
+        self._peeled = {}
+        for l in f.readlines():
+            sha, name = l.rstrip("\n").split("\t")
+            if name.endswith("^{}"):
+                name = name[:-3]
+                if not check_ref_format(name):
+                    raise ValueError("invalid ref name '%s'" % name)
+                self._peeled[name] = sha
+            else:
+                if not check_ref_format(name):
+                    raise ValueError("invalid ref name '%s'" % name)
+                self._refs[name] = sha
+
+    def allkeys(self):
+        return self._refs.keys()
+
+    def read_loose_ref(self, name):
+        return self._refs.get(name, None)
+
+    def get_packed_refs(self):
+        return {}
+
+    def get_peeled(self, name):
+        return self._peeled.get(name)
+
+
 class DiskRefsContainer(RefsContainer):
     """Refs container that reads refs from disk."""
 

dulwich/tests/test_repository.py

@@ -36,6 +36,7 @@ from dulwich import objects
 from dulwich.repo import (
     check_ref_format,
     DictRefsContainer,
+    InfoRefsContainer,
     Repo,
     MemoryRepo,
     read_packed_refs,
@@ -891,3 +892,49 @@ class DiskRefsContainerTests(RefsContainerTests, TestCase):
             self._refs.read_ref("refs/heads/packed"))
         self.assertEqual(None,
             self._refs.read_ref("nonexistant"))
+
+
+_TEST_REFS_SERIALIZED = (
+'42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/master\n'
+'42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/packed\n'
+'df6800012397fb85c56e7418dd4eb9405dee075c\trefs/tags/refs-0.1\n'
+'3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8\trefs/tags/refs-0.2\n')
+
+
+class InfoRefsContainerTests(TestCase):
+
+    def test_invalid_refname(self):
+        text = _TEST_REFS_SERIALIZED + '00' * 20 + '\trefs/stash\n'
+        refs = InfoRefsContainer(StringIO(text))
+        expected_refs = dict(_TEST_REFS)
+        del expected_refs['HEAD']
+        expected_refs["refs/stash"] = "00" * 20
+        self.assertEquals(expected_refs, refs.as_dict())
+
+    def test_keys(self):
+        refs = InfoRefsContainer(StringIO(_TEST_REFS_SERIALIZED))
+        actual_keys = set(refs.keys())
+        self.assertEqual(set(refs.allkeys()), actual_keys)
+        # ignore the symref loop if it exists
+        actual_keys.discard('refs/heads/loop')
+        expected_refs = dict(_TEST_REFS)
+        del expected_refs['HEAD']
+        self.assertEqual(set(expected_refs.iterkeys()), actual_keys)
+
+        actual_keys = refs.keys('refs/heads')
+        actual_keys.discard('loop')
+        self.assertEqual(['master', 'packed'], sorted(actual_keys))
+        self.assertEqual(['refs-0.1', 'refs-0.2'],
+                         sorted(refs.keys('refs/tags')))
+
+    def test_as_dict(self):
+        refs = InfoRefsContainer(StringIO(_TEST_REFS_SERIALIZED))
+        # refs/heads/loop does not show up even if it exists
+        expected_refs = dict(_TEST_REFS)
+        del expected_refs['HEAD']
+        self.assertEqual(expected_refs, refs.as_dict())
+
+    def test_contains(self):
+        refs = InfoRefsContainer(StringIO(_TEST_REFS_SERIALIZED))
+        self.assertTrue('refs/heads/master' in refs)
+        self.assertFalse('refs/heads/bar' in refs)