Coverage for britney2/migrationitem.py: 96%
139 statements
« prev ^ index » next coverage.py v6.5.0, created at 2025-03-23 07:34 +0000
1# -*- coding: utf-8 -*-
3# Copyright (C) 2011 Adam D. Barratt <adsb@debian.org>
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.
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.
15import logging
16from typing import Any, Optional
18import apt_pkg
20from britney2 import BinaryPackageId, Suite, SuiteClass, Suites
23class MigrationItem(object):
24 def __init__(
25 self,
26 package: str,
27 suite: Suite,
28 *,
29 version: Optional[str] = None,
30 architecture: Optional[str] = None,
31 is_removal: bool = False,
32 is_cruft_removal: bool = False,
33 ):
34 if architecture is None:
35 architecture = "source"
37 if is_cruft_removal:
38 is_removal = True
40 self._package = package
41 self._version = version
42 self._architecture = architecture
43 self._suite = suite
44 self._is_removal = is_removal
45 self._is_cruft_removal = is_cruft_removal
46 self._uvname = self.get_uvname()
47 self._name = self.get_name()
49 def get_name(self) -> str:
50 name = self._package
51 if self._architecture != "source":
52 name = "%s/%s" % (name, self._architecture)
53 if self._version:
54 name = "%s/%s" % (name, self._version)
55 if self._suite.excuses_suffix:
56 name = "%s_%s" % (name, self._suite.excuses_suffix)
57 if self._is_removal:
58 name = "-%s" % (name)
59 return name
61 def get_uvname(self) -> str:
62 name = self._package
63 if self._architecture != "source":
64 name = "%s/%s" % (name, self._architecture)
65 if self._suite.excuses_suffix:
66 name = "%s_%s" % (name, self._suite.excuses_suffix)
67 if self._is_removal:
68 name = "-%s" % (name)
69 return name
71 def __repr__(self) -> str:
72 return "MI(%s)" % (self.__str__())
74 def __str__(self) -> str:
75 if self.version is not None: 75 ↛ 78line 75 didn't jump to line 78, because the condition on line 75 was never false
76 return self.name
77 else:
78 return self.uvname
80 def __eq__(self, other: object) -> bool:
81 isequal = False
82 if isinstance(other, MigrationItem): 82 ↛ 89line 82 didn't jump to line 89, because the condition on line 82 was never false
83 if self.uvname == other.uvname:
84 if self.version is None or other.version is None:
85 isequal = True
86 else:
87 isequal = self.version == other.version
89 return isequal
91 def __hash__(self) -> int:
92 if not self.version: 92 ↛ 93line 92 didn't jump to line 93, because the condition on line 92 was never true
93 raise AssertionError(
94 "trying to hash unversioned MigrationItem: %s" % (self.name)
95 )
97 return hash((self.uvname, self.version))
99 def __lt__(self, other: "MigrationItem") -> bool:
100 return (self.uvname, self.version) < (other.uvname, other.version)
102 @property
103 def name(self) -> str:
104 return self._name
106 @property
107 def is_removal(self) -> bool:
108 return self._is_removal
110 @property
111 def architecture(self) -> str:
112 return self._architecture
114 @property
115 def package(self) -> str:
116 return self._package
118 @property
119 def suite(self) -> Suite:
120 return self._suite
122 @property
123 def version(self) -> Optional[str]:
124 return self._version
126 @property
127 def uvname(self) -> str:
128 return self._uvname
130 @property
131 def is_cruft_removal(self) -> bool:
132 return self._is_cruft_removal
135class MigrationItemFactory(object):
136 def __init__(self, suites: Suites) -> None:
137 self._suites = suites
138 self._all_architectures = frozenset(suites.target_suite.binaries)
139 logger_name = ".".join((self.__class__.__module__, self.__class__.__name__))
140 self.logger = logging.getLogger(logger_name)
142 def generate_removal_for_cruft_item(
143 self, pkg_id: "BinaryPackageId"
144 ) -> MigrationItem:
145 return MigrationItem(
146 package=pkg_id.package_name,
147 version=pkg_id.version,
148 architecture=pkg_id.architecture,
149 suite=self._suites.target_suite,
150 is_cruft_removal=True,
151 )
153 @staticmethod
154 def _is_right_version(
155 suite: Suite, package_name: str, expected_version: str
156 ) -> bool:
157 if package_name not in suite.sources:
158 return False
160 actual_version = suite.sources[package_name].version
161 if apt_pkg.version_compare(actual_version, expected_version) != 0:
162 return False
164 return True
166 def _find_suite_for_item(
167 self,
168 suites: Suites,
169 suite_name: str,
170 package_name: str,
171 version: Optional[str],
172 auto_correct: bool,
173 ) -> Suite:
174 suite = suites.by_name_or_alias[suite_name]
175 assert suite.suite_class != SuiteClass.TARGET_SUITE
176 if (
177 version is not None
178 and auto_correct
179 and not self._is_right_version(suite, package_name, version)
180 ):
181 for s in suites.source_suites:
182 if self._is_right_version(s, package_name, version):
183 suite = s
184 break
185 return suite
187 def parse_item(
188 self, item_text: str, versioned: bool = True, auto_correct: bool = True
189 ) -> MigrationItem:
190 """
192 :param item_text: The string describing the item (e.g. "glibc/2.5")
193 :param versioned: If true, a two-part item is assumed to be versioned.
194 otherwise, it is assumed to be versionless. This determines how
195 items like "foo/bar" is parsed (if versioned, "bar" is assumed to
196 be a version and otherwise "bar" is assumed to be an architecture).
197 If in doubt, use versioned=True with auto_correct=True and the
198 code will figure it out on its own.
199 :param auto_correct: If True, minor issues are automatically fixed
200 where possible. This includes handling architecture and version
201 being in the wrong order and missing/omitting a suite reference
202 for items. This feature is useful for migration items provided
203 by humans (e.g. via hints) to avoid rejecting the input over
204 trivial/minor issues with the input.
205 When False, there will be no attempt to correct the migration
206 input.
207 :return: A MigrationItem matching the spec
208 """
209 suites = self._suites
210 version = None
211 architecture = None
212 is_removal = False
213 if item_text.startswith("-"):
214 item_text = item_text[1:]
215 is_removal = True
216 parts = item_text.split("/", 3)
217 package_name = parts[0]
218 suite_name = suites.primary_source_suite.name
219 if "_" in package_name:
220 package_name, suite_name = package_name.split("_", 2)
222 if len(parts) == 3:
223 architecture = parts[1]
224 version = parts[2]
225 elif len(parts) == 2:
226 if versioned: 226 ↛ 229line 226 didn't jump to line 229, because the condition on line 226 was never false
227 version = parts[1]
228 else:
229 architecture = parts[1]
231 if auto_correct and version in self._all_architectures:
232 (architecture, version) = (version, architecture)
234 if architecture is None:
235 architecture = "source"
237 if "_" in architecture:
238 architecture, suite_name = architecture.split("_", 2)
240 if is_removal:
241 suite: Suite = suites.target_suite
242 else:
243 suite = self._find_suite_for_item(
244 suites, suite_name, package_name, version, auto_correct
245 )
247 return MigrationItem(
248 package=package_name,
249 version=version,
250 architecture=architecture,
251 suite=suite,
252 is_removal=is_removal,
253 )
255 def parse_items(self, *args: Any, **kwargs: Any) -> list[MigrationItem]:
256 return [self.parse_item(x, **kwargs) for x in args]