diff --git a/dulwich/bundle.py b/dulwich/bundle.py index 9848c207..1c1a2b8d 100644 --- a/dulwich/bundle.py +++ b/dulwich/bundle.py @@ -1,123 +1,123 @@ # bundle.py -- Bundle format support # Copyright (C) 2020 Jelmer Vernooij # # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Bundle format support. """ from typing import Dict, List, Tuple, Optional, Union, Sequence from .pack import PackData, write_pack_data class Bundle(object): - version: Optional[int] = None + version = None # type: Optional[int] - capabilities: Dict[str, str] = {} - prerequisites: List[Tuple[bytes, str]] = [] - references: Dict[str, bytes] = {} - pack_data: Union[PackData, Sequence[bytes]] = [] + capabilities = {} # type: Dict[str, str] + prerequisites = [] # type: List[Tuple[bytes, str]] + references = {} # type: Dict[str, bytes] + pack_data = [] # type: Union[PackData, Sequence[bytes]] def __eq__(self, other): if not isinstance(other, type(self)): return False if self.version != other.version: return False if self.capabilities != other.capabilities: return False if self.prerequisites != other.prerequisites: return False if self.references != other.references: return False if self.pack_data != other.pack_data: return False return True 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 ValueError: key = line value = None else: value = value.decode('utf-8') capabilities[key.decode('utf-8')] = value line = f.readline() while line.startswith(b'-'): (obj_id, comment) = line[1:].rstrip(b'\n').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 ret.version = version 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) def write_bundle(f, bundle): version = bundle.version if version is None: if bundle.capabilities: version = 3 else: version = 2 if version == 2: f.write(b'# v2 git bundle\n') elif version == 3: f.write(b'# v3 git bundle\n') else: raise AssertionError('unknown version %d' % version) if version == 3: for key, value in bundle.capabilities.items(): f.write(b'@' + key.encode('utf-8')) if value is not None: f.write(b'=' + value.encode('utf-8')) f.write(b'\n') for (obj_id, comment) in bundle.prerequisites: f.write(b'-%s %s\n' % (obj_id, comment.encode('utf-8'))) for ref, obj_id in bundle.references.items(): f.write(b'%s %s\n' % (obj_id, ref)) f.write(b'\n') write_pack_data(f, len(bundle.pack_data), iter(bundle.pack_data))