| #!/usr/bin/env python |
| # SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception |
| # -*- coding: utf-8 -*- |
| # -*- Mode: Python |
| # |
| # Copyright (C) 2013-2016 Red Hat, Inc. |
| # |
| # Author: Chenxiong Qi |
| |
| from __future__ import print_function |
| |
| import argparse |
| import functools |
| import glob |
| import logging |
| import mimetypes |
| import os |
| import re |
| import shutil |
| import six |
| import subprocess |
| import sys |
| |
| from collections import namedtuple |
| from itertools import chain |
| |
| import xdg.BaseDirectory |
| |
| import rpm |
| import koji |
| |
| # @file |
| # |
| # You might have known that abipkgdiff is a command line tool to compare two |
| # RPM packages to find potential differences of ABI. This is really useful for |
| # Fedora packagers and developers. Usually, excpet the RPM packages built |
| # locally, if a packager wants to compare RPM packages he just built with |
| # specific RPM packages that were already built and availabe in Koji, |
| # fedabipkgdiff is the right tool for him. |
| # |
| # With fedabipkgdiff, packager is able to specify certain criteria to tell |
| # fedabipkgdiff which RPM packages he wants to compare, then fedabipkgdiff will |
| # find them, download them, and boom, run the abipkgdiff for you. |
| # |
| # Currently, fedabipkgdiff returns 0 if everything works well, otherwise, 1 if |
| # something wrong. |
| |
| |
| # First, try proper Koji initialization. |
| try: |
| koji_config = koji.read_config('koji') |
| DEFAULT_KOJI_SERVER = koji_config['server'] |
| DEFAULT_KOJI_TOPURL = koji_config['topurl'] |
| except koji.ConfigurationError: |
| # ..., but if that fails because of a rather strict interpretation where |
| # 'read_config' looks for configuration files, just use dummy values. |
| # These fail upon use unless overridden, which for libabigail test suite |
| # usage they always are (all relevant artifacts are shipped in the |
| # libabigail distribution). |
| DEFAULT_KOJI_SERVER = 'dummy_DEFAULT_KOJI_SERVER' |
| DEFAULT_KOJI_TOPURL = 'dummy_DEFAULT_KOJI_TOPURL' |
| |
| |
| # The working directory where to hold all data including downloaded RPM |
| # packages Currently, it's not configurable and hardcode here. In the future |
| # version of fedabipkgdiff, I'll make it configurable by users. |
| HOME_DIR = os.path.join(xdg.BaseDirectory.xdg_cache_home, |
| os.path.splitext(os.path.basename(__file__))[0]) |
| |
| DEFAULT_ABIPKGDIFF = 'abipkgdiff' |
| |
| # Mask for determining if underlying fedabipkgdiff succeeds or not. |
| # This is for when the compared ABIs are equal |
| ABIDIFF_OK = 0 |
| # This bit is set if there an application error. |
| ABIDIFF_ERROR = 1 |
| # This bit is set if the tool is invoked in an non appropriate manner. |
| ABIDIFF_USAGE_ERROR = 1 << 1 |
| # This bit is set if the ABIs being compared are different. |
| ABIDIFF_ABI_CHANGE = 1 << 2 |
| |
| |
| # Used to construct abipkgdiff command line argument, package and associated |
| # debuginfo package |
| # fedabipkgdiff runs abipkgdiff in this form |
| # |
| # abipkgdiff \ |
| # --d1 /path/to/package1-debuginfo.rpm \ |
| # --d2 /path/to/package2-debuginfo.rpm \ |
| # /path/to/package1.rpm \ |
| # /path/to/package2.rpm |
| # |
| # ComparisonHalf is a three-elements tuple in format |
| # |
| # (package1.rpm, [package1-debuginfo.rpm..] package1-devel.rpm) |
| # |
| # - the first element is the subject representing the package to |
| # compare. It's a dict representing the RPM we are interested in. |
| # That dict was retrieved from Koji XMLRPC API. |
| # - the rest are ancillary packages used for the comparison. So, the |
| # second one is a vector containing the needed debuginfo packages |
| # (yes there can be more than one), and the last one is the package |
| # containing API of the ELF shared libraries carried by subject. |
| # All the packages are dicts representing RPMs and those dicts were |
| # retrieved fromt he KOji XMLRPC API. |
| # |
| # So, before calling abipkgdiff, fedabipkgdiff must prepare and pass |
| # the following information |
| # |
| # (/path/to/package1.rpm, [/paths/to/package1-debuginfo.rpm ..] /path/to/package1-devel.rpm) |
| # (/path/to/package2.rpm, [/paths/to/package2-debuginfo.rpm ..] /path/to/package1-devel.rpm) |
| # |
| ComparisonHalf = namedtuple('ComparisonHalf', |
| ['subject', 'ancillary_debug', 'ancillary_devel']) |
| |
| |
| global_config = None |
| pathinfo = None |
| session = None |
| |
| # There is no way to configure the log format so far. I hope I would have time |
| # to make it available so that if fedabipkgdiff is scheduled and run by some |
| # service, the logs logged into log file is muc usable. |
| logging.basicConfig(format='[%(levelname)s] %(message)s', level=logging.CRITICAL) |
| logger = logging.getLogger(os.path.basename(__file__)) |
| |
| |
| class KojiPackageNotFound(Exception): |
| """Package is not found in Koji""" |
| |
| |
| class PackageNotFound(Exception): |
| """Package is not found locally""" |
| |
| |
| class RpmNotFound(Exception): |
| """RPM is not found""" |
| |
| |
| class NoBuildsError(Exception): |
| """No builds returned from a method to select specific builds""" |
| |
| |
| class NoCompleteBuilds(Exception): |
| """No complete builds for a package |
| |
| This is a serious problem, nothing can be done if there is no complete |
| builds for a package. |
| """ |
| |
| |
| class InvalidDistroError(Exception): |
| """Invalid distro error""" |
| |
| |
| class CannotFindLatestBuildError(Exception): |
| """Cannot find latest build from a package""" |
| |
| |
| class SetCleanCacheAction(argparse._StoreTrueAction): |
| """Custom Action making clean-cache as bundle of clean-cache-before and clean-cache-after""" |
| |
| def __call__(self, parser, namespace, values, option_string=None): |
| setattr(namespace, 'clean_cache_before', self.const) |
| setattr(namespace, 'clean_cache_after', self.const) |
| |
| |
| def is_distro_valid(distro): |
| """Adjust if a distro is valid |
| |
| Currently, check for Fedora and RHEL. |
| |
| :param str distro: a string representing a distro value. |
| :return: True if distro is the one specific to Fedora, like fc24, el7. |
| "rtype: bool |
| """ |
| return re.match(r'^(fc|el)\d{1,2}$', distro) is not None |
| |
| |
| def get_distro_from_string(str): |
| """Get the part of a string that designates the Fedora distro version number |
| |
| For instance, when passed the string '2.3.fc12', this function |
| returns the string 'fc12'. |
| |
| :param str the string to consider |
| :return: The sub-string of the parameter that represents the |
| Fedora distro version number, or None if the parameter does not |
| contain such a sub-string. |
| """ |
| |
| m = re.match(r'(.*)((fc|el)\d{1,2})(.*)', str) |
| if not m: |
| return None |
| |
| distro = m.group(2) |
| return distro |
| |
| |
| def match_nvr(s): |
| """Determine if a string is a N-V-R""" |
| return re.match(r'^([^/]+)-(.+)-(.+)$', s) is not None |
| |
| |
| def match_nvra(s): |
| """Determine if a string is a N-V-R.A""" |
| return re.match(r'^([^/]+)-(.+)-(.+)\.(.+)$', s) is not None |
| |
| |
| def is_rpm_file(filename): |
| """Return if a file is a RPM""" |
| isfile = os.path.isfile(filename) |
| mimetype = mimetypes.guess_type(filename)[0] if isfile else None |
| isrpm = (mimetype == 'application/x-redhat-package-manager' |
| or mimetype == 'application/x-rpm') |
| |
| # Most systems won't have rpm defined as a mimetype |
| if not mimetype and filename.endswith('.rpm'): |
| isrpm = True |
| logger.debug('is_rpm_file(\'%s\'): isfile=%s, mimetype=\'%s\', isrpm=%s', |
| filename, isfile, mimetype, isrpm) |
| return isrpm |
| |
| |
| def cmp_nvr(left, right): |
| """Compare function for sorting a sequence of NVRs |
| |
| This is the compare function used in sorted function to sort builds so that |
| fedabipkgdiff is able to select the latest build. Return value follows the |
| rules described in the part of paramter cmp of sorted documentation. |
| |
| :param str left: left nvr to compare. |
| :param str right: right nvr to compare. |
| :return: -1, 0, or 1 that represents left is considered smaller than, |
| equal to, or larger than the right individually. |
| :rtype: int |
| """ |
| left_nvr = koji.parse_NVR(left['nvr']) |
| right_nvr = koji.parse_NVR(right['nvr']) |
| return rpm.labelCompare( |
| (left_nvr['epoch'], left_nvr['version'], left_nvr['release']), |
| (right_nvr['epoch'], right_nvr['version'], right_nvr['release'])) |
| |
| |
| def log_call(func): |
| """A decorator that logs a method invocation |
| |
| Method's name and all arguments, either positional or keyword arguments, |
| will be logged by logger.debug. Also, return value from the decorated |
| method will be logged just after the invocation is done. |
| |
| This decorator does not catch any exception thrown from the decorated |
| method. If there is any exception thrown from decorated method, you can |
| catch them in the caller and obviously, no return value is logged. |
| |
| :param callable func: a callable object to decorate |
| """ |
| def proxy(*args, **kwargs): |
| logger.debug('Call %s, args: %s, kwargs: %s', |
| func.__name__, |
| args if args else '', |
| kwargs if kwargs else '') |
| result = func(*args, **kwargs) |
| logger.debug('Result from %s: %s', func.__name__, result) |
| return result |
| return proxy |
| |
| |
| def delete_download_cache(): |
| """Delete download cache directory""" |
| download_dir = get_download_dir() |
| if global_config.dry_run: |
| print('DRY-RUN: Delete cached downloaded RPM packages at {0}'.format(download_dir)) |
| else: |
| logger.debug('Delete cached downloaded RPM packages at {0}'.format(download_dir)) |
| shutil.rmtree(download_dir) |
| |
| |
| class RPM(object): |
| """Wrapper around an RPM descriptor received from Koji |
| |
| The RPM descriptor that is returned from Koji XMLRPC API is a |
| dict. This wrapper class makes it eaiser to access all these |
| properties in the way of object.property. |
| """ |
| |
| def __init__(self, rpm_info): |
| """Initialize a RPM object |
| |
| :param dict rpm_info: a dict representing an RPM descriptor |
| received from the Koji API, either listRPMs or getRPM |
| """ |
| self.rpm_info = rpm_info |
| |
| def __str__(self): |
| """Return the string representation of this RPM |
| |
| Return the string representation of RPM information returned from Koji |
| directly so that RPM can be treated in same way. |
| """ |
| return str(self.rpm_info) |
| |
| def __getattr__(self, name): |
| """Access RPM information in the way of object.property |
| |
| :param str name: the property name to access. |
| :raises AttributeError: if name is not one of keys of RPM information. |
| """ |
| if name in self.rpm_info: |
| return self.rpm_info[name] |
| else: |
| raise AttributeError('No attribute name {0}'.format(name)) |
| |
| def is_peer(self, another_rpm): |
| """Determine if this is the peer of a given rpm. |
| |
| Here is what "peer" means. |
| |
| Consider a package P for which the tripplet Name, Version, |
| Release is made of the values {N,V,R}. Then, consider a |
| package P' for which the similar tripplet is {N', V', R'}. |
| |
| P' is a peer of P if N == N', and either V != V' or R != R'. |
| given package with a given NVR is another package with a N'V' |
| """ |
| return self.name == another_rpm.name and \ |
| self.arch == another_rpm.arch and \ |
| not (self.version == another_rpm.version |
| and self.release == another_rpm.release) |
| |
| @property |
| def nvra(self): |
| """Return a RPM's N-V-R-A representation |
| |
| An example: libabigail-1.0-0.8.rc4.1.fc23.x86_64 |
| """ |
| nvra, _ = os.path.splitext(self.filename) |
| return nvra |
| |
| @property |
| def filename(self): |
| """Return a RPM file name |
| |
| An example: libabigail-1.0-0.8.rc4.1.fc23.x86_64.rpm |
| """ |
| return os.path.basename(pathinfo.rpm(self.rpm_info)) |
| |
| @property |
| def is_debuginfo(self): |
| """Check if the name of the current RPM denotes a debug info package""" |
| return koji.is_debuginfo(self.rpm_info['name']) |
| |
| @property |
| def is_devel(self): |
| """Check if the name of current RPM denotes a development package""" |
| return self.rpm_info['name'].endswith('-devel') |
| |
| @property |
| def download_url(self): |
| """Get the URL from where to download this RPM""" |
| build = session.getBuild(self.build_id) |
| return os.path.join(pathinfo.build(build), pathinfo.rpm(self.rpm_info)) |
| |
| @property |
| def downloaded_file(self): |
| """Get a pridictable downloaded file name with absolute path""" |
| # arch should be removed from the result returned from PathInfo.rpm |
| filename = os.path.basename(pathinfo.rpm(self.rpm_info)) |
| return os.path.join(get_download_dir(), filename) |
| |
| @property |
| def is_downloaded(self): |
| """Check if this RPM was already downloaded to local disk""" |
| return os.path.exists(self.downloaded_file) |
| |
| |
| class LocalRPM(RPM): |
| """Representing a local RPM |
| |
| Local RPM means the one that could be already downloaded or built from |
| where I can find it |
| """ |
| |
| def __init__(self, filename): |
| """Initialize local RPM with a filename |
| |
| :param str filename: a filename pointing to a RPM file in local |
| disk. Note that, this file must not exist necessarily. |
| """ |
| self.local_filename = filename |
| self.rpm_info = koji.parse_NVRA(os.path.basename(filename)) |
| |
| @property |
| def downloaded_file(self): |
| """Return filename of this RPM |
| |
| Returned filename is just the one passed when initializing this RPM. |
| |
| :return: filename of this RPM |
| :rtype: str |
| """ |
| return self.local_filename |
| |
| @property |
| def download_url(self): |
| raise NotImplementedError('LocalRPM has no URL to download') |
| |
| def _find_rpm(self, rpm_filename): |
| """Search an RPM from the directory of the current instance of LocalRPM |
| |
| :param str rpm_filename: filename of rpm to find, for example |
| foo-devel-0.1-1.fc24. |
| :return: an instance of LocalRPM representing the found rpm, or None if |
| no RPM was found. |
| """ |
| search_dir = os.path.dirname(os.path.abspath(self.local_filename)) |
| filename = os.path.join(search_dir, rpm_filename) |
| return LocalRPM(filename) if os.path.exists(filename) else None |
| |
| @log_call |
| def find_debuginfo(self): |
| """Find debuginfo RPM package from a directory""" |
| filename = \ |
| '%(name)s-debuginfo-%(version)s-%(release)s.%(arch)s.rpm' % \ |
| self.rpm_info |
| return self._find_rpm(filename) |
| |
| @log_call |
| def find_devel(self): |
| """Find development package from a directory""" |
| filename = \ |
| '%(name)s-devel-%(version)s-%(release)s.%(arch)s.rpm' % \ |
| self.rpm_info |
| return self._find_rpm(filename) |
| |
| |
| class RPMCollection(object): |
| """Collection of RPMs |
| |
| This is a simple collection containing RPMs collected from a |
| directory on the local filesystem or retrieved from Koji. |
| |
| A collection can contain one or more sets of RPMs. Each set of |
| RPMs being for a particular architecture. |
| |
| For a given architecture, a set of RPMs is made of one RPM and its |
| ancillary RPMs. An ancillary RPM is either a debuginfo RPM or a |
| devel RPM. |
| |
| So a given RPMCollection would (informally) look like: |
| |
| { |
| i686 => {foo.i686.rpm, foo-debuginfo.i686.rpm, foo-devel.i686.rpm} |
| x86_64 => {foo.x86_64.rpm, foo-debuginfo.x86_64.rpm, foo-devel.x86_64.rpm,} |
| } |
| |
| """ |
| |
| def __init__(self, rpms=None): |
| # Mapping from arch to a list of rpm_infos. |
| # Note that *all* RPMs of the collections are present in this |
| # map; that is the RPM to consider and its ancillary RPMs. |
| self.rpms = {} |
| |
| # Mapping from arch to another mapping containing index of debuginfo |
| # and development package |
| # e.g. |
| # self.ancillary_rpms = {'i686', {'debuginfo': foo-debuginfo.rpm, |
| # 'devel': foo-devel.rpm}} |
| self.ancillary_rpms = {} |
| |
| if rpms: |
| for rpm in rpms: |
| self.add(rpm) |
| |
| @classmethod |
| def gather_from_dir(cls, rpm_file, all_rpms=None): |
| """Gather RPM collection from local directory""" |
| dir_name = os.path.dirname(os.path.abspath(rpm_file)) |
| filename = os.path.basename(rpm_file) |
| |
| nvra = koji.parse_NVRA(filename) |
| rpm_files = glob.glob(os.path.join( |
| dir_name, '*-%(version)s-%(release)s.%(arch)s.rpm' % nvra)) |
| rpm_col = cls() |
| |
| if all_rpms: |
| selector = lambda rpm: True |
| else: |
| selector = lambda rpm: local_rpm.is_devel or \ |
| local_rpm.is_debuginfo or local_rpm.filename == filename |
| |
| found_debuginfo = 1 |
| |
| for rpm_file in rpm_files: |
| local_rpm = LocalRPM(rpm_file) |
| |
| if local_rpm.is_debuginfo: |
| found_debuginfo <<= 1 |
| if found_debuginfo == 4: |
| raise RuntimeError( |
| 'Found more than one debuginfo package in ' |
| 'this directory. At the moment, fedabipkgdiff ' |
| 'is not able to deal with this case. ' |
| 'Please create two separate directories and ' |
| 'put an RPM and its ancillary debuginfo and ' |
| 'devel RPMs in each directory.') |
| |
| if selector(local_rpm): |
| rpm_col.add(local_rpm) |
| |
| return rpm_col |
| |
| def add(self, rpm): |
| """Add a RPM into this collection""" |
| self.rpms.setdefault(rpm.arch, []).append(rpm) |
| |
| devel_debuginfo_default = {'debuginfo': None, 'devel': None} |
| |
| if rpm.is_debuginfo: |
| self.ancillary_rpms.setdefault( |
| rpm.arch, devel_debuginfo_default)['debuginfo'] = rpm |
| |
| if rpm.is_devel: |
| self.ancillary_rpms.setdefault( |
| rpm.arch, devel_debuginfo_default)['devel'] = rpm |
| |
| def rpms_iter(self, arches=None, default_behavior=True): |
| """Iterator of RPMs to go through RPMs with specific arches""" |
| arches = sorted(self.rpms.keys()) |
| |
| for arch in arches: |
| for _rpm in self.rpms[arch]: |
| yield _rpm |
| |
| def get_sibling_debuginfo(self, rpm): |
| """Get sibling debuginfo package of given rpm |
| |
| The sibling debuginfo is a debug info package for the |
| 'rpm'. Note that if there are several debuginfo packages |
| associated to 'rpm' and users want to get the one which name |
| matches exactly 'rpm', then they might want to use the member |
| function 'get_matching_debuginfo' instead. |
| |
| """ |
| if rpm.arch not in self.ancillary_rpms: |
| return None |
| return self.ancillary_rpms[rpm.arch].get('debuginfo') |
| |
| def get_matching_debuginfo(self, rpm): |
| """Get the debuginfo package that matches a given one """ |
| all_debuginfo_list = self.get_all_debuginfo_rpms(rpm) |
| debuginfo_pkg = None |
| for d in all_debuginfo_list: |
| if d.name == '{0}-debuginfo'.format(rpm.name): |
| debuginfo_pkg = d |
| break |
| if not debuginfo_pkg: |
| debuginfo_pkg = self.get_sibling_debuginfo(rpm) |
| |
| return debuginfo_pkg |
| |
| def get_sibling_devel(self, rpm): |
| """Get sibling devel package of given rpm""" |
| if rpm.arch not in self.ancillary_rpms: |
| return None |
| return self.ancillary_rpms[rpm.arch].get('devel') |
| |
| def get_peer_rpm(self, rpm): |
| """Get peer rpm of rpm from this collection""" |
| if rpm.arch not in self.rpms: |
| return None |
| for _rpm in self.rpms[rpm.arch]: |
| if _rpm.is_peer(rpm): |
| return _rpm |
| return None |
| |
| def get_all_debuginfo_rpms(self, rpm_info): |
| """Return a list of descriptors of all the debuginfo RPMs associated |
| to a given RPM. |
| |
| :param: dict rpm_info a dict representing an RPM. This was |
| received from the Koji API, either from listRPMs or getRPM. |
| :return: a list of dicts containing RPM descriptors (dicts) |
| for the debuginfo RPMs associated to rpm_info |
| :retype: dict |
| """ |
| rpm_infos = self.rpms[rpm_info.arch] |
| result = [] |
| for r in rpm_infos: |
| if r.is_debuginfo: |
| result.append(r) |
| return result |
| |
| |
| def generate_comparison_halves(rpm_col1, rpm_col2): |
| """Iterate RPM collection and peer's to generate comparison halves""" |
| for _rpm in rpm_col1.rpms_iter(): |
| if _rpm.is_debuginfo: |
| continue |
| if _rpm.is_devel and not global_config.check_all_subpackages: |
| continue |
| |
| if global_config.self_compare: |
| rpm2 = _rpm |
| else: |
| rpm2 = rpm_col2.get_peer_rpm(_rpm) |
| if rpm2 is None: |
| logger.warning('Peer RPM of {0} is not found.'.format(_rpm.filename)) |
| continue |
| |
| debuginfo_list1 = [] |
| debuginfo_list2 = [] |
| |
| # If this is a *devel* package we are looking at, then get all |
| # the debug info packages associated to with the main package |
| # and stick them into the resulting comparison half. |
| |
| if _rpm.is_devel: |
| debuginfo_list1 = rpm_col1.get_all_debuginfo_rpms(_rpm) |
| else: |
| debuginfo_list1.append(rpm_col1.get_matching_debuginfo(_rpm)) |
| |
| devel1 = rpm_col1.get_sibling_devel(_rpm) |
| |
| if global_config.self_compare: |
| debuginfo_list2 = debuginfo_list1 |
| devel2 = devel1 |
| else: |
| if rpm2.is_devel: |
| debuginfo_list2 = rpm_col2.get_all_debuginfo_rpms(rpm2) |
| else: |
| debuginfo_list2.append(rpm_col2.get_matching_debuginfo(rpm2)) |
| devel2 = rpm_col2.get_sibling_devel(rpm2) |
| |
| yield (ComparisonHalf(subject=_rpm, |
| ancillary_debug=debuginfo_list1, |
| ancillary_devel=devel1), |
| ComparisonHalf(subject=rpm2, |
| ancillary_debug=debuginfo_list2, |
| ancillary_devel=devel2)) |
| |
| |
| class Brew(object): |
| """Interface to Koji XMLRPC API with enhancements specific to fedabipkgdiff |
| |
| kojihub XMLRPC APIs are well-documented in koji's source code. For more |
| details information, please refer to class RootExports within kojihub.py. |
| |
| For details of APIs used within fedabipkgdiff, refer to from line |
| |
| https://pagure.io/koji/blob/master/f/hub/kojihub.py#_7835 |
| """ |
| |
| def __init__(self, baseurl): |
| """Initialize Brew |
| |
| :param str baseurl: the kojihub URL to initialize a session, that is |
| used to access koji XMLRPC APIs. |
| """ |
| self.session = koji.ClientSession(baseurl) |
| |
| @log_call |
| def listRPMs(self, buildID=None, arches=None, selector=None): |
| """Get list of RPMs of a build from Koji |
| |
| Call kojihub.listRPMs to get list of RPMs. Return selected RPMs without |
| changing each RPM information. |
| |
| A RPM returned from listRPMs contains following keys: |
| |
| - id |
| - name |
| - version |
| - release |
| - nvr (synthesized for sorting purposes) |
| - arch |
| - epoch |
| - payloadhash |
| - size |
| - buildtime |
| - build_id |
| - buildroot_id |
| - external_repo_id |
| - external_repo_name |
| - metadata_only |
| - extra |
| |
| :param int buildID: id of a build from which to list RPMs. |
| :param arches: to restrict to list RPMs with specified arches. |
| :type arches: list or tuple |
| :param selector: called to determine if a RPM should be selected and |
| included in the final returned result. Selector must be a callable |
| object and accepts one parameter of a RPM. |
| :type selector: a callable object |
| :return: a list of RPMs, each of them is a dict object |
| :rtype: list |
| """ |
| if selector: |
| assert hasattr(selector, '__call__'), 'selector must be callable.' |
| rpms = self.session.listRPMs(buildID=buildID, arches=arches) |
| if selector: |
| rpms = [rpm for rpm in rpms if selector(rpm)] |
| return rpms |
| |
| @log_call |
| def getRPM(self, rpminfo): |
| """Get a RPM from koji |
| |
| Call kojihub.getRPM, and returns the result directly without any |
| change. |
| |
| When not found a RPM, koji.getRPM will return None, then |
| this method will raise RpmNotFound error immediately to claim what is |
| happening. I want to raise fedabipkgdiff specific error rather than |
| koji's GenericError and then raise RpmNotFound again, so I just simply |
| don't use strict parameter to call koji.getRPM. |
| |
| :param rpminfo: rpminfo may be a N-V-R.A or a map containing name, |
| version, release, and arch. For example, file-5.25-5.fc24.x86_64, and |
| `{'name': 'file', 'version': '5.25', 'release': '5.fc24', 'arch': |
| 'x86_64'}`. |
| :type rpminfo: str or dict |
| :return: a map containing RPM information, that contains same keys as |
| method `Brew.listRPMs`. |
| :rtype: dict |
| :raises RpmNotFound: if a RPM cannot be found with rpminfo. |
| """ |
| rpm = self.session.getRPM(rpminfo) |
| if rpm is None: |
| raise RpmNotFound('Cannot find RPM {0}'.format(rpminfo)) |
| return rpm |
| |
| @log_call |
| def listBuilds(self, packageID, state=None, topone=None, |
| selector=None, order_by=None, reverse=None): |
| """Get list of builds from Koji |
| |
| Call kojihub.listBuilds, and return selected builds without changing |
| each build information. |
| |
| By default, only builds with COMPLETE state are queried and returns |
| afterwards. |
| |
| :param int packageID: id of package to list builds from. |
| :param int state: build state. There are five states of a build in |
| Koji. fedabipkgdiff only cares about builds with COMPLETE state. If |
| state is omitted, builds with COMPLETE state are queried from Koji by |
| default. |
| :param bool topone: just return the top first build. |
| :param selector: a callable object used to select specific subset of |
| builds. Selector will be called immediately after Koji returns queried |
| builds. When each call to selector, a build is passed to |
| selector. Return True if select current build, False if not. |
| :type selector: a callable object |
| :param str order_by: the attribute name by which to order the builds, |
| for example, name, version, or nvr. |
| :param bool reverse: whether to order builds reversely. |
| :return: a list of builds, even if there is only one build. |
| :rtype: list |
| :raises TypeError: if selector is not callable, or if order_by is not a |
| string value. |
| """ |
| if state is None: |
| state = koji.BUILD_STATES['COMPLETE'] |
| |
| if selector is not None and not hasattr(selector, '__call__'): |
| raise TypeError( |
| '{0} is not a callable object.'.format(str(selector))) |
| |
| if order_by is not None and not isinstance(order_by, six.string_types): |
| raise TypeError('order_by {0} is invalid.'.format(order_by)) |
| |
| builds = self.session.listBuilds(packageID=packageID, state=state) |
| if selector is not None: |
| builds = [build for build in builds if selector(build)] |
| if order_by is not None: |
| # FIXME: is it possible to sort builds by using opts parameter of |
| # listBuilds |
| if order_by == 'nvr': |
| if six.PY2: |
| builds = sorted(builds, cmp=cmp_nvr, reverse=reverse) |
| else: |
| builds = sorted(builds, |
| key=functools.cmp_to_key(cmp_nvr), |
| reverse=reverse) |
| else: |
| builds = sorted( |
| builds, key=lambda b: b[order_by], reverse=reverse) |
| if topone: |
| builds = builds[0:1] |
| |
| return builds |
| |
| @log_call |
| def getPackage(self, name): |
| """Get a package from Koji |
| |
| :param str name: a package name. |
| :return: a mapping containing package information. For example, |
| `{'id': 1, 'name': 'package'}`. |
| :rtype: dict |
| """ |
| package = self.session.getPackage(name) |
| if package is None: |
| package = self.session.getPackage(name.rsplit('-', 1)[0]) |
| if package is None: |
| raise KojiPackageNotFound( |
| 'Cannot find package {0}.'.format(name)) |
| return package |
| |
| @log_call |
| def getBuild(self, buildID): |
| """Get a build from Koji |
| |
| Call kojihub.getBuild. Return got build directly without change. |
| |
| :param int buildID: id of build to get from Koji. |
| :return: the found build. Return None, if not found a build with |
| buildID. |
| :rtype: dict |
| """ |
| return self.session.getBuild(buildID) |
| |
| @log_call |
| def get_rpm_build_id(self, name, version, release, arch=None): |
| """Get build ID that contains a RPM with specific nvra |
| |
| If arch is not omitted, a RPM can be identified by its N-V-R-A. |
| |
| If arch is omitted, name is used to get associated package, and then |
| to get the build. |
| |
| Example: |
| |
| >>> brew = Brew('url to kojihub') |
| >>> brew.get_rpm_build_id('httpd', '2.4.18', '2.fc24') |
| >>> brew.get_rpm_build_id('httpd', '2.4.18', '2.fc25', 'x86_64') |
| |
| :param str name: name of a rpm |
| :param str version: version of a rpm |
| :param str release: release of a rpm |
| :param arch: arch of a rpm |
| :type arch: str or None |
| :return: id of the build from where the RPM is built |
| :rtype: dict |
| :raises KojiPackageNotFound: if name is not found from Koji if arch |
| is None. |
| """ |
| if arch is None: |
| package = self.getPackage(name) |
| selector = lambda item: item['version'] == version and \ |
| item['release'] == release |
| builds = self.listBuilds(packageID=package['id'], |
| selector=selector) |
| if not builds: |
| raise NoBuildsError( |
| 'No builds are selected from package {0}.'.format( |
| package['name'])) |
| return builds[0]['build_id'] |
| else: |
| rpm = self.getRPM({'name': name, |
| 'version': version, |
| 'release': release, |
| 'arch': arch, |
| }) |
| return rpm['build_id'] |
| |
| @log_call |
| def get_package_latest_build(self, package_name, distro): |
| """Get latest build from a package, for a particular distro. |
| |
| Example: |
| |
| >>> brew = Brew('url to kojihub') |
| >>> brew.get_package_latest_build('httpd', 'fc24') |
| |
| :param str package_name: from which package to get the latest build |
| :param str distro: which distro the latest build belongs to |
| :return: the found build |
| :rtype: dict or None |
| :raises NoCompleteBuilds: if there is no latest build of a package. |
| """ |
| package = self.getPackage(package_name) |
| selector = lambda item: item['release'].find(distro) > -1 |
| |
| builds = self.listBuilds(packageID=package['id'], |
| selector=selector, |
| order_by='nvr', |
| reverse=True) |
| if not builds: |
| # So we found no build which distro string exactly matches |
| # the 'distro' parameter. |
| # |
| # Now lets try to get builds which distro string are less |
| # than the value of the 'distro' parameter. This is for |
| # cases when, for instance, the build of package foo that |
| # is present in current Fedora 27 is foo-1.fc26. That |
| # build originates from Fedora 26 but is being re-used in |
| # Fedora 27. So we want this function to pick up that |
| # foo-1.fc26, even though we want the builds of foo that |
| # match the distro string fc27. |
| |
| selector = lambda build: get_distro_from_string(build['release']) and \ |
| get_distro_from_string(build['release']) <= distro |
| |
| builds = self.listBuilds(packageID=package['id'], |
| selector=selector, |
| order_by='nvr', |
| reverse=True); |
| |
| if not builds: |
| raise NoCompleteBuilds( |
| 'No complete builds of package {0}'.format(package_name)) |
| |
| return builds[0] |
| |
| @log_call |
| def select_rpms_from_a_build(self, build_id, package_name, arches=None, |
| select_subpackages=None): |
| """Select specific RPMs within a build |
| |
| RPMs could be filtered be specific criterias by the parameters. |
| |
| By default, fedabipkgdiff requires the RPM package, as well as |
| its associated debuginfo and devel packages. These three |
| packages are selected, and noarch and src are excluded. |
| |
| :param int build_id: from which build to select rpms. |
| :param str package_name: which rpm to select that matches this name. |
| :param arches: which arches to select. If arches omits, rpms with all |
| arches except noarch and src will be selected. |
| :type arches: list, tuple or None |
| :param bool select_subpackages: indicate whether to select all RPMs |
| with specific arch from build. |
| :return: a list of RPMs returned from listRPMs |
| :rtype: list |
| """ |
| excluded_arches = ('noarch', 'src') |
| |
| def rpms_selector(package_name, excluded_arches): |
| return lambda rpm: \ |
| rpm['arch'] not in excluded_arches and \ |
| (rpm['name'] == package_name or |
| rpm['name'].endswith('-debuginfo') or |
| rpm['name'].endswith('-devel')) |
| |
| if select_subpackages: |
| selector = lambda rpm: rpm['arch'] not in excluded_arches |
| else: |
| selector = rpms_selector(package_name, excluded_arches) |
| rpm_infos = self.listRPMs(buildID=build_id, |
| arches=arches, |
| selector=selector) |
| return RPMCollection((RPM(rpm_info) for rpm_info in rpm_infos)) |
| |
| @log_call |
| def get_latest_built_rpms(self, package_name, distro, arches=None): |
| """Get RPMs from latest build of a package |
| |
| :param str package_name: from which package to get the rpms |
| :param str distro: which distro the rpms belong to |
| :param arches: which arches the rpms belong to |
| :type arches: str or None |
| :return: the selected RPMs |
| :rtype: list |
| """ |
| latest_build = self.get_package_latest_build(package_name, distro) |
| # Get rpm and debuginfo rpm from each arch |
| return self.select_rpms_from_a_build(latest_build['build_id'], |
| package_name, |
| arches=arches) |
| |
| |
| @log_call |
| def get_session(): |
| """Get instance of Brew to talk with Koji""" |
| return Brew(global_config.koji_server) |
| |
| |
| @log_call |
| def get_download_dir(): |
| """Return the directory holding all downloaded RPMs |
| |
| If directory does not exist, it is created automatically. |
| |
| :return: path to directory holding downloaded RPMs. |
| :rtype: str |
| """ |
| download_dir = os.path.join(HOME_DIR, 'downloads') |
| if not os.path.exists(download_dir): |
| os.makedirs(download_dir) |
| return download_dir |
| |
| |
| @log_call |
| def download_rpm(url): |
| """Using curl to download a RPM from Koji |
| |
| Currently, curl is called and runs in a spawned process. pycurl would be a |
| good way instead. This would be changed in the future. |
| |
| :param str url: URL of a RPM to download. |
| :return: True if a RPM is downloaded successfully, False otherwise. |
| :rtype: bool |
| """ |
| cmd = 'curl --location --silent {0} -o {1}'.format( |
| url, os.path.join(get_download_dir(), |
| os.path.basename(url))) |
| if global_config.dry_run: |
| print('DRY-RUN: {0}'.format(cmd)) |
| return |
| |
| return_code = subprocess.call(cmd, shell=True) |
| if return_code > 0: |
| logger.error('curl fails with returned code: %d.', return_code) |
| return False |
| return True |
| |
| |
| @log_call |
| def download_rpms(rpms): |
| """Download RPMs |
| |
| :param list rpms: list of RPMs to download. |
| """ |
| def _download(rpm): |
| if rpm.is_downloaded: |
| logger.debug('Reuse %s', rpm.downloaded_file) |
| else: |
| logger.debug('Download %s', rpm.download_url) |
| download_rpm(rpm.download_url) |
| |
| for rpm in rpms: |
| _download(rpm) |
| |
| |
| @log_call |
| def build_path_to_abipkgdiff(): |
| """Build the path to the 'abipkgidiff' program to use. |
| |
| The path to 'abipkgdiff' is either the argument of the |
| --abipkgdiff command line option, or the path to 'abipkgdiff' as |
| found in the $PATH environment variable. |
| |
| :return: str a string representing the path to the 'abipkgdiff' |
| command. |
| """ |
| if global_config.abipkgdiff: |
| return global_config.abipkgdiff |
| return DEFAULT_ABIPKGDIFF |
| |
| |
| def format_debug_info_pkg_options(option, debuginfo_list): |
| """Given a list of debug info package descriptors return an option |
| string that looks like: |
| |
| option dbg.rpm1 option dbgrpm2 ... |
| |
| :param: list debuginfo_list a list of instances of the RPM class |
| representing the debug info rpms to use to construct the option |
| string. |
| |
| :return: str a string representing the option string that |
| concatenate the 'option' parameter before the path to each RPM |
| contained in 'debuginfo_list'. |
| """ |
| options = [] |
| |
| for dbg_pkg in debuginfo_list: |
| if dbg_pkg and dbg_pkg.downloaded_file: |
| options.append(' {0} {1}'.format(option, dbg_pkg.downloaded_file)) |
| |
| return ' '.join(options) if options else '' |
| |
| @log_call |
| def abipkgdiff(cmp_half1, cmp_half2): |
| """Run abipkgdiff against found two RPM packages |
| |
| Construct and execute abipkgdiff to get ABI diff |
| |
| abipkgdiff \ |
| --d1 package1-debuginfo --d2 package2-debuginfo \ |
| package1-rpm package2-rpm |
| |
| Output to stdout or stderr from abipkgdiff is not captured. abipkgdiff is |
| called synchronously. fedabipkgdiff does not return until underlying |
| abipkgdiff finishes. |
| |
| :param ComparisonHalf cmp_half1: the first comparison half. |
| :param ComparisonHalf cmp_half2: the second comparison half. |
| :return: return code of underlying abipkgdiff execution. |
| :rtype: int |
| """ |
| abipkgdiff_tool = build_path_to_abipkgdiff() |
| |
| suppressions = '' |
| |
| if global_config.suppr: |
| suppressions = '--suppressions {0}'.format(global_config.suppr) |
| |
| if global_config.no_devel_pkg: |
| devel_pkg1 = '' |
| devel_pkg2 = '' |
| else: |
| if cmp_half1.ancillary_devel is None: |
| msg = 'Development package for {0} does not exist.'.format(cmp_half1.subject.filename) |
| if global_config.error_on_warning: |
| raise RuntimeError(msg) |
| else: |
| devel_pkg1 = '' |
| logger.warning('{0} Ignored.'.format(msg)) |
| else: |
| devel_pkg1 = '--devel-pkg1 {0}'.format(cmp_half1.ancillary_devel.downloaded_file) |
| |
| if cmp_half2.ancillary_devel is None: |
| msg = 'Development package for {0} does not exist.'.format(cmp_half2.subject.filename) |
| if global_config.error_on_warning: |
| raise RuntimeError(msg) |
| else: |
| devel_pkg2 = '' |
| logger.warning('{0} Ignored.'.format(msg)) |
| else: |
| devel_pkg2 = '--devel-pkg2 {0}'.format(cmp_half2.ancillary_devel.downloaded_file) |
| |
| if cmp_half1.ancillary_debug is None: |
| msg = 'Debuginfo package for {0} does not exist.'.format(cmp_half1.subject.filename) |
| if global_config.error_on_warning: |
| raise RuntimeError(msg) |
| else: |
| debuginfo_pkg1 = '' |
| logger.warning('{0} Ignored.'.format(msg)) |
| else: |
| debuginfo_pkg1 = format_debug_info_pkg_options("--d1", cmp_half1.ancillary_debug) |
| |
| if cmp_half2.ancillary_debug is None: |
| msg = 'Debuginfo package for {0} does not exist.'.format(cmp_half2.subject.filename) |
| if global_config.error_on_warning: |
| raise RuntimeError(msg) |
| else: |
| debuginfo_pkg2 = '' |
| logger.warning('{0} Ignored.'.format(msg)) |
| else: |
| debuginfo_pkg2 = format_debug_info_pkg_options("--d2", cmp_half2.ancillary_debug); |
| |
| cmd = [] |
| |
| if global_config.self_compare: |
| cmd = [ |
| abipkgdiff_tool, |
| '--dso-only' if global_config.dso_only else '', |
| '--self-check', |
| debuginfo_pkg1, |
| cmp_half1.subject.downloaded_file, |
| ] |
| else: |
| cmd = [ |
| abipkgdiff_tool, |
| suppressions, |
| '--show-identical-binaries' if global_config.show_identical_binaries else '', |
| '--no-default-suppression' if global_config.no_default_suppr else '', |
| '--dso-only' if global_config.dso_only else '', |
| debuginfo_pkg1, |
| debuginfo_pkg2, |
| devel_pkg1, |
| devel_pkg2, |
| cmp_half1.subject.downloaded_file, |
| cmp_half2.subject.downloaded_file, |
| ] |
| cmd = [s for s in cmd if s != ''] |
| |
| if global_config.dry_run: |
| print('DRY-RUN: {0}'.format(' '.join(cmd))) |
| return |
| |
| logger.debug('Run: %s', ' '.join(cmd)) |
| |
| print('Comparing the ABI of binaries between {0} and {1}:'.format( |
| cmp_half1.subject.filename, cmp_half2.subject.filename)) |
| print() |
| |
| proc = subprocess.Popen(' '.join(cmd), shell=True, |
| stdout=subprocess.PIPE, stderr=subprocess.PIPE, |
| universal_newlines=True) |
| # So we could have done: stdout, stderr = proc.communicate() |
| # But then the documentatin of proc.communicate says: |
| # |
| # Note: The data read is buffered in memory, so do not use this |
| # method if the data size is large or unlimited. " |
| # |
| # In practice, we are seeing random cases where this |
| # proc.communicate() function does *NOT* terminate and seems to be |
| # in a deadlock state. So we are avoiding it altogether. We are |
| # then busy looping, waiting for the spawn process to finish, and |
| # then we get its output. |
| # |
| |
| while True: |
| if proc.poll() != None: |
| break |
| |
| stdout = ''.join(proc.stdout.readlines()) |
| stderr = ''.join(proc.stderr.readlines()) |
| |
| is_ok = proc.returncode == ABIDIFF_OK |
| is_internal_error = proc.returncode & ABIDIFF_ERROR or proc.returncode & ABIDIFF_USAGE_ERROR |
| has_abi_change = proc.returncode & ABIDIFF_ABI_CHANGE |
| |
| if is_internal_error: |
| six.print_(stderr, file=sys.stderr) |
| elif is_ok or has_abi_change: |
| print(stdout) |
| |
| return proc.returncode |
| |
| |
| @log_call |
| def run_abipkgdiff(rpm_col1, rpm_col2): |
| """Run abipkgdiff |
| |
| If one of the executions finds ABI differences, the return code is the |
| return code from abipkgdiff. |
| |
| :param RPMCollection rpm_col1: a collection of RPMs |
| :param RPMCollection rpm_col2: same as rpm_col1 |
| :return: exit code of the last non-zero returned from underlying abipkgdiff |
| :rtype: int |
| """ |
| return_codes = [ |
| abipkgdiff(cmp_half1, cmp_half2) for cmp_half1, cmp_half2 |
| in generate_comparison_halves(rpm_col1, rpm_col2)] |
| return max(return_codes, key=abs) if return_codes else 0 |
| |
| |
| @log_call |
| def diff_local_rpm_with_latest_rpm_from_koji(): |
| """Diff against local rpm and remove latest rpm |
| |
| This operation handles a local rpm and debuginfo rpm and remote ones |
| located in remote Koji server, that has specific distro specificed by |
| argument --from. |
| |
| 1/ Suppose the packager has just locally built a package named |
| foo-3.0.fc24.rpm. To compare the ABI of this locally build package with the |
| latest stable package from Fedora 23, one would do: |
| |
| fedabipkgdiff --from fc23 ./foo-3.0.fc24.rpm |
| """ |
| |
| from_distro = global_config.from_distro |
| if not is_distro_valid(from_distro): |
| raise InvalidDistroError('Invalid distro {0}'.format(from_distro)) |
| |
| local_rpm_file = global_config.NVR[0] |
| if not os.path.exists(local_rpm_file): |
| raise ValueError('{0} does not exist.'.format(local_rpm_file)) |
| |
| local_rpm = LocalRPM(local_rpm_file) |
| rpm_col1 = session.get_latest_built_rpms(local_rpm.name, |
| from_distro, |
| arches=local_rpm.arch) |
| rpm_col2 = RPMCollection.gather_from_dir(local_rpm_file) |
| |
| if global_config.clean_cache_before: |
| delete_download_cache() |
| |
| download_rpms(rpm_col1.rpms_iter()) |
| result = run_abipkgdiff(rpm_col1, rpm_col2) |
| |
| if global_config.clean_cache_after: |
| delete_download_cache() |
| |
| return result |
| |
| |
| @log_call |
| def diff_latest_rpms_based_on_distros(): |
| """abipkgdiff rpms based on two distros |
| |
| 2/ Suppose the packager wants to see how the ABIs of the package foo |
| evolved between fedora 19 and fedora 22. She would thus type the command: |
| |
| fedabipkgdiff --from fc19 --to fc22 foo |
| """ |
| |
| from_distro = global_config.from_distro |
| to_distro = global_config.to_distro |
| |
| if not is_distro_valid(from_distro): |
| raise InvalidDistroError('Invalid distro {0}'.format(from_distro)) |
| |
| if not is_distro_valid(to_distro): |
| raise InvalidDistroError('Invalid distro {0}'.format(to_distro)) |
| |
| package_name = global_config.NVR[0] |
| |
| rpm_col1 = session.get_latest_built_rpms(package_name, |
| distro=global_config.from_distro) |
| rpm_col2 = session.get_latest_built_rpms(package_name, |
| distro=global_config.to_distro) |
| |
| if global_config.clean_cache_before: |
| delete_download_cache() |
| |
| download_rpms(chain(rpm_col1.rpms_iter(), rpm_col2.rpms_iter())) |
| result = run_abipkgdiff(rpm_col1, rpm_col2) |
| |
| if global_config.clean_cache_after: |
| delete_download_cache() |
| |
| return result |
| |
| |
| @log_call |
| def diff_two_nvras_from_koji(): |
| """Diff two nvras from koji |
| |
| The arch probably omits, that means febabipkgdiff will diff all arches. If |
| specificed, the specific arch will be handled. |
| |
| 3/ Suppose the packager wants to compare the ABI of two packages designated |
| by their name and version. She would issue a command like this: |
| |
| fedabipkgdiff foo-1.0.fc19 foo-3.0.fc24 |
| fedabipkgdiff foo-1.0.fc19.i686 foo-1.0.fc24.i686 |
| """ |
| left_rpm = koji.parse_NVRA(global_config.NVR[0]) |
| right_rpm = koji.parse_NVRA(global_config.NVR[1]) |
| |
| if is_distro_valid(left_rpm['arch']) and \ |
| is_distro_valid(right_rpm['arch']): |
| nvr = koji.parse_NVR(global_config.NVR[0]) |
| params1 = (nvr['name'], nvr['version'], nvr['release'], None) |
| |
| nvr = koji.parse_NVR(global_config.NVR[1]) |
| params2 = (nvr['name'], nvr['version'], nvr['release'], None) |
| else: |
| params1 = (left_rpm['name'], |
| left_rpm['version'], |
| left_rpm['release'], |
| left_rpm['arch']) |
| params2 = (right_rpm['name'], |
| right_rpm['version'], |
| right_rpm['release'], |
| right_rpm['arch']) |
| |
| build_id = session.get_rpm_build_id(*params1) |
| rpm_col1 = session.select_rpms_from_a_build( |
| build_id, params1[0], arches=params1[3], |
| select_subpackages=global_config.check_all_subpackages) |
| |
| build_id = session.get_rpm_build_id(*params2) |
| rpm_col2 = session.select_rpms_from_a_build( |
| build_id, params2[0], arches=params2[3], |
| select_subpackages=global_config.check_all_subpackages) |
| |
| if global_config.clean_cache_before: |
| delete_download_cache() |
| |
| download_rpms(chain(rpm_col1.rpms_iter(), rpm_col2.rpms_iter())) |
| result = run_abipkgdiff(rpm_col1, rpm_col2) |
| |
| if global_config.clean_cache_after: |
| delete_download_cache() |
| |
| return result |
| |
| |
| @log_call |
| def self_compare_rpms_from_distro(): |
| """Compare ABI between same package from a distro |
| |
| Doing ABI comparison on self package should return no |
| ABI change and hence return code should be 0. This is useful |
| to ensure that functionality of libabigail itself |
| didn't break. This utility can be invoked like this: |
| |
| fedabipkgdiff --self-compare -a --from fc25 foo |
| """ |
| |
| from_distro = global_config.from_distro |
| |
| if not is_distro_valid(from_distro): |
| raise InvalidDistroError('Invalid distro {0}'.format(from_distro)) |
| |
| package_name = global_config.NVR[0] |
| |
| rpm_col1 = session.get_latest_built_rpms(package_name, |
| distro=global_config.from_distro) |
| |
| if global_config.clean_cache_before: |
| delete_download_cache() |
| |
| download_rpms(rpm_col1.rpms_iter()) |
| result = run_abipkgdiff(rpm_col1, rpm_col1) |
| |
| if global_config.clean_cache_after: |
| delete_download_cache() |
| |
| return result |
| |
| |
| @log_call |
| def diff_from_two_rpm_files(from_rpm_file, to_rpm_file): |
| """Diff two RPM files""" |
| rpm_col1 = RPMCollection.gather_from_dir(from_rpm_file) |
| rpm_col2 = RPMCollection.gather_from_dir(to_rpm_file) |
| if global_config.clean_cache_before: |
| delete_download_cache() |
| download_rpms(chain(rpm_col1.rpms_iter(), rpm_col2.rpms_iter())) |
| result = run_abipkgdiff(rpm_col1, rpm_col2) |
| if global_config.clean_cache_after: |
| delete_download_cache() |
| return result |
| |
| |
| def build_commandline_args_parser(): |
| parser = argparse.ArgumentParser( |
| description='Compare ABI of shared libraries in RPM packages from the ' |
| 'Koji build system') |
| |
| parser.add_argument( |
| 'NVR', |
| nargs='*', |
| help='RPM package N-V-R, N-V-R-A, N, or local RPM ' |
| 'file names with relative or absolute path.') |
| parser.add_argument( |
| '--dry-run', |
| required=False, |
| dest='dry_run', |
| action='store_true', |
| help='Don\'t actually do the work. The commands that should be ' |
| 'run will be sent to stdout.') |
| parser.add_argument( |
| '--from', |
| required=False, |
| metavar='DISTRO', |
| dest='from_distro', |
| help='baseline Fedora distribution name, for example, fc23') |
| parser.add_argument( |
| '--to', |
| required=False, |
| metavar='DISTRO', |
| dest='to_distro', |
| help='Fedora distribution name to compare against the baseline, for ' |
| 'example, fc24') |
| parser.add_argument( |
| '-a', |
| '--all-subpackages', |
| required=False, |
| action='store_true', |
| dest='check_all_subpackages', |
| help='Check all subpackages instead of only the package specificed in ' |
| 'command line.') |
| parser.add_argument( |
| '--dso-only', |
| required=False, |
| action='store_true', |
| dest='dso_only', |
| help='Compare the ABI of shared libraries only. If this option is not ' |
| 'provided, the tool compares the ABI of all ELF binaries.') |
| parser.add_argument( |
| '--debug', |
| required=False, |
| action='store_true', |
| dest='debug', |
| help='show debug output') |
| parser.add_argument( |
| '--traceback', |
| required=False, |
| action='store_true', |
| dest='show_traceback', |
| help='show traceback when there is an exception thrown.') |
| parser.add_argument( |
| '--server', |
| required=False, |
| metavar='URL', |
| dest='koji_server', |
| default=DEFAULT_KOJI_SERVER, |
| help='URL of koji XMLRPC service. Default is {0}'.format( |
| DEFAULT_KOJI_SERVER)) |
| parser.add_argument( |
| '--topurl', |
| required=False, |
| metavar='URL', |
| dest='koji_topurl', |
| default=DEFAULT_KOJI_TOPURL, |
| help='URL for RPM files access') |
| parser.add_argument( |
| '--abipkgdiff', |
| required=False, |
| metavar='ABIPKGDIFF', |
| dest='abipkgdiff', |
| default='', |
| help="The path to the 'abipkgtool' command to use. " |
| "By default use the one found in $PATH.") |
| parser.add_argument( |
| '--suppressions', |
| required=False, |
| metavar='SUPPR', |
| dest='suppr', |
| default='', |
| help='The suppression specification file to use during comparison') |
| parser.add_argument( |
| '--no-default-suppression', |
| required=False, |
| action='store_true', |
| dest='no_default_suppr', |
| help='Do not load default suppression specifications') |
| parser.add_argument( |
| '--no-devel-pkg', |
| required=False, |
| action='store_true', |
| dest='no_devel_pkg', |
| help='Do not compare ABI with development package') |
| parser.add_argument( |
| '--show-identical-binaries', |
| required=False, |
| action='store_true', |
| dest='show_identical_binaries', |
| help='Show information about binaries whose ABI are identical') |
| parser.add_argument( |
| '--error-on-warning', |
| required=False, |
| action='store_true', |
| dest='error_on_warning', |
| help='Raise error instead of warning') |
| parser.add_argument( |
| '--clean-cache', |
| required=False, |
| action=SetCleanCacheAction, |
| dest='clean_cache', |
| default=None, |
| help='A convenient way to clean cache without specifying ' |
| '--clean-cache-before and --clean-cache-after at same time') |
| parser.add_argument( |
| '--clean-cache-before', |
| required=False, |
| action='store_true', |
| dest='clean_cache_before', |
| default=None, |
| help='Clean cache before ABI comparison') |
| parser.add_argument( |
| '--clean-cache-after', |
| required=False, |
| action='store_true', |
| dest='clean_cache_after', |
| default=None, |
| help='Clean cache after ABI comparison') |
| parser.add_argument( |
| '--self-compare', |
| required=False, |
| action='store_true', |
| dest='self_compare', |
| default=None, |
| help='ABI comparison on same package') |
| return parser |
| |
| |
| def main(): |
| parser = build_commandline_args_parser() |
| |
| args = parser.parse_args() |
| |
| global global_config |
| global_config = args |
| |
| global pathinfo |
| pathinfo = koji.PathInfo(topdir=global_config.koji_topurl) |
| |
| global session |
| session = get_session() |
| |
| if global_config.debug: |
| logger.setLevel(logging.DEBUG) |
| |
| logger.debug(args) |
| |
| if global_config.from_distro and global_config.self_compare and \ |
| global_config.NVR: |
| return self_compare_rpms_from_distro() |
| |
| if global_config.from_distro and global_config.to_distro is None and \ |
| global_config.NVR: |
| return diff_local_rpm_with_latest_rpm_from_koji() |
| |
| if global_config.from_distro and global_config.to_distro and \ |
| global_config.NVR: |
| return diff_latest_rpms_based_on_distros() |
| |
| if global_config.from_distro is None and global_config.to_distro is None: |
| if len(global_config.NVR) > 1: |
| left_one = global_config.NVR[0] |
| right_one = global_config.NVR[1] |
| |
| if is_rpm_file(left_one) and is_rpm_file(right_one): |
| return diff_from_two_rpm_files(left_one, right_one) |
| |
| both_nvr = match_nvr(left_one) and match_nvr(right_one) |
| both_nvra = match_nvra(left_one) and match_nvra(right_one) |
| |
| if both_nvr or both_nvra: |
| return diff_two_nvras_from_koji() |
| |
| six.print_('Unknown arguments. Please refer to --help.', file=sys.stderr) |
| return 1 |
| |
| |
| if __name__ == '__main__': |
| try: |
| sys.exit(main()) |
| except KeyboardInterrupt: |
| if global_config is None: |
| raise |
| if global_config.debug: |
| logger.debug('Terminate by user') |
| else: |
| six.print_('Terminate by user', file=sys.stderr) |
| if global_config.show_traceback: |
| raise |
| else: |
| sys.exit(2) |
| except Exception as e: |
| if global_config is None: |
| raise |
| if global_config.debug: |
| logger.debug(str(e)) |
| else: |
| six.print_(str(e), file=sys.stderr) |
| if global_config.show_traceback: |
| raise |
| else: |
| sys.exit(1) |
