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

1# bundle.py -- Bundle format support 

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

3# 

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

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

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

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

8# 

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

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

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

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

13# limitations under the License. 

14# 

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

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

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

18# License, Version 2.0. 

19# 

20 

21"""Bundle format support.""" 

22 

23from typing import Dict, List, Optional, Sequence, Tuple, Union 

24 

25from .pack import PackData, write_pack_data 

26 

27 

28class Bundle: 

29 version: Optional[int] 

30 

31 capabilities: Dict[str, str] 

32 prerequisites: List[Tuple[bytes, str]] 

33 references: Dict[str, bytes] 

34 pack_data: Union[PackData, Sequence[bytes]] 

35 

36 def __repr__(self) -> str: 

37 return ( 

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

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

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

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

42 ) 

43 

44 def __eq__(self, other): 

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

46 return False 

47 if self.version != other.version: 

48 return False 

49 if self.capabilities != other.capabilities: 

50 return False 

51 if self.prerequisites != other.prerequisites: 

52 return False 

53 if self.references != other.references: 

54 return False 

55 if self.pack_data != other.pack_data: 

56 return False 

57 return True 

58 

59 

60def _read_bundle(f, version): 

61 capabilities = {} 

62 prerequisites = [] 

63 references = {} 

64 line = f.readline() 

65 if version >= 3: 

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

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

68 try: 

69 key, value = line.split(b"=", 1) 

70 except ValueError: 

71 key = line 

72 value = None 

73 else: 

74 value = value.decode("utf-8") 

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): 

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, bundle): 

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("unknown version %d" % 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 )