Coverage for britney2/migrationitem.py: 97%

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

2 

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

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

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

6# (at your option) any later version. 

7 

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

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

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

11# GNU General Public License for more details. 

12 

13import logging 

14from collections.abc import Iterable, Iterator 

15 

16import apt_pkg 

17 

18from britney2 import BinaryPackageId, Suite, SuiteClass, Suites 

19 

20 

21class MigrationItem: 

22 def __init__( 

23 self, 

24 package: str, 

25 suite: Suite, 

26 *, 

27 version: str | None = None, 

28 architecture: str | None = None, 

29 is_removal: bool = False, 

30 is_cruft_removal: bool = False, 

31 ): 

32 if architecture is None: 

33 architecture = "source" 

34 

35 if is_cruft_removal: 

36 is_removal = True 

37 

38 self._package = package 

39 self._version = version 

40 self._architecture = architecture 

41 self._suite = suite 

42 self._is_removal = is_removal 

43 self._is_cruft_removal = is_cruft_removal 

44 self._uvname = self.get_uvname() 

45 self._name = self.get_name() 

46 

47 def get_name(self) -> str: 

48 name = self._package 

49 if self._architecture != "source": 

50 name = f"{name}/{self._architecture}" 

51 if self._version: 

52 name = f"{name}/{self._version}" 

53 if self._suite.excuses_suffix: 

54 name = f"{name}_{self._suite.excuses_suffix}" 

55 if self._is_removal: 

56 name = f"-{name}" 

57 return name 

58 

59 def get_uvname(self) -> str: 

60 name = self._package 

61 if self._architecture != "source": 

62 name = f"{name}/{self._architecture}" 

63 if self._suite.excuses_suffix: 

64 name = f"{name}_{self._suite.excuses_suffix}" 

65 if self._is_removal: 

66 name = f"-{name}" 

67 return name 

68 

69 def __repr__(self) -> str: 

70 return f"MI({self.__str__()})" 

71 

72 def __str__(self) -> str: 

73 if self.version is not None: 

74 return self.name 

75 else: 

76 return self.uvname 

77 

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

79 isequal = False 

80 if isinstance(other, MigrationItem): 80 ↛ 87line 80 didn't jump to line 87 because the condition on line 80 was always true

81 if self.uvname == other.uvname: 

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

83 isequal = True 

84 else: 

85 isequal = self.version == other.version 

86 

87 return isequal 

88 

89 def __hash__(self) -> int: 

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

91 raise AssertionError( 

92 f"trying to hash unversioned MigrationItem: {self.name}" 

93 ) 

94 

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

96 

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

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

99 

100 @property 

101 def name(self) -> str: 

102 return self._name 

103 

104 @property 

105 def is_removal(self) -> bool: 

106 return self._is_removal 

107 

108 @property 

109 def architecture(self) -> str: 

110 return self._architecture 

111 

112 @property 

113 def package(self) -> str: 

114 return self._package 

115 

116 @property 

117 def suite(self) -> Suite: 

118 return self._suite 

119 

120 @property 

121 def version(self) -> str | None: 

122 return self._version 

123 

124 @property 

125 def uvname(self) -> str: 

126 return self._uvname 

127 

128 @property 

129 def is_cruft_removal(self) -> bool: 

130 return self._is_cruft_removal 

131 

132 

133class MigrationItemFactory: 

134 def __init__(self, suites: Suites) -> None: 

135 self._suites = suites 

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

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

138 self.logger = logging.getLogger(logger_name) 

139 

140 def generate_removal_for_cruft_item( 

141 self, pkg_id: "BinaryPackageId" 

142 ) -> MigrationItem: 

143 return MigrationItem( 

144 package=pkg_id.package_name, 

145 version=pkg_id.version, 

146 architecture=pkg_id.architecture, 

147 suite=self._suites.target_suite, 

148 is_cruft_removal=True, 

149 ) 

150 

151 @staticmethod 

152 def _is_right_version( 

153 suite: Suite, package_name: str, expected_version: str 

154 ) -> bool: 

155 source = suite.sources.get(package_name) 

156 if source is None: 

157 return False 

158 

159 actual_version = source.version 

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

161 return False 

162 

163 return True 

164 

165 def _find_suite_for_item( 

166 self, 

167 suites: Suites, 

168 suite_name: str, 

169 package_name: str, 

170 version: str | None, 

171 auto_correct: bool, 

172 ) -> Suite: 

173 suite = suites.by_name_or_alias[suite_name] 

174 assert suite.suite_class is not SuiteClass.TARGET_SUITE 

175 if ( 

176 version is not None 

177 and auto_correct 

178 and not self._is_right_version(suite, package_name, version) 

179 ): 

180 for s in suites.source_suites: 

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

182 suite = s 

183 break 

184 return suite 

185 

186 def parse_item( 

187 self, item_text: str, versioned: bool = True, auto_correct: bool = True 

188 ) -> MigrationItem: 

189 """ 

190 

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

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

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

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

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

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

197 code will figure it out on its own. 

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

199 where possible. This includes handling architecture and version 

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

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

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

203 trivial/minor issues with the input. 

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

205 input. 

206 :return: A MigrationItem matching the spec 

207 """ 

208 suites = self._suites 

209 version = None 

210 architecture = None 

211 is_removal = False 

212 if item_text.startswith("-"): 

213 item_text = item_text[1:] 

214 is_removal = True 

215 parts = item_text.split("/", 3) 

216 package_name = parts[0] 

217 suite_name = suites.primary_source_suite.name 

218 if "_" in package_name: 

219 package_name, suite_name = package_name.split("_", 2) 

220 

221 if len(parts) == 3: 

222 architecture = parts[1] 

223 version = parts[2] 

224 elif len(parts) == 2: 

225 if versioned: 225 ↛ 228line 225 didn't jump to line 228 because the condition on line 225 was always true

226 version = parts[1] 

227 else: 

228 architecture = parts[1] 

229 

230 if auto_correct and version in self._all_architectures: 

231 (architecture, version) = (version, architecture) 

232 

233 if architecture is None: 

234 architecture = "source" 

235 

236 if "_" in architecture: 

237 architecture, suite_name = architecture.split("_", 2) 

238 

239 if is_removal: 

240 suite: Suite = suites.target_suite 

241 else: 

242 suite = self._find_suite_for_item( 

243 suites, suite_name, package_name, version, auto_correct 

244 ) 

245 

246 return MigrationItem( 

247 package=package_name, 

248 version=version, 

249 architecture=architecture, 

250 suite=suite, 

251 is_removal=is_removal, 

252 ) 

253 

254 def parse_items( 

255 self, args: Iterable[str], versioned: bool = True, auto_correct: bool = True 

256 ) -> Iterator[MigrationItem]: 

257 yield from (self.parse_item(x, versioned, auto_correct) for x in args)