|
@@ -0,0 +1,78 @@
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+"""Bundle format support.
|
|
|
+"""
|
|
|
+
|
|
|
+from .pack import PackData
|
|
|
+from typing import Dict, List, Tuple
|
|
|
+
|
|
|
+
|
|
|
+class Bundle(object):
|
|
|
+
|
|
|
+ version: int
|
|
|
+
|
|
|
+ capabilities: Dict[str, str]
|
|
|
+ prerequisites: List[Tuple[bytes, str]]
|
|
|
+ references: Dict[str, bytes]
|
|
|
+ pack_data: PackData
|
|
|
+
|
|
|
+
|
|
|
+def _read_bundle(f, version):
|
|
|
+ capabilities = {}
|
|
|
+ prerequisites = []
|
|
|
+ references = {}
|
|
|
+ line = f.readline()
|
|
|
+ if version >= 3:
|
|
|
+ while line.startswith(b'@'):
|
|
|
+ line = line[1:].rstrip(b'\n')
|
|
|
+ try:
|
|
|
+ key, value = line.split(b'=', 1)
|
|
|
+ except IndexError:
|
|
|
+ key = line
|
|
|
+ value = None
|
|
|
+ capabilities[key] = value
|
|
|
+ line = f.readline()
|
|
|
+ while line.startswith(b'-'):
|
|
|
+ (obj_id, comment) = line[1:].split(b' ', 1)
|
|
|
+ prerequisites.append((obj_id, comment.decode('utf-8')))
|
|
|
+ line = f.readline()
|
|
|
+ while line != b'\n':
|
|
|
+ (obj_id, ref) = line.rstrip(b'\n').split(b' ', 1)
|
|
|
+ references[ref] = obj_id
|
|
|
+ line = f.readline()
|
|
|
+ pack_data = PackData.from_file(f)
|
|
|
+ ret = Bundle()
|
|
|
+ ret.references = references
|
|
|
+ ret.capabilities = capabilities
|
|
|
+ ret.prerequisites = prerequisites
|
|
|
+ ret.pack_data = pack_data
|
|
|
+ return ret
|
|
|
+
|
|
|
+
|
|
|
+def read_bundle(f):
|
|
|
+ """Read a bundle file."""
|
|
|
+ firstline = f.readline()
|
|
|
+ if firstline == b'# v2 git bundle\n':
|
|
|
+ return _read_bundle(f, 2)
|
|
|
+ if firstline == b'# v3 git bundle\n':
|
|
|
+ return _read_bundle(f, 3)
|
|
|
+ raise AssertionError(
|
|
|
+ 'unsupported bundle format header: %r' % firstline)
|