Coverage for britney2/__init__.py: 95%

1import logging 

2import sys 

3from collections.abc import Iterable, Iterator 

4from dataclasses import dataclass, field 

5from enum import Enum, unique 

6from typing import TYPE_CHECKING, Any 

7 

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

9 from .installability.tester import InstallabilityTester 

10 

11 

12@dataclass(slots=True, frozen=True, order=True) 

13class PackageId: 

14 package_name: str 

15 version: str 

16 architecture: str 

17 """Represent a source or binary package""" 

18 

19 def __post_init__(self) -> None: 

20 assert ( 

21 self.architecture != "all" 

22 ), f"all not allowed for PackageId ({self.name})" 

23 

24 def __repr__(self) -> str: 

25 return f"PID({self.name})" 

26 

27 @property 

28 def name(self) -> str: 

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

30 return f"{self.package_name}/{self.version}" 

31 else: 

32 return f"{self.package_name}/{self.version}/{self.architecture}" 

33 

34 @property 

35 def uvname(self) -> str: 

36 if self.architecture == "source": 36 ↛ 39line 36 didn't jump to line 39 because the condition on line 36 was always true

37 return self.package_name 

38 else: 

39 return f"{self.package_name}/{self.architecture}" 

40 

41 

42class BinaryPackageId(PackageId): 

43 """Represent a binary package""" 

44 

45 # The base class has slots, this class does not need __dict__. 

46 __slots__ = () 

47 

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

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

50 assert ( 

51 self.architecture != "source" 

52 ), f"Source not allowed for BinaryPackageId ({package_name})" 

53 

54 def __repr__(self) -> str: 

55 return f"BPID({self.name})" 

56 

57 

58class MultiArch(Enum): 

59 SAME = 0 

60 FOREIGN = 1 

61 ALLOWED = 2 

62 NO = 3 

63 

64 @staticmethod 

65 def from_str(value: str | None) -> "MultiArch": 

66 if value == "same": 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 return MultiArch.SAME 

68 elif value == "foreign": 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true

69 return MultiArch.FOREIGN 

70 elif value == "allowed": 

71 return MultiArch.ALLOWED 

72 elif value == "no" or value is None: 72 ↛ 75line 72 didn't jump to line 75 because the condition on line 72 was always true

73 return MultiArch.NO 

74 else: 

75 raise ValueError(f"invalid Multi-Arch value {value}") 

76 

77 

78@dataclass(slots=True) 

79class BinaryPackage: 

80 section: str 

81 source: str 

82 source_version: str 

83 architecture: str 

84 multi_arch: MultiArch 

85 depends: str | None 

86 conflicts: str | None 

87 # List of provided packages and versions 

88 # If the version is non-empty, the provides is versioned, e.g. the form $pkg (= $ver). 

89 provides: list[tuple[str, str]] | None 

90 is_essential: bool 

91 pkg_id: BinaryPackageId 

92 builtusing: list[tuple[str, str]] | None 

93 

94 def __post_init__(self) -> None: 

95 intern = sys.intern 

96 self.section = intern(self.section) 

97 self.source = intern(self.source) 

98 self.source_version = intern(self.source_version) 

99 self.architecture = intern(self.architecture) 

100 if self.depends is not None: 

101 self.depends = intern(self.depends) 

102 if self.conflicts is not None: 

103 self.conflicts = intern(self.conflicts) 

104 if self.provides is not None: 

105 for i, p in enumerate(self.provides): 

106 self.provides[i] = (intern(p[0]), intern(p[1])) 

107 if self.builtusing is not None: 

108 for i, p in enumerate(self.builtusing): 

109 self.builtusing[i] = (intern(p[0]), intern(p[1])) 

110 

111 @property 

112 def version(self) -> str: 

113 return self.pkg_id.version 

114 

115 

116@dataclass(slots=True) 

117class SourcePackage: 

118 source: str 

119 version: str 

120 section: str 

121 binaries: set[BinaryPackageId] 

122 maintainer: str | None 

123 is_fakesrc: bool 

124 build_deps_arch: str | None = None 

125 build_deps_indep: str | None = None 

126 testsuite: list[str] | None = None 

127 testsuite_triggers: list[str] | None = None 

128 

129 def __post_init__(self) -> None: 

130 intern = sys.intern 

131 self.section = intern(self.section) 

132 self.source = intern(self.source) 

133 self.version = intern(self.version) 

134 if self.maintainer is not None: 

135 self.maintainer = intern(self.maintainer) 

136 if self.build_deps_arch is not None: 

137 self.build_deps_arch = intern(self.build_deps_arch) 

138 if self.build_deps_indep is not None: 

139 self.build_deps_indep = intern(self.build_deps_indep) 

140 if self.testsuite is not None: 

141 for i, p in enumerate(self.testsuite): 

142 self.testsuite[i] = intern(p) 

143 if self.testsuite_triggers is not None: 

144 for i, p in enumerate(self.testsuite_triggers): 

145 self.testsuite_triggers[i] = intern(p) 

146 

147 

148class DependencyType(Enum): 

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

150 # BUILD_DEPENDS includes BUILD_DEPENDS_ARCH 

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

152 BUILD_DEPENDS_INDEP = ( 

153 "Build-Depends-Indep", 

154 "build-depends-indep", 

155 "build-dependency (indep)", 

156 ) 

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

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

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

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

161 # - can also happen with version ranges 

162 IMPLICIT_DEPENDENCY = ( 

163 "Implicit dependency", 

164 "implicit-dependency", 

165 "implicit-dependency", 

166 ) 

167 

168 def __str__(self) -> str: 

169 return self.value[0] 

170 

171 def get_reason(self) -> str: 

172 return self.value[1] 

173 

174 def get_description(self) -> str: 

175 return self.value[2] 

176 

177 

178@unique 

179class SuiteClass(Enum): 

180 TARGET_SUITE = (False, False) 

181 PRIMARY_SOURCE_SUITE = (True, True) 

182 ADDITIONAL_SOURCE_SUITE = (True, False) 

183 

184 @property 

185 def is_source(self) -> bool: 

186 return self.value[0] 

187 

188 @property 

189 def is_target(self) -> bool: 

190 return not self.is_source 

191 

192 @property 

193 def is_primary_source(self) -> bool: 

194 return self is SuiteClass.PRIMARY_SOURCE_SUITE 

195 

196 @property 

197 def is_additional_source(self) -> bool: 

198 return self is SuiteClass.ADDITIONAL_SOURCE_SUITE 

199 

200 

201class Suite: 

202 def __init__( 

203 self, 

204 suite_class: SuiteClass, 

205 name: str, 

206 path: str, 

207 suite_short_name: str | None = None, 

208 ) -> None: 

209 self.suite_class = suite_class 

210 self.name = name 

211 self.codename = name 

212 self.path = path 

213 self.suite_short_name = suite_short_name if suite_short_name else "" 

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

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

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

217 self._all_binaries_in_suite: dict[BinaryPackageId, BinaryPackage] | None = None 

218 

219 @property 

220 def excuses_suffix(self) -> str: 

221 return self.suite_short_name 

222 

223 @property 

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

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

226 # the self._all_binaries_in_suite cache 

227 return self._binaries 

228 

229 @binaries.setter 

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

231 self._binaries = binaries 

232 self._all_binaries_in_suite = None 

233 

234 @property 

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

236 if self._all_binaries_in_suite is None: 

237 self._all_binaries_in_suite = { 

238 x.pkg_id: x for a in self._binaries.values() for x in a.values() 

239 } 

240 return self._all_binaries_in_suite 

241 

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

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

244 

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

246 """ 

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

248 

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

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

251 

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

253 """ 

254 return pkg_id in self.all_binaries_in_suite 

255 

256 def which_of_these_are_in_the_suite( 

257 self, pkgs: Iterable[BinaryPackageId] 

258 ) -> Iterator[BinaryPackageId]: 

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

260 

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

262 """ 

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

264 

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

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

267 

268 :param pkg: which BinaryPackage to check 

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

270 """ 

271 newest_src_in_suite = self.sources[pkg.source] 

272 return pkg.source_version != newest_src_in_suite.version 

273 

274 

275class TargetSuite(Suite): 

276 inst_tester: "InstallabilityTester" 

277 

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

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

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

281 self._logger = logging.getLogger(logger_name) 

282 

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

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

285 

286 :param pkg_id: A BinaryPackageId 

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

288 """ 

289 return self.inst_tester.is_installable(pkg_id) 

290 

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

292 """Add a binary package to the suite 

293 

294 :param pkg_id: The id of the package 

295 :raises KeyError: if the package is not known 

296 """ 

297 

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

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

300 # more sense to do that here instead 

301 self.inst_tester.add_binary(pkg_id) 

302 self._all_binaries_in_suite = None 

303 

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

305 """Remove a binary from the suite 

306 

307 :param pkg_id: The id of the package 

308 :raises KeyError: if the package is not known 

309 """ 

310 

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

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

313 # more sense to do that here instead 

314 self.inst_tester.remove_binary(pkg_id) 

315 self._all_binaries_in_suite = None 

316 

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

318 sources_t = self.sources 

319 binaries_t = self.binaries 

320 logger = self._logger 

321 issues_found = False 

322 

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

324 

325 for binaries in binaries_t.values(): 

326 for pkg_name, pkg in binaries.items(): 

327 src = pkg.source 

328 

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

330 issues_found = True 

331 logger.error( 

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

333 comment, 

334 src, 

335 pkg_name, 

336 src, 

337 ) 

338 

339 for src, source_data in sources_t.items(): 

340 for pkg_id in source_data.binaries: 

341 if ( 

342 pkg_id.package_name not in binaries_t[pkg_id.architecture] 

343 ): # pragma: no cover 

344 issues_found = True 

345 logger.error( 

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

347 comment, 

348 pkg_id.package_name, 

349 src, 

350 pkg_id.architecture, 

351 ) 

352 

353 if issues_found: # pragma: no cover 

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

355 

356 

357@dataclass(slots=True) 

358class Suites: 

359 target_suite: TargetSuite 

360 source_suites: list[Suite] 

361 _suites: dict[str, Suite] = field(init=False, default_factory=dict) 

362 by_name_or_alias: dict[str, Suite] = field(init=False, default_factory=dict) 

363 

364 def __post_init__(self) -> None: 

365 self._suites[self.target_suite.name] = self.target_suite 

366 self.by_name_or_alias[self.target_suite.name] = self.target_suite 

367 if self.target_suite.suite_short_name: 367 ↛ 368line 367 didn't jump to line 368

368 self.by_name_or_alias[self.target_suite.suite_short_name] = ( 

369 self.target_suite 

370 ) 

371 for suite in self.source_suites: 

372 self._suites[suite.name] = suite 

373 self.by_name_or_alias[suite.name] = suite 

374 if suite.suite_short_name: 

375 self.by_name_or_alias[suite.suite_short_name] = suite 

376 

377 @property 

378 def primary_source_suite(self) -> Suite: 

379 return self.source_suites[0] 

380 

381 @property 

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

383 return self.source_suites[1:] 

384 

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

386 return self._suites[item] 

387 

388 def __len__(self) -> int: 

389 return len(self.source_suites) + 1 

390 

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

392 return item in self._suites 

393 

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

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

396 yield from self.source_suites 

397 yield self.target_suite