Coverage for britney2/migrationitem.py: 96%

1# -*- coding: utf-8 -*- 

2 

3# Copyright (C) 2011 Adam D. Barratt <adsb@debian.org> 

4 

5# This program is free software; you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation; either version 2 of the License, or 

8# (at your option) any later version. 

9 

10# This program is distributed in the hope that it will be useful, 

11# but WITHOUT ANY WARRANTY; without even the implied warranty of 

12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

13# GNU General Public License for more details. 

14 

15import apt_pkg 

16import logging 

17from typing import Optional 

18from britney2 import Suite, SuiteClass 

19 

20 

21class MigrationItem(object): 

22 

23 def __init__(self, package: str, suite: Suite, *, version: Optional[str] = None, 

24 architecture: Optional[str] = None, is_removal: bool = False, 

25 is_cruft_removal: bool = False): 

26 

27 if architecture is None: 

28 architecture = 'source' 

29 

30 if is_cruft_removal: 

31 is_removal = True 

32 

33 self._package = package 

34 self._version = version 

35 self._architecture = architecture 

36 self._suite = suite 

37 self._is_removal = is_removal 

38 self._is_cruft_removal = is_cruft_removal 

39 self._uvname = self.get_uvname() 

40 self._name = self.get_name() 

41 

42 def get_name(self) -> str: 

43 name = self._package 

44 if self._architecture != "source": 

45 name = "%s/%s" % (name, self._architecture) 

46 if self._version: 

47 name = "%s/%s" % (name, self._version) 

48 if self._suite.excuses_suffix: 

49 name = "%s_%s" % (name, self._suite.excuses_suffix) 

50 if self._is_removal: 

51 name = "-%s" % (name) 

52 return name 

53 

54 def get_uvname(self) -> str: 

55 name = self._package 

56 if self._architecture != "source": 

57 name = "%s/%s" % (name, self._architecture) 

58 if self._suite.excuses_suffix: 

59 name = "%s_%s" % (name, self._suite.excuses_suffix) 

60 if self._is_removal: 

61 name = "-%s" % (name) 

62 return name 

63 

64 def __repr__(self) -> str: 

65 return ("MI(%s)" % (self.__str__())) 

66 

67 def __str__(self) -> str: 

68 if self.version is not None: 68 ↛ 71line 68 didn't jump to line 71, because the condition on line 68 was never false

69 return self.name 

70 else: 

71 return self.uvname 

72 

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

74 isequal = False 

75 if isinstance(other, MigrationItem): 75 ↛ 82line 75 didn't jump to line 82, because the condition on line 75 was never false

76 if self.uvname == other.uvname: 

77 if self.version is None or other.version is None: 

78 isequal = True 

79 else: 

80 isequal = self.version == other.version 

81 

82 return isequal 

83 

84 def __hash__(self) -> int: 

85 if not self.version: 85 ↛ 86line 85 didn't jump to line 86, because the condition on line 85 was never true

86 raise AssertionError("trying to hash unversioned MigrationItem: %s" % 

87 (self.name)) 

88 

89 return hash((self.uvname, self.version)) 

90 

91 def __lt__(self, other: "MigrationItem") -> bool: 

92 return (self.uvname, self.version) < (other.uvname, other.version) 

93 

94 @property 

95 def name(self) -> str: 

96 return self._name 

97 

98 @property 

99 def is_removal(self) -> bool: 

100 return self._is_removal 

101 

102 @property 

103 def architecture(self) -> str: 

104 return self._architecture 

105 

106 @property 

107 def package(self) -> str: 

108 return self._package 

109 

110 @property 

111 def suite(self) -> Suite: 

112 return self._suite 

113 

114 @property 

115 def version(self) -> Optional[str]: 

116 return self._version 

117 

118 @property 

119 def uvname(self) -> str: 

120 return self._uvname 

121 

122 @property 

123 def is_cruft_removal(self) -> bool: 

124 return self._is_cruft_removal 

125 

126 

127class MigrationItemFactory(object): 

128 

129 def __init__(self, suites): 

130 self._suites = suites 

131 self._all_architectures = frozenset(suites.target_suite.binaries) 

132 logger_name = ".".join((self.__class__.__module__, self.__class__.__name__)) 

133 self.logger = logging.getLogger(logger_name) 

134 

135 def generate_removal_for_cruft_item(self, pkg_id): 

136 return MigrationItem(package=pkg_id.package_name, 

137 version=pkg_id.version, 

138 architecture=pkg_id.architecture, 

139 suite=self._suites.target_suite, 

140 is_cruft_removal=True, 

141 ) 

142 

143 @staticmethod 

144 def _is_right_version(suite, package_name, expected_version): 

145 if package_name not in suite.sources: 

146 return False 

147 

148 actual_version = suite.sources[package_name].version 

149 if apt_pkg.version_compare(actual_version, expected_version) != 0: 

150 return False 

151 

152 return True 

153 

154 def _find_suite_for_item(self, suites, suite_name, package_name, version, auto_correct): 

155 suite = suites.by_name_or_alias[suite_name] 

156 assert suite.suite_class != SuiteClass.TARGET_SUITE 

157 if version is not None and auto_correct and not self._is_right_version(suite, package_name, version): 

158 for s in suites.source_suites: 

159 if self._is_right_version(s, package_name, version): 

160 suite = s 

161 break 

162 return suite 

163 

164 def parse_item(self, item_text: str, versioned: bool = True, auto_correct: bool = True) -> MigrationItem: 

165 """ 

166 

167 :param item_text: The string describing the item (e.g. "glibc/2.5") 

168 :param versioned: If true, a two-part item is assumed to be versioned. 

169 otherwise, it is assumed to be versionless. This determines how 

170 items like "foo/bar" is parsed (if versioned, "bar" is assumed to 

171 be a version and otherwise "bar" is assumed to be an architecture). 

172 If in doubt, use versioned=True with auto_correct=True and the 

173 code will figure it out on its own. 

174 :param auto_correct: If True, minor issues are automatically fixed 

175 where possible. This includes handling architecture and version 

176 being in the wrong order and missing/omitting a suite reference 

177 for items. This feature is useful for migration items provided 

178 by humans (e.g. via hints) to avoid rejecting the input over 

179 trivial/minor issues with the input. 

180 When False, there will be no attempt to correct the migration 

181 input. 

182 :return: A MigrationItem matching the spec 

183 """ 

184 suites = self._suites 

185 version = None 

186 architecture = None 

187 is_removal = False 

188 if item_text.startswith('-'): 

189 item_text = item_text[1:] 

190 is_removal = True 

191 parts = item_text.split('/', 3) 

192 package_name = parts[0] 

193 suite_name = suites.primary_source_suite.name 

194 if '_' in package_name: 

195 package_name, suite_name = package_name.split('_', 2) 

196 

197 if len(parts) == 3: 

198 architecture = parts[1] 

199 version = parts[2] 

200 elif len(parts) == 2: 

201 if versioned: 201 ↛ 204line 201 didn't jump to line 204, because the condition on line 201 was never false

202 version = parts[1] 

203 else: 

204 architecture = parts[1] 

205 

206 if auto_correct and version in self._all_architectures: 

207 (architecture, version) = (version, architecture) 

208 

209 if architecture is None: 

210 architecture = 'source' 

211 

212 if '_' in architecture: 

213 architecture, suite_name = architecture.split('_', 2) 

214 

215 if is_removal: 

216 suite = suites.target_suite 

217 else: 

218 suite = self._find_suite_for_item(suites, suite_name, package_name, version, auto_correct) 

219 

220 return MigrationItem(package=package_name, 

221 version=version, 

222 architecture=architecture, 

223 suite=suite, 

224 is_removal=is_removal, 

225 ) 

226 

227 def parse_items(self, *args, **kwargs) -> list[MigrationItem]: 

228 return [self.parse_item(x, **kwargs) for x in args]