# -*- coding: utf-8 -*- # Copyright (C) 2011 Adam D. Barratt # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. import apt_pkg import logging from britney2 import SuiteClass class MigrationItem(object): def __init__(self, package=None, version=None, architecture=None, suite=None, is_removal=False, is_cruft_removal=False): if architecture is None: architecture = 'source' if is_cruft_removal: is_removal = True self._package = package self._version = version self._architecture = architecture self._suite = suite self._is_removal = is_removal self._is_cruft_removal = is_cruft_removal self._uvname = self.get_uvname() self._name = self.get_name() def get_name(self): name = self._package if self._architecture != "source": name = "%s/%s" % (name, self._architecture) if self._version: name = "%s/%s" % (name, self._version) if self._suite.excuses_suffix: name = "%s_%s" % (name, self._suite.excuses_suffix) if self._is_removal: name = "-%s" % (name) return name def get_uvname(self): name = self._package if self._architecture != "source": name = "%s/%s" % (name, self._architecture) if self._suite.excuses_suffix: name = "%s_%s" % (name, self._suite.excuses_suffix) if self._is_removal: name = "-%s" % (name) return name def __str__(self): if self.version is not None: return self.name else: return self.uvname def __eq__(self, other): isequal = False if self.uvname == other.uvname: if self.version is None or other.version is None: isequal = True else: isequal = self.version == other.version return isequal def __hash__(self): if not self.version: raise AssertionError("trying to hash unversioned MigrationItem: %s" % (self.name)) return hash((self.uvname, self.version)) def __lt__(self, other): return (self.uvname, self.version) < (other.uvname, other.version) @property def name(self): return self._name @property def is_removal(self): return self._is_removal @property def architecture(self): return self._architecture @property def package(self): return self._package @property def suite(self): return self._suite @property def version(self): return self._version @property def uvname(self): return self._uvname @property def is_cruft_removal(self): return self._is_cruft_removal class MigrationItemFactory(object): def __init__(self, suites): self._suites = suites self._all_architectures = frozenset(suites.target_suite.binaries) logger_name = ".".join((self.__class__.__module__, self.__class__.__name__)) self.logger = logging.getLogger(logger_name) def generate_removal_for_cruft_item(self, pkg_id): return MigrationItem(package=pkg_id.package_name, version=pkg_id.version, architecture=pkg_id.architecture, suite=self._suites.target_suite, is_cruft_removal=True, ) @staticmethod def _is_right_version(suite, package_name, expected_version): if package_name not in suite.sources: return False actual_version = suite.sources[package_name].version if apt_pkg.version_compare(actual_version, expected_version) != 0: return False return True def _find_suite_for_item(self, suites, suite_name, package_name, version, auto_correct): suite = suites.by_name_or_alias[suite_name] assert suite.suite_class != SuiteClass.TARGET_SUITE if version is not None and auto_correct and not self._is_right_version(suite, package_name, version): for s in suites.source_suites: if self._is_right_version(s, package_name, version): suite = s break return suite def parse_item(self, item_text, versioned=True, auto_correct=True): """ :param item_text: The string describing the item (e.g. "glibc/2.5") :param versioned: If true, a two-part item is assumed to be versioned. otherwise, it is assumed to be versionless. This determines how items like "foo/bar" is parsed (if versioned, "bar" is assumed to be a version and otherwise "bar" is assumed to be an architecture). If in doubt, use versioned=True with auto_correct=True and the code will figure it out on its own. :param auto_correct: If True, minor issues are automatically fixed where possible. This includes handling architecture and version being in the wrong order and missing/omitting a suite reference for items. This feature is useful for migration items provided by humans (e.g. via hints) to avoid rejecting the input over trivial/minor issues with the input. When False, there will be no attempt to correct the migration input. :return: A MigrationItem matching the spec """ suites = self._suites version = None architecture = None is_removal = False if item_text.startswith('-'): item_text = item_text[1:] is_removal = True parts = item_text.split('/', 3) package_name = parts[0] suite_name = suites.primary_source_suite.name if '_' in package_name: package_name, suite_name = package_name.split('_', 2) if len(parts) == 3: architecture = parts[1] version = parts[2] elif len(parts) == 2: if versioned: version = parts[1] else: architecture = parts[1] if auto_correct and version in self._all_architectures: (architecture, version) = (version, architecture) if architecture is None: architecture = 'source' if '_' in architecture: architecture, suite_name = architecture.split('_', 2) if is_removal: suite = suites.target_suite else: suite = self._find_suite_for_item(suites, suite_name, package_name, version, auto_correct) return MigrationItem(package=package_name, version=version, architecture=architecture, suite=suite, is_removal=is_removal, ) def parse_items(self, *args, **kwargs): return [self.parse_item(x, **kwargs) for x in args]