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 typing import Any, Optional 

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 = "-%s" % (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 = "-%s" % (name) 

67 return name 

68 

69 def __repr__(self) -> str: 

70 return "MI(%s)" % (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 "trying to hash unversioned MigrationItem: %s" % (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 if package_name not in suite.sources: 

156 return False 

157 

158 actual_version = suite.sources[package_name].version 

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

160 return False 

161 

162 return True 

163 

164 def _find_suite_for_item( 

165 self, 

166 suites: Suites, 

167 suite_name: str, 

168 package_name: str, 

169 version: str | None, 

170 auto_correct: bool, 

171 ) -> Suite: 

172 suite = suites.by_name_or_alias[suite_name] 

173 assert suite.suite_class != SuiteClass.TARGET_SUITE 

174 if ( 

175 version is not None 

176 and auto_correct 

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

178 ): 

179 for s in suites.source_suites: 

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

181 suite = s 

182 break 

183 return suite 

184 

185 def parse_item( 

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

187 ) -> MigrationItem: 

188 """ 

189 

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

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

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

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

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

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

196 code will figure it out on its own. 

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

198 where possible. This includes handling architecture and version 

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

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

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

202 trivial/minor issues with the input. 

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

204 input. 

205 :return: A MigrationItem matching the spec 

206 """ 

207 suites = self._suites 

208 version = None 

209 architecture = None 

210 is_removal = False 

211 if item_text.startswith("-"): 

212 item_text = item_text[1:] 

213 is_removal = True 

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

215 package_name = parts[0] 

216 suite_name = suites.primary_source_suite.name 

217 if "_" in package_name: 

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

219 

220 if len(parts) == 3: 

221 architecture = parts[1] 

222 version = parts[2] 

223 elif len(parts) == 2: 

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

225 version = parts[1] 

226 else: 

227 architecture = parts[1] 

228 

229 if auto_correct and version in self._all_architectures: 

230 (architecture, version) = (version, architecture) 

231 

232 if architecture is None: 

233 architecture = "source" 

234 

235 if "_" in architecture: 

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

237 

238 if is_removal: 

239 suite: Suite = suites.target_suite 

240 else: 

241 suite = self._find_suite_for_item( 

242 suites, suite_name, package_name, version, auto_correct 

243 ) 

244 

245 return MigrationItem( 

246 package=package_name, 

247 version=version, 

248 architecture=architecture, 

249 suite=suite, 

250 is_removal=is_removal, 

251 ) 

252 

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

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