#!/usr/bin/python """ Software package management library. This is an abstraction layer on top of the existing distributions high level package managers. It supports package operations useful for testing purposes, and multiple high level package managers (here called backends). If you want to make this lib to support your particular package manager/distro, please implement the given backend class. @author: Higor Vieira Alves (halves@br.ibm.com) @author: Lucas Meneghel Rodrigues (lmr@redhat.com) @author: Ramon de Carvalho Valle (rcvalle@br.ibm.com) @copyright: IBM 2008-2009 @copyright: Red Hat 2009-2010 """ import os, re, logging, ConfigParser, optparse, random, string try: import yum except: pass import common from autotest_lib.client.bin import os_dep, utils from autotest_lib.client.common_lib import error from autotest_lib.client.common_lib import logging_config, logging_manager def generate_random_string(length): """ Return a random string using alphanumeric characters. @length: Length of the string that will be generated. """ r = random.SystemRandom() str = "" chars = string.letters + string.digits while length > 0: str += r.choice(chars) length -= 1 return str class SoftwareManagerLoggingConfig(logging_config.LoggingConfig): """ Used with the sole purpose of providing convenient logging setup for the KVM test auxiliary programs. """ def configure_logging(self, results_dir=None, verbose=False): super(SoftwareManagerLoggingConfig, self).configure_logging( use_console=True, verbose=verbose) class SystemInspector(object): """ System inspector class. This may grow up to include more complete reports of operating system and machine properties. """ def __init__(self): """ Probe system, and save information for future reference. """ self.distro = utils.get_os_vendor() self.high_level_pms = ['apt-get', 'yum', 'zypper'] def get_package_management(self): """ Determine the supported package management systems present on the system. If more than one package management system installed, try to find the best supported system. """ list_supported = [] for high_level_pm in self.high_level_pms: try: os_dep.command(high_level_pm) list_supported.append(high_level_pm) except: pass pm_supported = None if len(list_supported) == 0: pm_supported = None if len(list_supported) == 1: pm_supported = list_supported[0] elif len(list_supported) > 1: if 'apt-get' in list_supported and self.distro in ['Debian', 'Ubuntu']: pm_supported = 'apt-get' elif 'yum' in list_supported and self.distro == 'Fedora': pm_supported = 'yum' else: pm_supported = list_supported[0] logging.debug('Package Manager backend: %s' % pm_supported) return pm_supported class SoftwareManager(object): """ Package management abstraction layer. It supports a set of common package operations for testing purposes, and it uses the concept of a backend, a helper class that implements the set of operations of a given package management tool. """ def __init__(self): """ Class constructor. Determines the best supported package management system for the given operating system running and initializes the appropriate backend. """ inspector = SystemInspector() backend_type = inspector.get_package_management() if backend_type == 'yum': self.backend = YumBackend() elif backend_type == 'zypper': self.backend = ZypperBackend() elif backend_type == 'apt-get': self.backend = AptBackend() else: raise NotImplementedError('Unimplemented package management ' 'system: %s.' % backend_type) def check_installed(self, name, version=None, arch=None): """ Check whether a package is installed on this system. @param name: Package name. @param version: Package version. @param arch: Package architecture. """ return self.backend.check_installed(name, version, arch) def list_all(self): """ List all installed packages. """ return self.backend.list_all() def list_files(self, name): """ Get a list of all files installed by package [name]. @param name: Package name. """ return self.backend.list_files(name) def install(self, name): """ Install package [name]. @param name: Package name. """ return self.backend.install(name) def remove(self, name): """ Remove package [name]. @param name: Package name. """ return self.backend.remove(name) def add_repo(self, url): """ Add package repo described by [url]. @param name: URL of the package repo. """ return self.backend.add_repo(url) def remove_repo(self, url): """ Remove package repo described by [url]. @param url: URL of the package repo. """ return self.backend.remove_repo(url) def upgrade(self): """ Upgrade all packages available. """ return self.backend.upgrade() def provides(self, file): """ Returns a list of packages that provides a given capability to the system (be it a binary, a library). @param file: Path to the file. """ return self.backend.provides(file) def install_what_provides(self, file): """ Installs package that provides [file]. @param file: Path to file. """ provides = self.provides(file) if provides is not None: self.install(provides) else: logging.warning('No package seems to provide %s', file) class RpmBackend(object): """ This class implements operations executed with the rpm package manager. rpm is a lower level package manager, used by higher level managers such as yum and zypper. """ def __init__(self): self.lowlevel_base_cmd = os_dep.command('rpm') def _check_installed_version(self, name, version): """ Helper for the check_installed public method. @param name: Package name. @param version: Package version. """ cmd = (self.lowlevel_base_cmd + ' -q --qf %{VERSION} ' + name + ' 2> /dev/null') inst_version = utils.system_output(cmd) if inst_version >= version: return True else: return False def check_installed(self, name, version=None, arch=None): """ Check if package [name] is installed. @param name: Package name. @param version: Package version. @param arch: Package architecture. """ if arch: cmd = (self.lowlevel_base_cmd + ' -q --qf %{ARCH} ' + name + ' 2> /dev/null') inst_archs = utils.system_output(cmd) inst_archs = inst_archs.split('\n') for inst_arch in inst_archs: if inst_arch == arch: return self._check_installed_version(name, version) return False elif version: return self._check_installed_version(name, version) else: cmd = 'rpm -q ' + name + ' 2> /dev/null' return (os.system(cmd) == 0) def list_all(self): """ List all installed packages. """ installed_packages = utils.system_output('rpm -qa').splitlines() return installed_packages def list_files(self, name): """ List files installed on the system by package [name]. @param name: Package name. """ path = os.path.abspath(name) if os.path.isfile(path): option = '-qlp' name = path else: option = '-ql' l_cmd = 'rpm' + ' ' + option + ' ' + name + ' 2> /dev/null' try: result = utils.system_output(l_cmd) list_files = result.split('\n') return list_files except error.CmdError: return [] class DpkgBackend(object): """ This class implements operations executed with the dpkg package manager. dpkg is a lower level package manager, used by higher level managers such as apt and aptitude. """ def __init__(self): self.lowlevel_base_cmd = os_dep.command('dpkg') def check_installed(self, name): if os.path.isfile(name): n_cmd = (self.lowlevel_base_cmd + ' -f ' + name + ' Package 2>/dev/null') name = utils.system_output(n_cmd) i_cmd = self.lowlevel_base_cmd + ' -s ' + name + ' 2>/dev/null' # Checking if package is installed package_status = utils.system_output(i_cmd, ignore_status=True) not_inst_pattern = re.compile('not-installed', re.IGNORECASE) dpkg_not_installed = re.search(not_inst_pattern, package_status) if dpkg_not_installed: return False return True def list_all(self): """ List all packages available in the system. """ installed_packages = [] raw_list = utils.system_output('dpkg -l').splitlines()[5:] for line in raw_list: parts = line.split() if parts[0] == "ii": # only grab "installed" packages installed_packages.append("%s-%s" % (parts[1], parts[2])) def list_files(self, package): """ List files installed by package [package]. @param package: Package name. @return: List of paths installed by package. """ if os.path.isfile(package): l_cmd = self.lowlevel_base_cmd + ' -c ' + package else: l_cmd = self.lowlevel_base_cmd + ' -l ' + package return utils.system_output(l_cmd).split('\n') class YumBackend(RpmBackend): """ Implements the yum backend for software manager. Set of operations for the yum package manager, commonly found on Yellow Dog Linux and Red Hat based distributions, such as Fedora and Red Hat Enterprise Linux. """ def __init__(self): """ Initializes the base command and the yum package repository. """ super(YumBackend, self).__init__() executable = os_dep.command('yum') base_arguments = '-y' self.base_command = executable + ' ' + base_arguments self.repo_file_path = '/etc/yum.repos.d/autotest.repo' self.cfgparser = ConfigParser.ConfigParser() self.cfgparser.read(self.repo_file_path) y_cmd = executable + ' --version | head -1' self.yum_version = utils.system_output(y_cmd, ignore_status=True) logging.debug('Yum backend initialized') logging.debug('Yum version: %s' % self.yum_version) self.yum_base = yum.YumBase() def _cleanup(self): """ Clean up the yum cache so new package information can be downloaded. """ utils.system("yum clean all") def install(self, name): """ Installs package [name]. Handles local installs. """ if os.path.isfile(name): name = os.path.abspath(name) command = 'localinstall' else: command = 'install' i_cmd = self.base_command + ' ' + command + ' ' + name try: utils.system(i_cmd) return True except: return False def remove(self, name): """ Removes package [name]. @param name: Package name (eg. 'ipython'). """ r_cmd = self.base_command + ' ' + 'erase' + ' ' + name try: utils.system(r_cmd) return True except: return False def add_repo(self, url): """ Adds package repository located on [url]. @param url: Universal Resource Locator of the repository. """ # Check if we URL is already set for section in self.cfgparser.sections(): for option, value in self.cfgparser.items(section): if option == 'url' and value == url: return True # Didn't find it, let's set it up while True: section_name = 'software_manager' + '_' + generate_random_string(4) if not self.cfgparser.has_section(section_name): break self.cfgparser.add_section(section_name) self.cfgparser.set(section_name, 'name', 'Repository added by the autotest software manager.') self.cfgparser.set(section_name, 'url', url) self.cfgparser.set(section_name, 'enabled', 1) self.cfgparser.set(section_name, 'gpgcheck', 0) self.cfgparser.write(self.repo_file_path) def remove_repo(self, url): """ Removes package repository located on [url]. @param url: Universal Resource Locator of the repository. """ for section in self.cfgparser.sections(): for option, value in self.cfgparser.items(section): if option == 'url' and value == url: self.cfgparser.remove_section(section) self.cfgparser.write(self.repo_file_path) def upgrade(self): """ Upgrade all available packages. """ r_cmd = self.base_command + ' ' + 'update' try: utils.system(r_cmd) return True except: return False def provides(self, name): """ Returns a list of packages that provides a given capability. @param name: Capability name (eg, 'foo'). """ d_provides = self.yum_base.searchPackageProvides(args=[name]) provides_list = [key for key in d_provides] if provides_list: logging.info("Package %s provides %s", provides_list[0], name) return str(provides_list[0]) else: return None class ZypperBackend(RpmBackend): """ Implements the zypper backend for software manager. Set of operations for the zypper package manager, found on SUSE Linux. """ def __init__(self): """ Initializes the base command and the yum package repository. """ super(ZypperBackend, self).__init__() self.base_command = os_dep.command('zypper') + ' -n' z_cmd = self.base_command + ' --version' self.zypper_version = utils.system_output(z_cmd, ignore_status=True) logging.debug('Zypper backend initialized') logging.debug('Zypper version: %s' % self.zypper_version) def install(self, name): """ Installs package [name]. Handles local installs. @param name: Package Name. """ path = os.path.abspath(name) i_cmd = self.base_command + ' install -l ' + name try: utils.system(i_cmd) return True except: return False def add_repo(self, url): """ Adds repository [url]. @param url: URL for the package repository. """ ar_cmd = self.base_command + ' addrepo ' + url try: utils.system(ar_cmd) return True except: return False def remove_repo(self, url): """ Removes repository [url]. @param url: URL for the package repository. """ rr_cmd = self.base_command + ' removerepo ' + url try: utils.system(rr_cmd) return True except: return False def remove(self, name): """ Removes package [name]. """ r_cmd = self.base_command + ' ' + 'erase' + ' ' + name try: utils.system(r_cmd) return True except: return False def upgrade(self): """ Upgrades all packages of the system. """ u_cmd = self.base_command + ' update -l' try: utils.system(u_cmd) return True except: return False def provides(self, name): """ Searches for what provides a given file. @param name: File path. """ p_cmd = self.base_command + ' what-provides ' + name list_provides = [] try: p_output = utils.system_output(p_cmd).split('\n')[4:] for line in p_output: line = [a.strip() for a in line.split('|')] try: state, pname, type, version, arch, repository = line if pname not in list_provides: list_provides.append(pname) except IndexError: pass if len(list_provides) > 1: logging.warning('More than one package found, ' 'opting by the first queue result') if list_provides: logging.info("Package %s provides %s", list_provides[0], name) return list_provides[0] return None except: return None class AptBackend(DpkgBackend): """ Implements the apt backend for software manager. Set of operations for the apt package manager, commonly found on Debian and Debian based distributions, such as Ubuntu Linux. """ def __init__(self): """ Initializes the base command and the debian package repository. """ super(AptBackend, self).__init__() executable = os_dep.command('apt-get') self.base_command = executable + ' -y' self.repo_file_path = '/etc/apt/sources.list.d/autotest' self.apt_version = utils.system_output('apt-get -v | head -1', ignore_status=True) logging.debug('Apt backend initialized') logging.debug('apt version: %s' % self.apt_version) def install(self, name): """ Installs package [name]. @param name: Package name. """ command = 'install' i_cmd = self.base_command + ' ' + command + ' ' + name try: utils.system(i_cmd) return True except: return False def remove(self, name): """ Remove package [name]. @param name: Package name. """ command = 'remove' flag = '--purge' r_cmd = self.base_command + ' ' + command + ' ' + flag + ' ' + name try: utils.system(r_cmd) return True except: return False def add_repo(self, repo): """ Add an apt repository. @param repo: Repository string. Example: 'deb http://archive.ubuntu.com/ubuntu/ maverick universe' """ repo_file = open(self.repo_file_path, 'a') repo_file_contents = repo_file.read() if repo not in repo_file_contents: repo_file.write(repo) def remove_repo(self, repo): """ Remove an apt repository. @param repo: Repository string. Example: 'deb http://archive.ubuntu.com/ubuntu/ maverick universe' """ repo_file = open(self.repo_file_path, 'r') new_file_contents = [] for line in repo_file.readlines: if not line == repo: new_file_contents.append(line) repo_file.close() new_file_contents = "\n".join(new_file_contents) repo_file.open(self.repo_file_path, 'w') repo_file.write(new_file_contents) repo_file.close() def upgrade(self): """ Upgrade all packages of the system with eventual new versions. """ ud_command = 'update' ud_cmd = self.base_command + ' ' + ud_command try: utils.system(ud_cmd) except: logging.error("Apt package update failed") up_command = 'upgrade' up_cmd = self.base_command + ' ' + up_command try: utils.system(up_cmd) return True except: return False def provides(self, file): """ Return a list of packages that provide [file]. @param file: File path. """ if not self.check_installed('apt-file'): self.install('apt-file') command = os_dep.command('apt-file') cache_update_cmd = command + ' update' try: utils.system(cache_update_cmd, ignore_status=True) except: logging.error("Apt file cache update failed") fu_cmd = command + ' search ' + file try: provides = utils.system_output(fu_cmd).split('\n') list_provides = [] for line in provides: if line: try: line = line.split(':') package = line[0].strip() path = line[1].strip() if path == file and package not in list_provides: list_provides.append(package) except IndexError: pass if len(list_provides) > 1: logging.warning('More than one package found, ' 'opting by the first queue result') if list_provides: logging.info("Package %s provides %s", list_provides[0], file) return list_provides[0] return None except: return None if __name__ == '__main__': parser = optparse.OptionParser( "usage: %prog [install|remove|list-all|list-files|add-repo|remove-repo|" "upgrade|what-provides|install-what-provides] arguments") parser.add_option('--verbose', dest="debug", action='store_true', help='include debug messages in console output') options, args = parser.parse_args() debug = options.debug logging_manager.configure_logging(SoftwareManagerLoggingConfig(), verbose=debug) software_manager = SoftwareManager() if args: action = args[0] args = " ".join(args[1:]) else: action = 'show-help' if action == 'install': software_manager.install(args) elif action == 'remove': software_manager.remove(args) if action == 'list-all': software_manager.list_all() elif action == 'list-files': software_manager.list_files(args) elif action == 'add-repo': software_manager.add_repo(args) elif action == 'remove-repo': software_manager.remove_repo(args) elif action == 'upgrade': software_manager.upgrade() elif action == 'what-provides': software_manager.provides(args) elif action == 'install-what-provides': software_manager.install_what_provides(args) elif action == 'show-help': parser.print_help()