Coverage for britney2/__init__.py: 95%

1import logging 

2from collections import namedtuple 

3from enum import Enum, unique 

4from typing import TYPE_CHECKING, Any, NamedTuple, Optional 

5from collections.abc import Iterable, Iterator 

6 

7if TYPE_CHECKING: 7 ↛ 8line 7 didn't jump to line 8, because the condition on line 7 was never true

8 from .installability.tester import InstallabilityTester 

9 

10 

11class DependencyType(Enum): 

12 DEPENDS = ("Depends", "depends", "dependency") 

13 # BUILD_DEPENDS includes BUILD_DEPENDS_ARCH 

14 BUILD_DEPENDS = ("Build-Depends(-Arch)", "build-depends", "build-dependency") 

15 BUILD_DEPENDS_INDEP = ( 

16 "Build-Depends-Indep", 

17 "build-depends-indep", 

18 "build-dependency (indep)", 

19 ) 

20 BUILT_USING = ("Built-Using", "built-using", "built-using") 

21 # Pseudo dependency where Breaks/Conflicts effectively become a inverted dependency. E.g. 

22 # p Depends on q plus q/2 breaks p/1 implies that p/2 must migrate before q/2 can migrate 

23 # (or they go at the same time). 

24 # - can also happen with version ranges 

25 IMPLICIT_DEPENDENCY = ( 

26 "Implicit dependency", 

27 "implicit-dependency", 

28 "implicit-dependency", 

29 ) 

30 

31 def __str__(self) -> str: 

32 return self.value[0] 

33 

34 def get_reason(self) -> str: 

35 return self.value[1] 

36 

37 def get_description(self) -> str: 

38 return self.value[2] 

39 

40 

41@unique 

42class SuiteClass(Enum): 

43 TARGET_SUITE = (False, False) 

44 PRIMARY_SOURCE_SUITE = (True, True) 

45 ADDITIONAL_SOURCE_SUITE = (True, False) 

46 

47 @property 

48 def is_source(self) -> bool: 

49 return self.value[0] 

50 

51 @property 

52 def is_target(self) -> bool: 

53 return not self.is_source 

54 

55 @property 

56 def is_primary_source(self) -> bool: 

57 return self is SuiteClass.PRIMARY_SOURCE_SUITE 

58 

59 @property 

60 def is_additional_source(self) -> bool: 

61 return self is SuiteClass.ADDITIONAL_SOURCE_SUITE 

62 

63 

64class Suite(object): 

65 def __init__( 

66 self, 

67 suite_class: SuiteClass, 

68 name: str, 

69 path: str, 

70 suite_short_name: Optional[str] = None, 

71 ) -> None: 

72 self.suite_class = suite_class 

73 self.name = name 

74 self.codename = name 

75 self.path = path 

76 self.suite_short_name = suite_short_name if suite_short_name else "" 

77 self.sources: dict[str, SourcePackage] = {} 

78 self._binaries: dict[str, dict[str, BinaryPackage]] = {} 

79 self.provides_table: dict[str, dict[str, set[tuple[str, str]]]] = {} 

80 self._all_binaries_in_suite: Optional[dict[BinaryPackageId, BinaryPackage]] = ( 

81 None 

82 ) 

83 

84 @property 

85 def excuses_suffix(self) -> str: 

86 return self.suite_short_name 

87 

88 @property 

89 def binaries(self) -> dict[str, dict[str, "BinaryPackage"]]: 

90 # TODO some callers modify this structure, which doesn't invalidate 

91 # the self._all_binaries_in_suite cache 

92 return self._binaries 

93 

94 @binaries.setter 

95 def binaries(self, binaries: dict[str, dict[str, "BinaryPackage"]]) -> None: 

96 self._binaries = binaries 

97 self._all_binaries_in_suite = None 

98 

99 @property 

100 def all_binaries_in_suite(self) -> dict["BinaryPackageId", "BinaryPackage"]: 

101 if not self._all_binaries_in_suite: 

102 self._all_binaries_in_suite = { 

103 x.pkg_id: x for a in self._binaries for x in self._binaries[a].values() 

104 } 

105 return self._all_binaries_in_suite 

106 

107 def any_of_these_are_in_the_suite(self, pkgs: Iterable["BinaryPackageId"]) -> bool: 

108 """Test if at least one package of a given set is in the suite 

109 

110 :return: True if any of the packages in pkgs are currently in the suite 

111 """ 

112 return not self.all_binaries_in_suite.keys().isdisjoint(pkgs) 

113 

114 def is_pkg_in_the_suite(self, pkg_id: "BinaryPackageId") -> bool: 

115 """Test if the package of is in testing 

116 

117 :return: True if the pkg is currently in the suite 

118 """ 

119 return pkg_id in self.all_binaries_in_suite 

120 

121 def which_of_these_are_in_the_suite( 

122 self, pkgs: Iterable["BinaryPackageId"] 

123 ) -> Iterator["BinaryPackageId"]: 

124 """Iterate over all packages that are in the suite 

125 

126 :return: An iterable of package ids that are in the suite 

127 """ 

128 yield from (x for x in pkgs if x in self.all_binaries_in_suite) 

129 

130 def is_cruft(self, pkg: "BinaryPackage") -> bool: 

131 """Check if the package is cruft in the suite 

132 

133 :param pkg: which BinaryPackage to check 

134 Note that this package is assumed to be in the suite 

135 """ 

136 newest_src_in_suite = self.sources[pkg.source] 

137 return pkg.source_version != newest_src_in_suite.version 

138 

139 

140class TargetSuite(Suite): 

141 inst_tester: "InstallabilityTester" 

142 

143 def __init__(self, *args: Any, **kwargs: Any) -> None: 

144 super().__init__(*args, **kwargs) 

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

146 self._logger = logging.getLogger(logger_name) 

147 

148 def is_installable(self, pkg_id: "BinaryPackageId") -> bool: 

149 """Determine whether the given package can be installed in the suite 

150 

151 :param pkg_id: A BinaryPackageId 

152 :return: True if the pkg is currently installable in the suite 

153 """ 

154 return self.inst_tester.is_installable(pkg_id) 

155 

156 def add_binary(self, pkg_id: "BinaryPackageId") -> None: 

157 """Add a binary package to the suite 

158 

159 :param pkg_id: The id of the package 

160 :raises KeyError: if the package is not known 

161 """ 

162 

163 # TODO The calling code currently manually updates the contents of 

164 # target_suite.binaries when this is called. It would probably make 

165 # more sense to do that here instead 

166 self.inst_tester.add_binary(pkg_id) 

167 self._all_binaries_in_suite = None 

168 

169 def remove_binary(self, pkg_id: "BinaryPackageId") -> None: 

170 """Remove a binary from the suite 

171 

172 :param pkg_id: The id of the package 

173 :raises KeyError: if the package is not known 

174 """ 

175 

176 # TODO The calling code currently manually updates the contents of 

177 # target_suite.binaries when this is called. It would probably make 

178 # more sense to do that here instead 

179 self.inst_tester.remove_binary(pkg_id) 

180 self._all_binaries_in_suite = None 

181 

182 def check_suite_source_pkg_consistency(self, comment: str) -> None: 

183 sources_t = self.sources 

184 binaries_t = self.binaries 

185 logger = self._logger 

186 issues_found = False 

187 

188 logger.info("check_target_suite_source_pkg_consistency %s", comment) 

189 

190 for arch in binaries_t: 

191 for pkg_name in binaries_t[arch]: 

192 pkg = binaries_t[arch][pkg_name] 

193 src = pkg.source 

194 

195 if src not in sources_t: # pragma: no cover 

196 issues_found = True 

197 logger.error( 

198 "inconsistency found (%s): src %s not in target, target has pkg %s with source %s" 

199 % (comment, src, pkg_name, src) 

200 ) 

201 

202 for src in sources_t: 

203 source_data = sources_t[src] 

204 for pkg_id in source_data.binaries: 

205 binary, _, parch = pkg_id 

206 if binary not in binaries_t[parch]: # pragma: no cover 

207 issues_found = True 

208 logger.error( 

209 "inconsistency found (%s): binary %s from source %s not in binaries_t[%s]" 

210 % (comment, binary, src, parch) 

211 ) 

212 

213 if issues_found: # pragma: no cover 

214 raise AssertionError("inconsistencies found in target suite") 

215 

216 

217class Suites(object): 

218 def __init__(self, target_suite: TargetSuite, source_suites: list[Suite]) -> None: 

219 self._suites: dict[str, Suite] = {} 

220 self._by_name_or_alias: dict[str, Suite] = {} 

221 self.target_suite = target_suite 

222 self.source_suites = source_suites 

223 self._suites[target_suite.name] = target_suite 

224 self._by_name_or_alias[target_suite.name] = target_suite 

225 if target_suite.suite_short_name: 225 ↛ 226line 225 didn't jump to line 226, because the condition on line 225 was never true

226 self._by_name_or_alias[target_suite.suite_short_name] = target_suite 

227 for suite in source_suites: 

228 self._suites[suite.name] = suite 

229 self._by_name_or_alias[suite.name] = suite 

230 if suite.suite_short_name: 

231 self._by_name_or_alias[suite.suite_short_name] = suite 

232 

233 @property 

234 def primary_source_suite(self) -> Suite: 

235 return self.source_suites[0] 

236 

237 @property 

238 def by_name_or_alias(self) -> dict[str, Suite]: 

239 return self._by_name_or_alias 

240 

241 @property 

242 def additional_source_suites(self) -> list[Suite]: 

243 return self.source_suites[1:] 

244 

245 def __getitem__(self, item: str) -> Suite: 

246 return self._suites[item] 

247 

248 def __len__(self) -> int: 

249 return len(self.source_suites) + 1 

250 

251 def __contains__(self, item: str) -> bool: 

252 return item in self._suites 

253 

254 def __iter__(self) -> Iterator[Suite]: 

255 # Sources first (as we will rely on this for loading data in the old live-data tests) 

256 yield from self.source_suites 

257 yield self.target_suite 

258 

259 

260class SourcePackage(object): 

261 __slots__ = [ 

262 "source", 

263 "version", 

264 "section", 

265 "binaries", 

266 "maintainer", 

267 "is_fakesrc", 

268 "build_deps_arch", 

269 "build_deps_indep", 

270 "testsuite", 

271 "testsuite_triggers", 

272 ] 

273 

274 def __init__( 

275 self, 

276 source: str, 

277 version: str, 

278 section: str, 

279 binaries: set["BinaryPackageId"], 

280 maintainer: Optional[str], 

281 is_fakesrc: bool, 

282 build_deps_arch: Optional[str], 

283 build_deps_indep: Optional[str], 

284 testsuite: list[str], 

285 testsuite_triggers: list[str], 

286 ) -> None: 

287 self.source = source 

288 self.version = version 

289 self.section = section 

290 self.binaries = binaries 

291 self.maintainer = maintainer 

292 self.is_fakesrc = is_fakesrc 

293 self.build_deps_arch = build_deps_arch 

294 self.build_deps_indep = build_deps_indep 

295 self.testsuite = testsuite 

296 self.testsuite_triggers = testsuite_triggers 

297 

298 def __getitem__(self, item: int) -> Any: 

299 return getattr(self, self.__slots__[item]) 

300 

301 

302class PackageId( 

303 namedtuple( 

304 "PackageId", 

305 [ 

306 "package_name", 

307 "version", 

308 "architecture", 

309 ], 

310 ) 

311): 

312 """Represent a source or binary package""" 

313 

314 def __init__(self, package_name: str, version: str, architecture: str) -> None: 

315 assert self.architecture != "all", "all not allowed for PackageId (%s)" % ( 

316 self.name 

317 ) 

318 

319 def __repr__(self) -> str: 

320 return "PID(%s)" % (self.name) 

321 

322 @property 

323 def name(self) -> str: 

324 if self.architecture == "source": 324 ↛ 325line 324 didn't jump to line 325, because the condition on line 324 was never true

325 return "%s/%s" % (self.package_name, self.version) 

326 else: 

327 return "%s/%s/%s" % (self.package_name, self.version, self.architecture) 

328 

329 @property 

330 def uvname(self) -> str: 

331 if self.architecture == "source": 331 ↛ 334line 331 didn't jump to line 334, because the condition on line 331 was never false

332 return "%s" % (self.package_name) 

333 else: 

334 return "%s/%s" % (self.package_name, self.architecture) 

335 

336 

337class BinaryPackageId(PackageId): 

338 """Represent a binary package""" 

339 

340 def __init__(self, package_name: str, version: str, architecture: str) -> None: 

341 assert ( 

342 self.architecture != "source" 

343 ), "Source not allowed for BinaryPackageId (%s)" % (self.name) 

344 super().__init__(package_name, version, architecture) 

345 

346 def __repr__(self) -> str: 

347 return "BPID(%s)" % (self.name) 

348 

349 

350class BinaryPackage(NamedTuple): 

351 version: str 

352 section: str 

353 source: str 

354 source_version: str 

355 architecture: str 

356 multi_arch: Optional[str] 

357 depends: Optional[str] 

358 conflicts: Optional[str] 

359 provides: list[tuple[str, str, str]] 

360 is_essential: bool 

361 pkg_id: BinaryPackageId 

362 builtusing: list[tuple[str, str]]