Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/bundle.py: 35%

1# bundle.py -- Bundle format support 

2# Copyright (C) 2020 Jelmer Vernooij <jelmer@jelmer.uk> 

3# 

4# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later 

5# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU 

6# General Public License as public by the Free Software Foundation; version 2.0 

7# or (at your option) any later version. You can redistribute it and/or 

8# modify it under the terms of either of these two licenses. 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15# 

16# You should have received a copy of the licenses; if not, see 

17# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License 

18# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache 

19# License, Version 2.0. 

20# 

21 

22"""Bundle format support.""" 

23 

24from typing import BinaryIO, Optional 

25 

26from .pack import PackData, write_pack_data 

27 

28 

29class Bundle: 

30 version: Optional[int] 

31 

32 capabilities: dict[str, Optional[str]] 

33 prerequisites: list[tuple[bytes, str]] 

34 references: dict[bytes, bytes] 

35 pack_data: PackData 

36 

37 def __repr__(self) -> str: 

38 return ( 

39 f"<{type(self).__name__}(version={self.version}, " 

40 f"capabilities={self.capabilities}, " 

41 f"prerequisites={self.prerequisites}, " 

42 f"references={self.references})>" 

43 ) 

44 

45 def __eq__(self, other: object) -> bool: 

46 if not isinstance(other, type(self)): 

47 return False 

48 if self.version != other.version: 

49 return False 

50 if self.capabilities != other.capabilities: 

51 return False 

52 if self.prerequisites != other.prerequisites: 

53 return False 

54 if self.references != other.references: 

55 return False 

56 if self.pack_data != other.pack_data: 

57 return False 

58 return True 

59 

60 

61def _read_bundle(f: BinaryIO, version: int) -> Bundle: 

62 capabilities = {} 

63 prerequisites = [] 

64 references = {} 

65 line = f.readline() 

66 if version >= 3: 

67 while line.startswith(b"@"): 

68 line = line[1:].rstrip(b"\n") 

69 try: 

70 key, value_bytes = line.split(b"=", 1) 

71 value = value_bytes.decode("utf-8") 

72 except ValueError: 

73 key = line 

74 value = None 

75 capabilities[key.decode("utf-8")] = value 

76 line = f.readline() 

77 while line.startswith(b"-"): 

78 (obj_id, comment) = line[1:].rstrip(b"\n").split(b" ", 1) 

79 prerequisites.append((obj_id, comment.decode("utf-8"))) 

80 line = f.readline() 

81 while line != b"\n": 

82 (obj_id, ref) = line.rstrip(b"\n").split(b" ", 1) 

83 references[ref] = obj_id 

84 line = f.readline() 

85 pack_data = PackData.from_file(f) 

86 ret = Bundle() 

87 ret.references = references 

88 ret.capabilities = capabilities 

89 ret.prerequisites = prerequisites 

90 ret.pack_data = pack_data 

91 ret.version = version 

92 return ret 

93 

94 

95def read_bundle(f: BinaryIO) -> Bundle: 

96 """Read a bundle file.""" 

97 firstline = f.readline() 

98 if firstline == b"# v2 git bundle\n": 

99 return _read_bundle(f, 2) 

100 if firstline == b"# v3 git bundle\n": 

101 return _read_bundle(f, 3) 

102 raise AssertionError(f"unsupported bundle format header: {firstline!r}") 

103 

104 

105def write_bundle(f: BinaryIO, bundle: Bundle) -> None: 

106 version = bundle.version 

107 if version is None: 

108 if bundle.capabilities: 

109 version = 3 

110 else: 

111 version = 2 

112 if version == 2: 

113 f.write(b"# v2 git bundle\n") 

114 elif version == 3: 

115 f.write(b"# v3 git bundle\n") 

116 else: 

117 raise AssertionError(f"unknown version {version}") 

118 if version == 3: 

119 for key, value in bundle.capabilities.items(): 

120 f.write(b"@" + key.encode("utf-8")) 

121 if value is not None: 

122 f.write(b"=" + value.encode("utf-8")) 

123 f.write(b"\n") 

124 for obj_id, comment in bundle.prerequisites: 

125 f.write(b"-%s %s\n" % (obj_id, comment.encode("utf-8"))) 

126 for ref, obj_id in bundle.references.items(): 

127 f.write(b"%s %s\n" % (obj_id, ref)) 

128 f.write(b"\n") 

129 write_pack_data( 

130 f.write, 

131 num_records=len(bundle.pack_data), 

132 records=bundle.pack_data.iter_unpacked(), 

133 )