Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-21 15:47
📄 __init__.py # Copyright 2006 Duke University # Copyright (C) 2012-2016 Red Hat, Inc. # # 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 Library General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. # # Written by Seth Vidal """ Classes for subcommands of the yum command line interface. """ from __future__ import print_function from __future__ import unicode_literals from dnf.cli.option_parser import OptionParser from dnf.i18n import _ import dnf.cli import dnf.exceptions import dnf.pycomp import dnf.util import logging import os logger = logging.getLogger('dnf') _RPM_VERIFY = _("To diagnose the problem, try running: '%s'.") % \ 'rpm -Va --nofiles --nodigest' _RPM_REBUILDDB = _("You probably have corrupted RPMDB, running '%s'" " might fix the issue.") % 'rpm --rebuilddb' gpg_msg = \ _("""You have enabled checking of packages via GPG keys. This is a good thing. However, you do not have any GPG public keys installed. You need to download the keys for packages you wish to install and install them. You can do that by running the command: rpm --import public.gpg.key Alternatively you can specify the url to the key you would like to use for a repository in the 'gpgkey' option in a repository section and {prog} will install it for you. For more information contact your distribution or package provider.""") def _checkGPGKey(base, cli): """Verify that there are gpg keys for the enabled repositories in the rpm database. :param base: a :class:`dnf.Base` object. :raises: :class:`cli.CliError` """ if not base.conf.gpgcheck: return if not base._gpg_key_check(): for repo in base.repos.iter_enabled(): if (repo.gpgcheck or repo.repo_gpgcheck) and not repo.gpgkey: logger.critical("\n%s\n", gpg_msg.format(prog=dnf.util.MAIN_PROG_UPPER)) logger.critical(_("Problem repository: %s"), repo) raise dnf.cli.CliError def _checkEnabledRepo(base, possible_local_files=()): """Verify that there is at least one enabled repo. :param base: a :class:`dnf.Base` object. :param possible_local_files: the list of strings that could be a local rpms :raises: :class:`cli.CliError`: """ if base.repos._any_enabled(): return for lfile in possible_local_files: if lfile.endswith(".rpm") and os.path.exists(lfile): return scheme = dnf.pycomp.urlparse.urlparse(lfile)[0] if scheme in ('http', 'ftp', 'file', 'https'): return msg = _('There are no enabled repositories in "{}".').format('", "'.join(base.conf.reposdir)) raise dnf.cli.CliError(msg) class Command(object): """Abstract base class for CLI commands.""" aliases = [] # :api summary = "" # :api opts = None def __init__(self, cli): # :api self.cli = cli @property def base(self): # :api return self.cli.base @property def _basecmd(self): return self.aliases[0] @property def output(self): return self.cli.base.output def set_argparser(self, parser): """Define command specific options and arguments. #:api""" pass def pre_configure(self): # :api """Do any command-specific pre-configuration.""" pass def configure(self): # :api """Do any command-specific configuration.""" pass def get_error_output(self, error): """Get suggestions for resolving the given error.""" if isinstance(error, dnf.exceptions.TransactionCheckError): return (_RPM_VERIFY, _RPM_REBUILDDB) raise NotImplementedError('error not supported yet: %s' % error) def run(self): # :api """Execute the command.""" pass def run_resolved(self): """Finalize operation after resolvement""" pass def run_transaction(self): """Finalize operations post-transaction.""" pass class InfoCommand(Command): """A class containing methods needed by the cli to execute the info command. """ aliases = ('info', 'if') summary = _('display details about a package or group of packages') DEFAULT_PKGNARROW = 'all' pkgnarrows = {'available', 'installed', 'extras', 'updates', 'upgrades', 'autoremove', 'recent', 'obsoletes', DEFAULT_PKGNARROW} @classmethod def set_argparser(cls, parser): narrows = parser.add_mutually_exclusive_group() narrows.add_argument('--all', dest='_packages_action', action='store_const', const='all', default=None, help=_("show all packages (default)")) narrows.add_argument('--available', dest='_packages_action', action='store_const', const='available', help=_("show only available packages")) narrows.add_argument('--installed', dest='_packages_action', action='store_const', const='installed', help=_("show only installed packages")) narrows.add_argument('--extras', dest='_packages_action', action='store_const', const='extras', help=_("show only extras packages")) narrows.add_argument('--updates', dest='_packages_action', action='store_const', const='upgrades', help=_("show only upgrades packages")) narrows.add_argument('--upgrades', dest='_packages_action', action='store_const', const='upgrades', help=_("show only upgrades packages")) narrows.add_argument('--autoremove', dest='_packages_action', action='store_const', const='autoremove', help=_("show only autoremove packages")) narrows.add_argument('--recent', dest='_packages_action', action='store_const', const='recent', help=_("show only recently changed packages")) parser.add_argument('packages', nargs='', metavar=_('PACKAGE'), choices=cls.pkgnarrows, default=cls.DEFAULT_PKGNARROW, action=OptionParser.PkgNarrowCallback, help=_("Package name specification")) def configure(self): demands = self.cli.demands demands.sack_activation = True if self.opts._packages_action: self.opts.packages_action = self.opts._packages_action if self.opts.packages_action != 'installed': demands.available_repos = True if self.opts.obsoletes: if self.opts._packages_action: self.cli._option_conflict("--obsoletes", "--" + self.opts._packages_action) else: self.opts.packages_action = 'obsoletes' if self.opts.packages_action == 'updates': self.opts.packages_action = 'upgrades' def run(self): self.cli._populate_update_security_filter(self.opts) return self.base.output_packages('info', self.opts.packages_action, self.opts.packages) class ListCommand(InfoCommand): """A class containing methods needed by the cli to execute the list command. """ aliases = ('list', 'ls') summary = _('list a package or groups of packages') def run(self): self.cli._populate_update_security_filter(self.opts) return self.base.output_packages('list', self.opts.packages_action, self.opts.packages) class ProvidesCommand(Command): """A class containing methods needed by the cli to execute the provides command. """ aliases = ('provides', 'whatprovides', 'prov', 'wp') summary = _('find what package provides the given value') @staticmethod def set_argparser(parser): parser.add_argument('dependency', nargs='+', metavar=_('PROVIDE'), help=_("Provide specification to search for")) def configure(self): demands = self.cli.demands demands.available_repos = True demands.fresh_metadata = False demands.sack_activation = True def run(self): logger.debug(_("Searching Packages: ")) return self.base.provides(self.opts.dependency) class CheckUpdateCommand(Command): """A class containing methods needed by the cli to execute the check-update command. """ aliases = ('check-update', 'check-upgrade') summary = _('check for available package upgrades') @staticmethod def set_argparser(parser): parser.add_argument('--changelogs', dest='changelogs', default=False, action='store_true', help=_('show changelogs before update')) parser.add_argument('packages', nargs='', metavar=_('PACKAGE')) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.plugin_filtering_enabled = True if self.opts.changelogs: demands.changelogs = True _checkEnabledRepo(self.base) def run(self): self.cli._populate_update_security_filter(self.opts, cmp_type="gte") found = self.base.check_updates(self.opts.packages, print_=True, changelogs=self.opts.changelogs) if found: self.cli.demands.success_exit_status = 100 if self.base.conf.autocheck_running_kernel: self.cli._check_running_kernel() class RepoPkgsCommand(Command): """Implementation of the repository-packages command.""" class CheckUpdateSubCommand(Command): """Implementation of the info sub-command.""" aliases = ('check-update',) def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" found = self.base.check_updates(self.opts.pkg_specs, self.reponame, print_=True) if found: self.cli.demands.success_exit_status = 100 class InfoSubCommand(Command): """Implementation of the info sub-command.""" aliases = ('info',) def configure(self): demands = self.cli.demands demands.sack_activation = True if self.opts._pkg_specs_action: self.opts.pkg_specs_action = self.opts._pkg_specs_action if self.opts.pkg_specs_action != 'installed': demands.available_repos = True if self.opts.obsoletes: if self.opts._pkg_specs_action: self.cli._option_conflict("--obsoletes", "--" + self.opts._pkg_specs_action) else: self.opts.pkg_specs_action = 'obsoletes' def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" self.cli._populate_update_security_filter(self.opts) self.base.output_packages('info', self.opts.pkg_specs_action, self.opts.pkg_specs, self.reponame) class InstallSubCommand(Command): """Implementation of the install sub-command.""" aliases = ('install',) def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True demands.resolving = True demands.root_user = True def run_on_repo(self): self.cli._populate_update_security_filter(self.opts) """Execute the command with respect to given arguments cli_args.""" _checkGPGKey(self.base, self.cli) done = False if not self.opts.pkg_specs: # Install all packages. try: self.base.install('', self.reponame) except dnf.exceptions.MarkingError: logger.info(_('No package available.')) else: done = True else: # Install packages. for pkg_spec in self.opts.pkg_specs: try: self.base.install(pkg_spec, self.reponame) except dnf.exceptions.MarkingError as e: msg = '{}: {}'.format(e.value, self.base.output.term.bold(pkg_spec)) logger.info(msg) else: done = True if not done: raise dnf.exceptions.Error(_('No packages marked for install.')) class ListSubCommand(InfoSubCommand): """Implementation of the list sub-command.""" aliases = ('list',) def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" self.cli._populate_update_security_filter(self.opts) self.base.output_packages('list', self.opts.pkg_specs_action, self.opts.pkg_specs, self.reponame) class MoveToSubCommand(Command): """Implementation of the move-to sub-command.""" aliases = ('move-to',) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" _checkGPGKey(self.base, self.cli) done = False if not self.opts.pkg_specs: # Reinstall all packages. try: self.base.reinstall('', new_reponame=self.reponame) except dnf.exceptions.PackagesNotInstalledError: logger.info(_('No package installed.')) except dnf.exceptions.PackagesNotAvailableError: logger.info(_('No package available.')) except dnf.exceptions.MarkingError: assert False, 'Only the above marking errors are expected.' else: done = True else: # Reinstall packages. for pkg_spec in self.opts.pkg_specs: try: self.base.reinstall(pkg_spec, new_reponame=self.reponame) except dnf.exceptions.PackagesNotInstalledError: msg = _('No match for argument: %s') logger.info(msg, pkg_spec) except dnf.exceptions.PackagesNotAvailableError as err: for pkg in err.packages: xmsg = '' pkgrepo = self.base.history.repo(pkg) if pkgrepo: xmsg = _(' (from %s)') % pkgrepo msg = _('Installed package %s%s not available.') logger.info(msg, self.output.term.bold(pkg), xmsg) except dnf.exceptions.MarkingError: assert False, \ 'Only the above marking errors are expected.' else: done = True if not done: raise dnf.exceptions.Error(_('Nothing to do.')) class ReinstallOldSubCommand(Command): """Implementation of the reinstall-old sub-command.""" aliases = ('reinstall-old',) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" _checkGPGKey(self.base, self.cli) done = False if not self.opts.pkg_specs: # Reinstall all packages. try: self.base.reinstall('', self.reponame, self.reponame) except dnf.exceptions.PackagesNotInstalledError: msg = _('No package installed from the repository.') logger.info(msg) except dnf.exceptions.PackagesNotAvailableError: logger.info(_('No package available.')) except dnf.exceptions.MarkingError: assert False, 'Only the above marking errors are expected.' else: done = True else: # Reinstall packages. for pkg_spec in self.opts.pkg_specs: try: self.base.reinstall(pkg_spec, self.reponame, self.reponame) except dnf.exceptions.PackagesNotInstalledError: msg = _('No match for argument: %s') logger.info(msg, pkg_spec) except dnf.exceptions.PackagesNotAvailableError as err: for pkg in err.packages: xmsg = '' pkgrepo = self.base.history.repo(pkg) if pkgrepo: xmsg = _(' (from %s)') % pkgrepo msg = _('Installed package %s%s not available.') logger.info(msg, self.output.term.bold(pkg), xmsg) except dnf.exceptions.MarkingError: assert False, \ 'Only the above marking errors are expected.' else: done = True if not done: raise dnf.exceptions.Error(_('Nothing to do.')) class ReinstallSubCommand(Command): """Implementation of the reinstall sub-command.""" aliases = ('reinstall',) def __init__(self, cli): """Initialize the command.""" super(RepoPkgsCommand.ReinstallSubCommand, self).__init__(cli) self.wrapped_commands = (RepoPkgsCommand.ReinstallOldSubCommand(cli), RepoPkgsCommand.MoveToSubCommand(cli)) def configure(self): self.cli.demands.available_repos = True for command in self.wrapped_commands: command.opts = self.opts command.reponame = self.reponame command.configure() def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" _checkGPGKey(self.base, self.cli) for command in self.wrapped_commands: try: command.run_on_repo() except dnf.exceptions.Error: continue else: break else: raise dnf.exceptions.Error(_('No packages marked for reinstall.')) class RemoveOrDistroSyncSubCommand(Command): """Implementation of the remove-or-distro-sync sub-command.""" aliases = ('remove-or-distro-sync',) def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True demands.resolving = True demands.root_user = True def _replace(self, pkg_spec, reponame): """Synchronize a package with another repository or remove it.""" self.cli.base.sack.disable_repo(reponame) subject = dnf.subject.Subject(pkg_spec) matches = subject.get_best_query(self.cli.base.sack) history = self.cli.base.history installed = [ pkg for pkg in matches.installed() if history.repo(pkg) == reponame] if not installed: raise dnf.exceptions.PackagesNotInstalledError( 'no package matched', pkg_spec) available = matches.available() clean_deps = self.cli.base.conf.clean_requirements_on_remove for package in installed: if available.filter(name=package.name, arch=package.arch): self.cli.base._goal.distupgrade(package) else: self.cli.base._goal.erase(package, clean_deps=clean_deps) def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" _checkGPGKey(self.base, self.cli) done = False if not self.opts.pkg_specs: # Sync all packages. try: self._replace('', self.reponame) except dnf.exceptions.PackagesNotInstalledError: msg = _('No package installed from the repository.') logger.info(msg) else: done = True else: # Reinstall packages. for pkg_spec in self.opts.pkg_specs: try: self._replace(pkg_spec, self.reponame) except dnf.exceptions.PackagesNotInstalledError: msg = _('No match for argument: %s') logger.info(msg, pkg_spec) else: done = True if not done: raise dnf.exceptions.Error(_('Nothing to do.')) class RemoveOrReinstallSubCommand(Command): """Implementation of the remove-or-reinstall sub-command.""" aliases = ('remove-or-reinstall',) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" _checkGPGKey(self.base, self.cli) done = False if not self.opts.pkg_specs: # Reinstall all packages. try: self.base.reinstall('', old_reponame=self.reponame, new_reponame_neq=self.reponame, remove_na=True) except dnf.exceptions.PackagesNotInstalledError: msg = _('No package installed from the repository.') logger.info(msg) except dnf.exceptions.MarkingError: assert False, 'Only the above marking error is expected.' else: done = True else: # Reinstall packages. for pkg_spec in self.opts.pkg_specs: try: self.base.reinstall( pkg_spec, old_reponame=self.reponame, new_reponame_neq=self.reponame, remove_na=True) except dnf.exceptions.PackagesNotInstalledError: msg = _('No match for argument: %s') logger.info(msg, pkg_spec) except dnf.exceptions.MarkingError: assert False, 'Only the above marking error is expected.' else: done = True if not done: raise dnf.exceptions.Error(_('Nothing to do.')) class RemoveSubCommand(Command): """Implementation of the remove sub-command.""" aliases = ('remove',) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.allow_erasing = True demands.available_repos = False demands.resolving = True demands.root_user = True def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" done = False if not self.opts.pkg_specs: # Remove all packages. try: self.base.remove('', self.reponame) except dnf.exceptions.MarkingError: msg = _('No package installed from the repository.') logger.info(msg) else: done = True else: # Remove packages. for pkg_spec in self.opts.pkg_specs: try: self.base.remove(pkg_spec, self.reponame) except dnf.exceptions.MarkingError as e: logger.info(str(e)) else: done = True if not done: logger.warning(_('No packages marked for removal.')) class UpgradeSubCommand(Command): """Implementation of the upgrade sub-command.""" aliases = ('upgrade', 'upgrade-to') def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True def run_on_repo(self): """Execute the command with respect to given arguments cli_args.""" _checkGPGKey(self.base, self.cli) done = False if not self.opts.pkg_specs: # Update all packages. self.base.upgrade_all(self.reponame) done = True else: # Update packages. for pkg_spec in self.opts.pkg_specs: try: self.base.upgrade(pkg_spec, self.reponame) except dnf.exceptions.MarkingError: logger.info(_('No match for argument: %s'), pkg_spec) else: done = True if not done: raise dnf.exceptions.Error(_('No packages marked for upgrade.')) SUBCMDS = {CheckUpdateSubCommand, InfoSubCommand, InstallSubCommand, ListSubCommand, MoveToSubCommand, ReinstallOldSubCommand, ReinstallSubCommand, RemoveOrDistroSyncSubCommand, RemoveOrReinstallSubCommand, RemoveSubCommand, UpgradeSubCommand} aliases = ('repository-packages', 'repo-pkgs', 'repo-packages', 'repository-pkgs') summary = _('run commands on top of all packages in given repository') def __init__(self, cli): """Initialize the command.""" super(RepoPkgsCommand, self).__init__(cli) subcmd_objs = (subcmd(cli) for subcmd in self.SUBCMDS) self.subcmd = None self._subcmd_name2obj = { alias: subcmd for subcmd in subcmd_objs for alias in subcmd.aliases} def set_argparser(self, parser): narrows = parser.add_mutually_exclusive_group() narrows.add_argument('--all', dest='_pkg_specs_action', action='store_const', const='all', default=None, help=_("show all packages (default)")) narrows.add_argument('--available', dest='_pkg_specs_action', action='store_const', const='available', help=_("show only available packages")) narrows.add_argument('--installed', dest='_pkg_specs_action', action='store_const', const='installed', help=_("show only installed packages")) narrows.add_argument('--extras', dest='_pkg_specs_action', action='store_const', const='extras', help=_("show only extras packages")) narrows.add_argument('--updates', dest='_pkg_specs_action', action='store_const', const='upgrades', help=_("show only upgrades packages")) narrows.add_argument('--upgrades', dest='_pkg_specs_action', action='store_const', const='upgrades', help=_("show only upgrades packages")) narrows.add_argument('--autoremove', dest='_pkg_specs_action', action='store_const', const='autoremove', help=_("show only autoremove packages")) narrows.add_argument('--recent', dest='_pkg_specs_action', action='store_const', const='recent', help=_("show only recently changed packages")) parser.add_argument( 'reponame', nargs=1, action=OptionParser._RepoCallbackEnable, metavar=_('REPOID'), help=_("Repository ID")) subcommand_choices = [subcmd.aliases[0] for subcmd in self.SUBCMDS] subcommand_choices_all = [alias for subcmd in self.SUBCMDS for alias in subcmd.aliases] parser.add_argument('subcmd', nargs=1, metavar="SUBCOMMAND", choices=subcommand_choices_all, help=", ".join(subcommand_choices)) DEFAULT_PKGNARROW = 'all' pkgnarrows = {DEFAULT_PKGNARROW, 'installed', 'available', 'autoremove', 'extras', 'obsoletes', 'recent', 'upgrades'} parser.add_argument('pkg_specs', nargs='', metavar=_('PACKAGE'), choices=pkgnarrows, default=DEFAULT_PKGNARROW, action=OptionParser.PkgNarrowCallback, help=_("Package specification")) def configure(self): """Verify whether the command can run with given arguments.""" # Check sub-command. try: self.subcmd = self._subcmd_name2obj[self.opts.subcmd[0]] except (dnf.cli.CliError, KeyError) as e: self.cli.optparser.print_usage() raise dnf.cli.CliError self.subcmd.opts = self.opts self.subcmd.reponame = self.opts.reponame[0] self.subcmd.configure() def run(self): """Execute the command with respect to given arguments extcmds*.""" self.subcmd.run_on_repo() class HelpCommand(Command): """A class containing methods needed by the cli to execute the help command. """ aliases = ('help',) summary = _('display a helpful usage message') @staticmethod def set_argparser(parser): parser.add_argument('cmd', nargs='?', metavar=_('COMMAND'), help=_("{prog} command to get help for").format( prog=dnf.util.MAIN_PROG_UPPER)) def run(self): if (not self.opts.cmd or self.opts.cmd not in self.cli.cli_commands): self.cli.optparser.print_help() else: command = self.cli.cli_commands[self.opts.cmd] self.cli.optparser.print_help(command(self))	31.38 KB	2022-09-09 11:21
📁 __pycache__	-	2026-03-21 15:47
📄 alias.py # alias.py # Alias CLI command. # # Copyright (C) 2018 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import print_function from __future__ import unicode_literals import logging import os.path import dnf.cli import dnf.cli.aliases from dnf.cli import commands import dnf.conf import dnf.exceptions from dnf.i18n import _ logger = logging.getLogger('dnf') class AliasCommand(commands.Command): aliases = ('alias',) summary = _('List or create command aliases') @staticmethod def set_argparser(parser): enable_group = parser.add_mutually_exclusive_group() enable_group.add_argument( '--enable-resolving', default=False, action='store_true', help=_('enable aliases resolving')) enable_group.add_argument( '--disable-resolving', default=False, action='store_true', help=_('disable aliases resolving')) parser.add_argument("subcommand", nargs='?', default='list', choices=['add', 'list', 'delete'], help=_("action to do with aliases")) parser.add_argument("alias", nargs="*", metavar="command[=result]", help=_("alias definition")) def configure(self): demands = self.cli.demands if self.opts.subcommand in ('add', 'delete'): demands.root_user = True self.aliases_base = dnf.cli.aliases.Aliases() self.aliases_base._load_aliases() self.resolving_enabled = self.aliases_base.enabled self._update_config_from_options() def _update_config_from_options(self): enabled = None if self.opts.enable_resolving: enabled = True logger.info(_("Aliases are now enabled")) if self.opts.disable_resolving: enabled = False logger.info(_("Aliases are now disabled")) if enabled is not None: if not os.path.exists(dnf.cli.aliases.ALIASES_CONF_PATH): open(dnf.cli.aliases.ALIASES_CONF_PATH, 'w').close() dnf.conf.BaseConfig.write_raw_configfile( dnf.cli.aliases.ALIASES_CONF_PATH, 'main', None, {'enabled': enabled}) if not self.aliases_base._disabled_by_environ(): self.aliases_base.enabled = enabled def _parse_option_alias(self): new_aliases = {} for alias in self.opts.alias: alias = alias.split('=', 1) cmd = alias[0].strip() if len(cmd.split()) != 1: logger.warning(_("Invalid alias key: %s"), cmd) continue if cmd.startswith('-'): logger.warning(_("Invalid alias key: %s"), cmd) continue if len(alias) == 1: logger.warning(_("Alias argument has no value: %s"), cmd) continue new_aliases[cmd] = alias[1].split() return new_aliases def _load_user_aliases(self): if not os.path.exists(dnf.cli.aliases.ALIASES_USER_PATH): open(dnf.cli.aliases.ALIASES_USER_PATH, 'w').close() try: conf = dnf.cli.aliases.AliasesConfig( dnf.cli.aliases.ALIASES_USER_PATH) except dnf.exceptions.ConfigError as e: logger.warning(_('Config error: %s'), e) return None return conf def _store_user_aliases(self, user_aliases, enabled): fileobj = open(dnf.cli.aliases.ALIASES_USER_PATH, 'w') output = "[main]\n" output += "enabled = {}\n\n".format(enabled) output += "[aliases]\n" for key, value in user_aliases.items(): output += "{} = {}\n".format(key, ' '.join(value)) fileobj.write(output) def add_aliases(self, aliases): conf = self._load_user_aliases() user_aliases = conf.aliases if user_aliases is None: return user_aliases.update(aliases) self._store_user_aliases(user_aliases, conf.enabled) logger.info(_("Aliases added: %s"), ', '.join(aliases.keys())) def remove_aliases(self, cmds): conf = self._load_user_aliases() user_aliases = conf.aliases if user_aliases is None: return valid_cmds = [] for cmd in cmds: try: del user_aliases[cmd] valid_cmds.append(cmd) except KeyError: logger.info(_("Alias not found: %s"), cmd) self._store_user_aliases(user_aliases, conf.enabled) logger.info(_("Aliases deleted: %s"), ', '.join(valid_cmds)) def list_alias(self, cmd): args = [cmd] try: args = self.aliases_base._resolve(args) except dnf.exceptions.Error as e: logger.error( _('%s, alias %s="%s"'), e, cmd, (' ').join(self.aliases_base.aliases[cmd])) else: print(_("Alias %s='%s'") % (cmd, " ".join(args))) def run(self): if not self.aliases_base.enabled: logger.warning(_("Aliases resolving is disabled.")) if self.opts.subcommand == 'add': # Add new alias aliases = self._parse_option_alias() if not aliases: raise dnf.exceptions.Error(_("No aliases specified.")) self.add_aliases(aliases) return if self.opts.subcommand == 'delete': # Remove alias cmds = self.opts.alias if cmds == []: raise dnf.exceptions.Error(_("No alias specified.")) self.remove_aliases(cmds) return if not self.opts.alias: # List all aliases if not self.aliases_base.aliases: logger.info(_("No aliases defined.")) return for cmd in self.aliases_base.aliases: self.list_alias(cmd) else: # List alias by key for cmd in self.opts.alias: if cmd not in self.aliases_base.aliases: logger.info(_("No match for alias: %s") % cmd) continue self.list_alias(cmd)	6.92 KB	2022-09-09 11:21
📄 autoremove.py # autoremove.py # Autoremove CLI command. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.cli import commands from dnf.cli.option_parser import OptionParser from dnf.i18n import _ import dnf.exceptions import hawkey import logging logger = logging.getLogger("dnf") class AutoremoveCommand(commands.Command): nevra_forms = {'autoremove-n': hawkey.FORM_NAME, 'autoremove-na': hawkey.FORM_NA, 'autoremove-nevra': hawkey.FORM_NEVRA} aliases = ('autoremove',) + tuple(nevra_forms.keys()) summary = _('remove all unneeded packages that were originally installed ' 'as dependencies') @staticmethod def set_argparser(parser): parser.add_argument('packages', nargs='*', help=_('Package to remove'), action=OptionParser.ParseSpecGroupFileCallback, metavar=_('PACKAGE')) def configure(self): demands = self.cli.demands demands.resolving = True demands.root_user = True demands.sack_activation = True if any([self.opts.grp_specs, self.opts.pkg_specs, self.opts.filenames]): self.base.conf.clean_requirements_on_remove = True demands.allow_erasing = True # disable all available repos to delete whole dependency tree # instead of replacing removable package with available packages demands.available_repos = False else: demands.available_repos = True demands.fresh_metadata = False def run(self): if any([self.opts.grp_specs, self.opts.pkg_specs, self.opts.filenames]): forms = [] if self.opts.command in self.nevra_forms: forms = [self.nevra_forms[self.opts.command]] self.base.autoremove(forms, self.opts.pkg_specs, self.opts.grp_specs, self.opts.filenames) else: self.base.autoremove()	2.97 KB	2022-09-09 11:21
📄 check.py # # Copyright (C) 2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import _ from dnf.cli import commands import argparse import dnf.exceptions class CheckCommand(commands.Command): """A class containing methods needed by the cli to execute the check command. """ aliases = ('check',) summary = _('check for problems in the packagedb') @staticmethod def set_argparser(parser): parser.add_argument('--all', dest='check_types', action='append_const', const='all', help=_('show all problems; default')) parser.add_argument('--dependencies', dest='check_types', action='append_const', const='dependencies', help=_('show dependency problems')) parser.add_argument('--duplicates', dest='check_types', action='append_const', const='duplicates', help=_('show duplicate problems')) parser.add_argument('--obsoleted', dest='check_types', action='append_const', const='obsoleted', help=_('show obsoleted packages')) parser.add_argument('--provides', dest='check_types', action='append_const', const='provides', help=_('show problems with provides')) # Add compatibility with yum but invisible in help # In choices [] allows to return empty list if no argument otherwise it fails parser.add_argument('check_yum_types', nargs='*', choices=[ 'all', 'dependencies', 'duplicates', 'obsoleted', 'provides', []], help=argparse.SUPPRESS) def configure(self): self.cli.demands.sack_activation = True if self.opts.check_yum_types: if self.opts.check_types: self.opts.check_types = self.opts.check_types + \ self.opts.check_yum_types else: self.opts.check_types = self.opts.check_yum_types if not self.opts.check_types: self.opts.check_types = {'all'} else: self.opts.check_types = set(self.opts.check_types) self.base.conf.disable_excludes += ["all"] def run(self): output_set = set() q = self.base.sack.query().installed() if self.opts.check_types.intersection({'all', 'dependencies'}): sack = None for pkg in q: for require in set(pkg.regular_requires) \| set(set(pkg.requires_pre) - set(pkg.prereq_ignoreinst)): if str(require).startswith('rpmlib'): continue if not len(q.filter(provides=[require])): if str(require).startswith('('): # rich deps can be only tested by solver if sack is None: sack = dnf.sack.rpmdb_sack(self.base) selector = dnf.selector.Selector(sack) selector.set(provides=str(require)) goal = dnf.goal.Goal(sack) goal.protect_running_kernel = self.base.conf.protect_running_kernel goal.install(select=selector, optional=False) solved = goal.run() # there ase only @system repo in sack, therefore solved is only in case # when rich deps doesn't require any additional package if solved: continue msg = _("{} has missing requires of {}") output_set.add(msg.format( self.base.output.term.bold(pkg), self.base.output.term.bold(require))) for conflict in pkg.conflicts: conflicted = q.filter(provides=[conflict], name=str(conflict).split()[0]) for conflict_pkg in conflicted: msg = '{} has installed conflict "{}": {}' output_set.add(msg.format( self.base.output.term.bold(pkg), self.base.output.term.bold(conflict), self.base.output.term.bold(conflict_pkg))) if self.opts.check_types.intersection({'all', 'duplicates'}): installonly = self.base._get_installonly_query(q) dups = q.duplicated().difference(installonly)._name_dict() for name, pkgs in dups.items(): pkgs.sort() for dup in pkgs[1:]: msg = _("{} is a duplicate with {}").format( self.base.output.term.bold(pkgs[0]), self.base.output.term.bold(dup)) output_set.add(msg) if self.opts.check_types.intersection({'all', 'obsoleted'}): for pkg in q: for obsolete in pkg.obsoletes: obsoleted = q.filter(provides=[obsolete], name=str(obsolete).split()[0]) if len(obsoleted): msg = _("{} is obsoleted by {}").format( self.base.output.term.bold(obsoleted[0]), self.base.output.term.bold(pkg)) output_set.add(msg) if self.opts.check_types.intersection({'all', 'provides'}): for pkg in q: for provide in pkg.provides: if pkg not in q.filter(provides=[provide]): msg = _("{} provides {} but it cannot be found") output_set.add(msg.format( self.base.output.term.bold(pkg), self.base.output.term.bold(provide))) for msg in sorted(output_set): print(msg) if output_set: raise dnf.exceptions.Error( 'Check discovered {} problem(s)'.format(len(output_set)))	7.06 KB	2022-09-09 11:21
📄 clean.py # clean.py # Clean CLI command. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.cli import commands from dnf.i18n import _, P_ from dnf.yum import misc import dnf.cli import dnf.exceptions import dnf.lock import dnf.logging import dnf.repo import logging import os import re import time logger = logging.getLogger("dnf") # Dict mapping cmdline arguments to actual data types to be cleaned up _CACHE_TYPES = { 'metadata': ['metadata', 'dbcache', 'expire-cache'], 'packages': ['packages'], 'dbcache': ['dbcache'], 'expire-cache': ['expire-cache'], 'all': ['metadata', 'packages', 'dbcache'], } def _tree(dirpath): """Traverse dirpath recursively and yield relative filenames.""" for root, dirs, files in os.walk(dirpath): base = os.path.relpath(root, dirpath) for f in files: path = os.path.join(base, f) yield os.path.normpath(path) def _filter(files, patterns): """Yield those filenames that match any of the patterns.""" return (f for f in files for p in patterns if re.match(p, f)) def _clean(dirpath, files): """Remove the given filenames from dirpath.""" count = 0 for f in files: path = os.path.join(dirpath, f) logger.log(dnf.logging.DDEBUG, _('Removing file %s'), path) misc.unlink_f(path) count += 1 return count def _cached_repos(files): """Return the repo IDs that have some cached metadata around.""" metapat = dnf.repo.CACHE_FILES['metadata'] matches = (re.match(metapat, f) for f in files) return set(m.group('repoid') for m in matches if m) class CleanCommand(commands.Command): """A class containing methods needed by the cli to execute the clean command. """ aliases = ('clean',) summary = _('remove cached data') @staticmethod def set_argparser(parser): parser.add_argument('type', nargs='+', choices=_CACHE_TYPES.keys(), help=_('Metadata type to clean')) def run(self): cachedir = self.base.conf.cachedir md_lock = dnf.lock.build_metadata_lock(cachedir, True) download_lock = dnf.lock.build_download_lock(cachedir, True) rpmdb_lock = dnf.lock.build_rpmdb_lock(self.base.conf.persistdir, True) while True: try: with md_lock, download_lock, rpmdb_lock: types = set(t for c in self.opts.type for t in _CACHE_TYPES[c]) files = list(_tree(cachedir)) logger.debug(_('Cleaning data: ' + ' '.join(types))) if 'expire-cache' in types: expired = _cached_repos(files) self.base._repo_persistor.expired_to_add.update(expired) types.remove('expire-cache') logger.info(_('Cache was expired')) patterns = [dnf.repo.CACHE_FILES[t] for t in types] count = _clean(cachedir, _filter(files, patterns)) logger.info(P_('%d file removed', '%d files removed', count) % count) return except dnf.exceptions.LockError as e: if not self.base.conf.exit_on_lock: msg = _('Waiting for process with pid %d to finish.') % (e.pid) logger.info(msg) time.sleep(3) else: raise e	4.36 KB	2022-09-09 11:21
📄 deplist.py # # Copyright (C) 2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import print_function from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import _ from dnf.cli.commands.repoquery import RepoQueryCommand class DeplistCommand(RepoQueryCommand): """ The command is alias for 'dnf repoquery --deplist' """ aliases = ('deplist',) summary = _("[deprecated, use repoquery --deplist] List package's dependencies and what packages provide them") def configure(self): RepoQueryCommand.configure(self) self.opts.deplist = True	1.47 KB	2022-09-09 11:21
📄 distrosync.py # distrosync.py # distro-sync CLI command. # # Copyright (C) 2012-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from dnf.cli import commands from dnf.i18n import _ class DistroSyncCommand(commands.Command): """A class containing methods needed by the cli to execute the distro-synch command. """ aliases = ('distro-sync', 'distrosync', 'distribution-synchronization', 'dsync') summary = _('synchronize installed packages to the latest available versions') @staticmethod def set_argparser(parser): parser.add_argument('package', nargs='*', help=_('Package to synchronize')) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True commands._checkGPGKey(self.base, self.cli) commands._checkEnabledRepo(self.base, self.opts.package) def run(self): return self.base.distro_sync_userlist(self.opts.package)	1.91 KB	2022-09-09 11:21
📄 downgrade.py # downgrade.py # Downgrade CLI command. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.cli import commands from dnf.cli.option_parser import OptionParser from dnf.i18n import _ class DowngradeCommand(commands.Command): """A class containing methods needed by the cli to execute the downgrade command. """ aliases = ('downgrade', 'dg') summary = _("Downgrade a package") @staticmethod def set_argparser(parser): parser.add_argument('package', nargs='*', help=_('Package to downgrade'), action=OptionParser.ParseSpecGroupFileCallback) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True commands._checkGPGKey(self.base, self.cli) if not self.opts.filenames: commands._checkEnabledRepo(self.base) def run(self): file_pkgs = self.base.add_remote_rpms(self.opts.filenames, strict=False, progress=self.base.output.progress) return self.base.downgradePkgs( specs=self.opts.pkg_specs + ['@' + x for x in self.opts.grp_specs], file_pkgs=file_pkgs, strict=self.base.conf.strict)	2.27 KB	2022-09-09 11:21
📄 group.py # group.py # Group CLI command. # # Copyright (C) 2012-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.comps import CompsQuery from dnf.cli import commands from dnf.i18n import _, ucd import libdnf.transaction import dnf.cli import dnf.exceptions import dnf.util import logging logger = logging.getLogger("dnf") class GroupCommand(commands.Command): """ Single sub-command interface for most groups interaction. """ direct_commands = {'grouplist' : 'list', 'groupinstall' : 'install', 'groupupdate' : 'install', 'groupremove' : 'remove', 'grouperase' : 'remove', 'groupinfo' : 'info'} aliases = ('group', 'groups', 'grp') + tuple(direct_commands.keys()) summary = _('display, or use, the groups information') _CMD_ALIASES = {'update' : 'upgrade', 'erase' : 'remove'} _MARK_CMDS = ('install', 'remove') _GROUP_SUBCOMMANDS = ('summary', 'list', 'info', 'remove', 'install', 'upgrade', 'mark') def _canonical(self): # were we called with direct command? direct = self.direct_commands.get(self.opts.command) if direct: # canonize subcmd and args if self.opts.subcmd is not None: self.opts.args.insert(0, self.opts.subcmd) self.opts.subcmd = direct if self.opts.subcmd is None: self.opts.subcmd = 'summary' self.opts.subcmd = self._CMD_ALIASES.get(self.opts.subcmd, self.opts.subcmd) def __init__(self, cli): super(GroupCommand, self).__init__(cli) self._remark = False def _assert_comps(self): msg = _('No group data available for configured repositories.') if not len(self.base.comps): raise dnf.exceptions.CompsError(msg) def _environment_lists(self, patterns): def available_pred(env): env_found = self.base.history.env.get(env.id) return not(env_found) self._assert_comps() if patterns is None: envs = self.base.comps.environments else: envs = self.base.comps.environments_by_pattern(",".join(patterns)) return dnf.util.mapall(list, dnf.util.partition(available_pred, envs)) def _group_lists(self, uservisible, patterns): def installed_pred(group): group_found = self.base.history.group.get(group.id) if group_found: return True return False installed = [] available = [] self._assert_comps() if patterns is None: grps = self.base.comps.groups else: grps = self.base.comps.groups_by_pattern(",".join(patterns)) for grp in grps: tgt_list = available if installed_pred(grp): tgt_list = installed if not uservisible or grp.uservisible: tgt_list.append(grp) return installed, available def _info(self, userlist): for strng in userlist: group_matched = False for env in self.base.comps.environments_by_pattern(strng): self.output.display_groups_in_environment(env) group_matched = True for group in self.base.comps.groups_by_pattern(strng): self.output.display_pkgs_in_groups(group) group_matched = True if not group_matched: logger.error(_('Warning: Group %s does not exist.'), strng) return 0, [] def _list(self, userlist): uservisible = 1 showinstalled = 0 showavailable = 0 print_ids = self.base.conf.verbose or self.opts.ids while userlist: if userlist[0] == 'hidden': uservisible = 0 userlist.pop(0) elif userlist[0] == 'installed': showinstalled = 1 userlist.pop(0) elif userlist[0] == 'available': showavailable = 1 userlist.pop(0) elif userlist[0] == 'ids': print_ids = True userlist.pop(0) else: break if self.opts.hidden: uservisible = 0 if self.opts.installed: showinstalled = 1 if self.opts.available: showavailable = 1 if not userlist: userlist = None # Match everything... errs = False if userlist is not None: for group in userlist: comps = self.base.comps in_group = len(comps.groups_by_pattern(group)) > 0 in_environment = len(comps.environments_by_pattern(group)) > 0 if not in_group and not in_environment: logger.error(_('Warning: No groups match:') + '\n %s', group) errs = True if errs: return 0, [] env_inst, env_avail = self._environment_lists(userlist) installed, available = self._group_lists(uservisible, userlist) def _out_grp(sect, group): if not done: print(sect) msg = ' %s' % dnf.util._name_unset_wrapper(group.ui_name) if print_ids: msg += ' (%s)' % group.id if group.lang_only: msg += ' [%s]' % group.lang_only print('{}'.format(msg)) def _out_env(sect, envs): if envs: print(sect) for e in envs: msg = ' %s' % dnf.util._name_unset_wrapper(e.ui_name) if print_ids: msg += ' (%s)' % e.id print(msg) if not showinstalled: _out_env(_('Available Environment Groups:'), env_avail) if not showavailable: _out_env(_('Installed Environment Groups:'), env_inst) if not showavailable: done = False for group in installed: if group.lang_only: continue _out_grp(_('Installed Groups:'), group) done = True done = False for group in installed: if not group.lang_only: continue _out_grp(_('Installed Language Groups:'), group) done = True if showinstalled: return 0, [] done = False for group in available: if group.lang_only: continue _out_grp(_('Available Groups:'), group) done = True done = False for group in available: if not group.lang_only: continue _out_grp(_('Available Language Groups:'), group) done = True return 0, [] def _mark_install(self, patterns): q = CompsQuery(self.base.comps, self.base.history, CompsQuery.GROUPS \| CompsQuery.ENVIRONMENTS, CompsQuery.AVAILABLE \| CompsQuery.INSTALLED) solver = self.base._build_comps_solver() res = q.get(patterns) if self.opts.with_optional: types = tuple(self.base.conf.group_package_types + ['optional']) else: types = tuple(self.base.conf.group_package_types) pkg_types = libdnf.transaction.listToCompsPackageType(types) for env_id in res.environments: solver._environment_install(env_id, pkg_types) for group_id in res.groups: solver._group_install(group_id, pkg_types) def _mark_remove(self, patterns): q = CompsQuery(self.base.comps, self.base.history, CompsQuery.GROUPS \| CompsQuery.ENVIRONMENTS, CompsQuery.INSTALLED) solver = self.base._build_comps_solver() res = q.get(patterns) for env_id in res.environments: assert dnf.util.is_string_type(env_id) solver._environment_remove(env_id) for grp_id in res.groups: assert dnf.util.is_string_type(grp_id) solver._group_remove(grp_id) def _mark_subcmd(self, extcmds): if extcmds[0] in self._MARK_CMDS: return extcmds[0], extcmds[1:] return 'install', extcmds def _summary(self, userlist): uservisible = 1 if len(userlist) > 0: if userlist[0] == 'hidden': uservisible = 0 userlist.pop(0) if self.opts.hidden: uservisible = 0 if not userlist: userlist = None # Match everything... installed, available = self._group_lists(uservisible, userlist) def _out_grp(sect, num): if not num: return logger.info('%s %u', sect, num) done = 0 for group in installed: if group.lang_only: continue done += 1 _out_grp(_('Installed Groups:'), done) done = 0 for group in installed: if not group.lang_only: continue done += 1 _out_grp(_('Installed Language Groups:'), done) done = False for group in available: if group.lang_only: continue done += 1 _out_grp(_('Available Groups:'), done) done = False for group in available: if not group.lang_only: continue done += 1 _out_grp(_('Available Language Groups:'), done) return 0, [] @staticmethod def set_argparser(parser): parser.add_argument('--with-optional', action='store_true', help=_("include optional packages from group")) grpparser = parser.add_mutually_exclusive_group() grpparser.add_argument('--hidden', action='store_true', help=_("show also hidden groups")) grpparser.add_argument('--installed', action='store_true', help=_("show only installed groups")) grpparser.add_argument('--available', action='store_true', help=_("show only available groups")) grpparser.add_argument('--ids', action='store_true', help=_("show also ID of groups")) parser.add_argument('subcmd', nargs='?', metavar='COMMAND', help=_('available subcommands: {} (default), {}').format( GroupCommand._GROUP_SUBCOMMANDS[0], ', '.join(GroupCommand._GROUP_SUBCOMMANDS[1:]))) parser.add_argument('args', nargs='*', metavar='COMMAND_ARG', help=_('argument for group subcommand')) def configure(self): self._canonical() cmd = self.opts.subcmd args = self.opts.args if cmd not in self._GROUP_SUBCOMMANDS: logger.critical(_('Invalid groups sub-command, use: %s.'), ", ".join(self._GROUP_SUBCOMMANDS)) raise dnf.cli.CliError if cmd in ('install', 'remove', 'mark', 'info') and not args: self.cli.optparser.print_help(self) raise dnf.cli.CliError demands = self.cli.demands demands.sack_activation = True if cmd in ('install', 'mark', 'remove', 'upgrade'): demands.root_user = True demands.resolving = True if cmd == 'remove': demands.allow_erasing = True demands.available_repos = False else: demands.available_repos = True if cmd not in ('remove'): commands._checkEnabledRepo(self.base) if cmd in ('install', 'upgrade'): commands._checkGPGKey(self.base, self.cli) def run(self): cmd = self.opts.subcmd extcmds = self.opts.args if cmd == 'summary': return self._summary(extcmds) if cmd == 'list': return self._list(extcmds) if cmd == 'info': return self._info(extcmds) if cmd == 'mark': (subcmd, extcmds) = self._mark_subcmd(extcmds) if subcmd == 'remove': return self._mark_remove(extcmds) else: assert subcmd == 'install' return self._mark_install(extcmds) if cmd == 'install': if self.opts.with_optional: types = tuple(self.base.conf.group_package_types + ['optional']) else: types = tuple(self.base.conf.group_package_types) self._remark = True try: return self.base.env_group_install(extcmds, types, self.base.conf.strict) except dnf.exceptions.MarkingError as e: msg = _('No package %s available.') logger.info(msg, self.base.output.term.bold(e)) raise dnf.exceptions.PackagesNotAvailableError( _("Unable to find a mandatory group package.")) if cmd == 'upgrade': return self.base.env_group_upgrade(extcmds) if cmd == 'remove': for arg in extcmds: try: self.base.env_group_remove([arg]) except dnf.exceptions.Error: pass def run_transaction(self): if not self._remark: return goal = self.base._goal history = self.base.history names = goal.group_members for pkg in self.base.sack.query().installed().filterm(name=names): reason = history.rpm.get_reason(pkg) history.set_reason(pkg, goal.group_reason(pkg, reason))	14.61 KB	2022-09-09 11:21
📄 history.py # Copyright 2006 Duke University # Copyright (C) 2012-2016 Red Hat, Inc. # # 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 Library General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. from __future__ import absolute_import from __future__ import print_function from __future__ import unicode_literals import libdnf import hawkey from dnf.i18n import _, ucd from dnf.cli import commands from dnf.transaction_sr import TransactionReplay, serialize_transaction import dnf.cli import dnf.exceptions import dnf.transaction import dnf.util import json import logging import os logger = logging.getLogger('dnf') class HistoryCommand(commands.Command): """A class containing methods needed by the cli to execute the history command. """ aliases = ('history', 'hist') summary = _('display, or use, the transaction history') _CMDS = ['list', 'info', 'redo', 'replay', 'rollback', 'store', 'undo', 'userinstalled'] def __init__(self, args, kw): super(HistoryCommand, self).__init__(args, *kw) self._require_one_transaction_id = False @staticmethod def set_argparser(parser): parser.add_argument('transactions_action', nargs='?', metavar="COMMAND", help="Available commands: {} (default), {}".format( HistoryCommand._CMDS[0], ", ".join(HistoryCommand._CMDS[1:]))) parser.add_argument('--reverse', action='store_true', help="display history list output reversed") parser.add_argument("-o", "--output", default=None, help=_("For the store command, file path to store the transaction to")) parser.add_argument("--ignore-installed", action="store_true", help=_("For the replay command, don't check for installed packages matching " "those in transaction")) parser.add_argument("--ignore-extras", action="store_true", help=_("For the replay command, don't check for extra packages pulled " "into the transaction")) parser.add_argument("--skip-unavailable", action="store_true", help=_("For the replay command, skip packages that are not available or have " "missing dependencies")) parser.add_argument('transactions', nargs='', metavar="TRANSACTION", help="For commands working with history transactions, " "Transaction ID (<number>, 'last' or 'last-<number>' " "for one transaction, <transaction-id>..<transaction-id> " "for a range)") parser.add_argument('transaction_filename', nargs='?', metavar="TRANSACTION_FILE", help="For the replay command, path to the stored " "transaction file to replay") def configure(self): if not self.opts.transactions_action: # no positional argument given self.opts.transactions_action = self._CMDS[0] elif self.opts.transactions_action not in self._CMDS: # first positional argument is not a command self.opts.transactions.insert(0, self.opts.transactions_action) self.opts.transactions_action = self._CMDS[0] self._require_one_transaction_id_msg = _("Found more than one transaction ID.\n" "'{}' requires one transaction ID or package name." ).format(self.opts.transactions_action) demands = self.cli.demands if self.opts.transactions_action == 'replay': if not self.opts.transactions: raise dnf.cli.CliError(_('No transaction file name given.')) if len(self.opts.transactions) > 1: raise dnf.cli.CliError(_('More than one argument given as transaction file name.')) # in case of replay, copy over the file name to it's appropriate variable # (the arg parser can't distinguish here) self.opts.transaction_filename = os.path.abspath(self.opts.transactions[0]) self.opts.transactions = [] demands.available_repos = True demands.resolving = True demands.root_user = True # Override configuration options that affect how the transaction is resolved self.base.conf.clean_requirements_on_remove = False self.base.conf.install_weak_deps = False dnf.cli.commands._checkGPGKey(self.base, self.cli) elif self.opts.transactions_action == 'store': self._require_one_transaction_id = True if not self.opts.transactions: raise dnf.cli.CliError(_('No transaction ID or package name given.')) elif self.opts.transactions_action in ['redo', 'undo', 'rollback']: demands.available_repos = True demands.resolving = True demands.root_user = True self._require_one_transaction_id = True if not self.opts.transactions: msg = _('No transaction ID or package name given.') logger.critical(msg) raise dnf.cli.CliError(msg) elif len(self.opts.transactions) > 1: logger.critical(self._require_one_transaction_id_msg) raise dnf.cli.CliError(self._require_one_transaction_id_msg) demands.available_repos = True dnf.cli.commands._checkGPGKey(self.base, self.cli) else: demands.fresh_metadata = False demands.sack_activation = True if self.base.history.path != ":memory:" and not os.access(self.base.history.path, os.R_OK): msg = _("You don't have access to the history DB: %s" % self.base.history.path) logger.critical(msg) raise dnf.cli.CliError(msg) def get_error_output(self, error): """Get suggestions for resolving the given error.""" if isinstance(error, dnf.exceptions.TransactionCheckError): if self.opts.transactions_action == 'undo': id_, = self.opts.transactions return (_('Cannot undo transaction %s, doing so would result ' 'in an inconsistent package database.') % id_,) elif self.opts.transactions_action == 'rollback': id_, = (self.opts.transactions if self.opts.transactions[0] != 'force' else self.opts.transactions[1:]) return (_('Cannot rollback transaction %s, doing so would ' 'result in an inconsistent package database.') % id_,) return dnf.cli.commands.Command.get_error_output(self, error) def _hcmd_redo(self, extcmds): old = self._history_get_transaction(extcmds) data = serialize_transaction(old) self.replay = TransactionReplay( self.base, data=data, ignore_installed=True, ignore_extras=True, skip_unavailable=self.opts.skip_unavailable ) self.replay.run() def _history_get_transactions(self, extcmds): if not extcmds: raise dnf.cli.CliError(_('No transaction ID given')) old = self.base.history.old(extcmds) if not old: raise dnf.cli.CliError(_('Transaction ID "{0}" not found.').format(extcmds[0])) return old def _history_get_transaction(self, extcmds): old = self._history_get_transactions(extcmds) if len(old) > 1: raise dnf.cli.CliError(_('Found more than one transaction ID!')) return old[0] def _hcmd_undo(self, extcmds): old = self._history_get_transaction(extcmds) self._revert_transaction(old) def _hcmd_rollback(self, extcmds): old = self._history_get_transaction(extcmds) last = self.base.history.last() merged_trans = None if old.tid != last.tid: # history.old([]) returns all transactions and we don't want that # so skip merging the transactions when trying to rollback to the last transaction # which is the current system state and rollback is not applicable for trans in self.base.history.old(list(range(old.tid + 1, last.tid + 1))): if trans.altered_lt_rpmdb: logger.warning(_('Transaction history is incomplete, before %u.'), trans.tid) elif trans.altered_gt_rpmdb: logger.warning(_('Transaction history is incomplete, after %u.'), trans.tid) if merged_trans is None: merged_trans = dnf.db.history.MergedTransactionWrapper(trans) else: merged_trans.merge(trans) self._revert_transaction(merged_trans) def _revert_transaction(self, trans): action_map = { "Install": "Removed", "Removed": "Install", "Upgrade": "Downgraded", "Upgraded": "Downgrade", "Downgrade": "Upgraded", "Downgraded": "Upgrade", "Reinstalled": "Reinstall", "Reinstall": "Reinstalled", "Obsoleted": "Install", "Obsolete": "Obsoleted", "Reason Change": "Reason Change", } data = serialize_transaction(trans) # revert actions in the serialized transaction data to perform rollback/undo for content_type in ("rpms", "groups", "environments"): for ti in data.get(content_type, []): ti["action"] = action_map[ti["action"]] if ti["action"] == "Install" and ti.get("reason", None) == "clean": ti["reason"] = "dependency" if ti["action"] == "Reason Change" and "nevra" in ti: subj = hawkey.Subject(ti["nevra"]) nevra = subj.get_nevra_possibilities(forms=[hawkey.FORM_NEVRA])[0] reason = self.output.history.swdb.resolveRPMTransactionItemReason( nevra.name, nevra.arch, trans.tids()[0] - 1 ) ti["reason"] = libdnf.transaction.TransactionItemReasonToString(reason) if ti.get("repo_id") == hawkey.SYSTEM_REPO_NAME: # erase repo_id, because it's not possible to perform forward actions from the @System repo ti["repo_id"] = None self.replay = TransactionReplay( self.base, data=data, ignore_installed=True, ignore_extras=True, skip_unavailable=self.opts.skip_unavailable ) self.replay.run() def _hcmd_userinstalled(self): """Execute history userinstalled command.""" pkgs = tuple(self.base.iter_userinstalled()) n_listed = self.output.listPkgs(pkgs, 'Packages installed by user', 'nevra') if n_listed == 0: raise dnf.cli.CliError(_('No packages to list')) def _args2transaction_ids(self): """Convert commandline arguments to transaction ids""" def str2transaction_id(s): if s == 'last': s = '0' elif s.startswith('last-'): s = s[4:] transaction_id = int(s) if transaction_id <= 0: transaction_id += self.output.history.last().tid return transaction_id tids = set() merged_tids = set() for t in self.opts.transactions: if '..' in t: try: begin_transaction_id, end_transaction_id = t.split('..', 2) except ValueError: logger.critical( _("Invalid transaction ID range definition '{}'.\n" "Use '<transaction-id>..<transaction-id>'." ).format(t)) raise dnf.cli.CliError cant_convert_msg = _("Can't convert '{}' to transaction ID.\n" "Use '<number>', 'last', 'last-<number>'.") try: begin_transaction_id = str2transaction_id(begin_transaction_id) except ValueError: logger.critical(_(cant_convert_msg).format(begin_transaction_id)) raise dnf.cli.CliError try: end_transaction_id = str2transaction_id(end_transaction_id) except ValueError: logger.critical(_(cant_convert_msg).format(end_transaction_id)) raise dnf.cli.CliError if self._require_one_transaction_id and begin_transaction_id != end_transaction_id: logger.critical(self._require_one_transaction_id_msg) raise dnf.cli.CliError if begin_transaction_id > end_transaction_id: begin_transaction_id, end_transaction_id = \ end_transaction_id, begin_transaction_id merged_tids.add((begin_transaction_id, end_transaction_id)) tids.update(range(begin_transaction_id, end_transaction_id + 1)) else: try: tids.add(str2transaction_id(t)) except ValueError: # not a transaction id, assume it's package name transact_ids_from_pkgname = self.output.history.search([t]) if transact_ids_from_pkgname: tids.update(transact_ids_from_pkgname) else: msg = _("No transaction which manipulates package '{}' was found." ).format(t) if self._require_one_transaction_id: logger.critical(msg) raise dnf.cli.CliError else: logger.info(msg) return sorted(tids, reverse=True), merged_tids def run(self): vcmd = self.opts.transactions_action if vcmd == 'replay': self.replay = TransactionReplay( self.base, filename=self.opts.transaction_filename, ignore_installed = self.opts.ignore_installed, ignore_extras = self.opts.ignore_extras, skip_unavailable = self.opts.skip_unavailable ) self.replay.run() else: tids, merged_tids = self._args2transaction_ids() if vcmd == 'list' and (tids or not self.opts.transactions): self.output.historyListCmd(tids, reverse=self.opts.reverse) elif vcmd == 'info' and (tids or not self.opts.transactions): self.output.historyInfoCmd(tids, self.opts.transactions, merged_tids) elif vcmd == 'undo': self._hcmd_undo(tids) elif vcmd == 'redo': self._hcmd_redo(tids) elif vcmd == 'rollback': self._hcmd_rollback(tids) elif vcmd == 'userinstalled': self._hcmd_userinstalled() elif vcmd == 'store': tid = self._history_get_transaction(tids) data = serialize_transaction(tid) try: filename = self.opts.output if self.opts.output is not None else "transaction.json" # it is absolutely possible for both assumeyes and assumeno to be True, go figure if (self.base.conf.assumeno or not self.base.conf.assumeyes) and os.path.isfile(filename): msg = _("{} exists, overwrite?").format(filename) if self.base.conf.assumeno or not self.base.output.userconfirm( msg='\n{} [y/N]: '.format(msg), defaultyes_msg='\n{} [Y/n]: '.format(msg)): print(_("Not overwriting {}, exiting.").format(filename)) return with open(filename, "w") as f: json.dump(data, f, indent=4, sort_keys=True) f.write("\n") print(_("Transaction saved to {}.").format(filename)) except OSError as e: raise dnf.cli.CliError(_('Error storing transaction: {}').format(str(e))) def run_resolved(self): if self.opts.transactions_action not in ("replay", "redo", "rollback", "undo"): return self.replay.post_transaction() def run_transaction(self): if self.opts.transactions_action not in ("replay", "redo", "rollback", "undo"): return warnings = self.replay.get_warnings() if warnings: logger.log( dnf.logging.WARNING, _("Warning, the following problems occurred while running a transaction:") ) for w in warnings: logger.log(dnf.logging.WARNING, " " + w)	17.54 KB	2022-09-09 11:21
📄 install.py # install.py # Install CLI command. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals import logging from itertools import chain import hawkey import dnf.exceptions from dnf.cli import commands from dnf.cli.option_parser import OptionParser from dnf.i18n import _ logger = logging.getLogger('dnf') class InstallCommand(commands.Command): """A class containing methods needed by the cli to execute the install command. """ nevra_forms = {'install-n': hawkey.FORM_NAME, 'install-na': hawkey.FORM_NA, 'install-nevra': hawkey.FORM_NEVRA} alternatives_provide = 'alternative-for({})' aliases = ('install', 'localinstall', 'in') + tuple(nevra_forms.keys()) summary = _('install a package or packages on your system') @staticmethod def set_argparser(parser): parser.add_argument('package', nargs='+', metavar=_('PACKAGE'), action=OptionParser.ParseSpecGroupFileCallback, help=_('Package to install')) def configure(self): """Verify that conditions are met so that this command can run. That there are enabled repositories with gpg keys, and that this command is called with appropriate arguments. """ demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True commands._checkGPGKey(self.base, self.cli) if not self.opts.filenames: commands._checkEnabledRepo(self.base) def run(self): err_pkgs = [] errs = [] error_module_specs = [] nevra_forms = self._get_nevra_forms_from_command() self.cli._populate_update_security_filter(self.opts) if self.opts.command == 'localinstall' and (self.opts.grp_specs or self.opts.pkg_specs): self._log_not_valid_rpm_file_paths(self.opts.grp_specs) if self.base.conf.strict: raise dnf.exceptions.Error(_('Nothing to do.')) skipped_grp_specs = [] if self.opts.grp_specs and self.opts.command != 'localinstall': if dnf.base.WITH_MODULES: try: module_base = dnf.module.module_base.ModuleBase(self.base) module_base.install(self.opts.grp_specs, strict=self.base.conf.strict) except dnf.exceptions.MarkingErrors as e: if e.no_match_group_specs: for e_spec in e.no_match_group_specs: skipped_grp_specs.append(e_spec) if e.error_group_specs: for e_spec in e.error_group_specs: error_module_specs.append("@" + e_spec) module_depsolv_errors = e.module_depsolv_errors if module_depsolv_errors: logger.error(dnf.module.module_base.format_modular_solver_errors( module_depsolv_errors[0])) else: skipped_grp_specs = self.opts.grp_specs if self.opts.filenames and nevra_forms: self._inform_not_a_valid_combination(self.opts.filenames) if self.base.conf.strict: raise dnf.exceptions.Error(_('Nothing to do.')) else: err_pkgs = self._install_files() if skipped_grp_specs and nevra_forms: self._inform_not_a_valid_combination(skipped_grp_specs) if self.base.conf.strict: raise dnf.exceptions.Error(_('Nothing to do.')) elif skipped_grp_specs and self.opts.command != 'localinstall': self._install_groups(skipped_grp_specs) if self.opts.command != 'localinstall': errs = self._install_packages(nevra_forms) if (len(errs) != 0 or len(err_pkgs) != 0 or error_module_specs) and self.base.conf.strict: raise dnf.exceptions.PackagesNotAvailableError(_("Unable to find a match"), pkg_spec=' '.join(errs), packages=err_pkgs) def _get_nevra_forms_from_command(self): if self.opts.command in self.nevra_forms: return [self.nevra_forms[self.opts.command]] else: return [] def _log_not_valid_rpm_file_paths(self, grp_specs): group_names = map(lambda g: '@' + g, grp_specs) for pkg in chain(self.opts.pkg_specs, group_names): msg = _('Not a valid rpm file path: %s') logger.info(msg, self.base.output.term.bold(pkg)) def _inform_not_a_valid_combination(self, forms): for form in forms: msg = _('Not a valid form: %s') logger.warning(msg, self.base.output.term.bold(form)) def _install_files(self): err_pkgs = [] strict = self.base.conf.strict for pkg in self.base.add_remote_rpms(self.opts.filenames, strict=strict, progress=self.base.output.progress): try: self.base.package_install(pkg, strict=strict) except dnf.exceptions.MarkingError: msg = _('No match for argument: %s') logger.info(msg, self.base.output.term.bold(pkg.location)) err_pkgs.append(pkg) return err_pkgs def _install_groups(self, grp_specs): try: self.base.env_group_install(grp_specs, tuple(self.base.conf.group_package_types), strict=self.base.conf.strict) except dnf.exceptions.Error: if self.base.conf.strict: raise def _report_alternatives(self, pkg_spec): query = self.base.sack.query().filterm( provides=self.alternatives_provide.format(pkg_spec)) if query: msg = _('There are following alternatives for "{0}": {1}') logger.info(msg.format( pkg_spec, ', '.join(sorted(set([alt.name for alt in query]))))) def _install_packages(self, nevra_forms): errs = [] strict = self.base.conf.strict for pkg_spec in self.opts.pkg_specs: try: self.base.install(pkg_spec, strict=strict, forms=nevra_forms) except dnf.exceptions.MarkingError as e: msg = '{}: {}'.format(e.value, self.base.output.term.bold(pkg_spec)) logger.info(msg) self.base._report_icase_hint(pkg_spec) self._report_alternatives(pkg_spec) errs.append(pkg_spec) return errs	7.58 KB	2022-09-09 11:21
📄 makecache.py # makecache.py # Makecache CLI command. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.cli import commands from dnf.i18n import _ import argparse import dnf.cli import dnf.exceptions import dnf.util import logging logger = logging.getLogger("dnf") class MakeCacheCommand(commands.Command): aliases = ('makecache', 'mc') summary = _('generate the metadata cache') @staticmethod def set_argparser(parser): parser.add_argument('--timer', action='store_true', dest="timer_opt") # compatibility with dnf < 2.0 parser.add_argument('timer', nargs='?', choices=['timer'], metavar='timer', help=argparse.SUPPRESS) def run(self): timer = self.opts.timer is not None or self.opts.timer_opt msg = _("Making cache files for all metadata files.") logger.debug(msg) return self.base.update_cache(timer)	1.86 KB	2022-09-09 11:21
📄 mark.py # mark.py # Mark CLI command. # # Copyright (C) 2015-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import print_function from __future__ import unicode_literals import libdnf.transaction from dnf.i18n import _ from dnf.cli import commands import dnf import functools import logging logger = logging.getLogger("dnf") class MarkCommand(commands.Command): aliases = ('mark',) summary = _('mark or unmark installed packages as installed by user.') @staticmethod def set_argparser(parser): parser.add_argument('mark', nargs=1, choices=['install', 'remove', 'group'], help=_("install: mark as installed by user\n" "remove: unmark as installed by user\n" "group: mark as installed by group")) parser.add_argument('package', nargs='+', metavar="PACKAGE", help=_("Package specification")) def _mark_install(self, pkg): self.base.history.set_reason(pkg, libdnf.transaction.TransactionItemReason_USER) logger.info(_('%s marked as user installed.'), str(pkg)) def _mark_remove(self, pkg): self.base.history.set_reason(pkg, libdnf.transaction.TransactionItemReason_DEPENDENCY) logger.info(_('%s unmarked as user installed.'), str(pkg)) def _mark_group(self, pkg): self.base.history.set_reason(pkg, libdnf.transaction.TransactionItemReason_GROUP) logger.info(_('%s marked as group installed.'), str(pkg)) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.root_user = True demands.available_repos = False demands.resolving = False def run(self): cmd = self.opts.mark[0] pkgs = self.opts.package mark_func = functools.partial(getattr(self, '_mark_' + cmd)) notfound = [] for pkg in pkgs: subj = dnf.subject.Subject(pkg) q = subj.get_best_query(self.base.sack) for pkg in q: mark_func(pkg) if len(q) == 0: notfound.append(pkg) if notfound: logger.error(_('Error:')) for pkg in notfound: logger.error(_('Package %s is not installed.'), pkg) raise dnf.cli.CliError old = self.base.history.last() if old is None: rpmdb_version = self.base._ts.dbCookie() else: rpmdb_version = old.end_rpmdb_version self.base.history.beg(rpmdb_version, [], []) self.base.history.end(rpmdb_version)	3.45 KB	2022-09-09 11:21
📄 module.py # supplies the 'module' command. # # Copyright (C) 2014-2017 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import print_function from dnf.cli import commands, CliError from dnf.i18n import _ from dnf.module.exceptions import NoModuleException from dnf.util import logger import dnf.util import sys import os import hawkey import libdnf import dnf.module.module_base import dnf.exceptions class ModuleCommand(commands.Command): class SubCommand(commands.Command): def __init__(self, cli): super(ModuleCommand.SubCommand, self).__init__(cli) self.module_base = dnf.module.module_base.ModuleBase(self.base) def _get_modules_from_name_stream_specs(self): modules_from_specs = set() for module_spec in self.opts.module_spec: __, nsvcap = self.module_base._get_modules(module_spec) # When there is no match, the problem was already reported by module_base.remove() if nsvcap is None: continue name = nsvcap.name if nsvcap.name else "" stream = nsvcap.stream if nsvcap.stream else "" if (nsvcap.version and nsvcap.version != -1) or nsvcap.context: logger.info(_("Only module name, stream, architecture or profile is used. " "Ignoring unneeded information in argument: '{}'").format( module_spec)) arch = nsvcap.arch if nsvcap.arch else "" modules = self.base._moduleContainer.query(name, stream, "", "", arch) modules_from_specs.update(modules) return modules_from_specs def _get_module_artifact_names(self, use_modules, skip_modules): artifacts = set() pkg_names = set() for module in use_modules: if module not in skip_modules: if self.base._moduleContainer.isModuleActive(module): artifacts.update(module.getArtifacts()) for artifact in artifacts: subj = hawkey.Subject(artifact) for nevra_obj in subj.get_nevra_possibilities( forms=[hawkey.FORM_NEVRA]): if nevra_obj.name: pkg_names.add(nevra_obj.name) return pkg_names, artifacts class ListSubCommand(SubCommand): aliases = ('list',) summary = _('list all module streams, profiles and states') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True def run_on_module(self): mods = self.module_base if self.opts.enabled: output = mods._get_brief_description( self.opts.module_spec, libdnf.module.ModulePackageContainer.ModuleState_ENABLED) elif self.opts.disabled: output = mods._get_brief_description( self.opts.module_spec, libdnf.module.ModulePackageContainer.ModuleState_DISABLED) elif self.opts.installed: output = mods._get_brief_description( self.opts.module_spec, libdnf.module.ModulePackageContainer.ModuleState_INSTALLED) else: output = mods._get_brief_description( self.opts.module_spec, libdnf.module.ModulePackageContainer.ModuleState_UNKNOWN) if output: print(output) return if self.opts.module_spec: msg = _('No matching Modules to list') raise dnf.exceptions.Error(msg) class InfoSubCommand(SubCommand): aliases = ('info',) summary = _('print detailed information about a module') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True def run_on_module(self): if self.opts.verbose: output = self.module_base._get_full_info(self.opts.module_spec) elif self.opts.profile: output = self.module_base._get_info_profiles(self.opts.module_spec) else: output = self.module_base._get_info(self.opts.module_spec) if output: print(output) else: raise dnf.exceptions.Error(_('No matching Modules to list')) class EnableSubCommand(SubCommand): aliases = ('enable',) summary = _('enable a module stream') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True demands.resolving = True demands.root_user = True def run_on_module(self): try: self.module_base.enable(self.opts.module_spec) except dnf.exceptions.MarkingErrors as e: if self.base.conf.strict: if e.no_match_group_specs or e.error_group_specs: raise e if e.module_depsolv_errors and e.module_depsolv_errors[1] != \ libdnf.module.ModulePackageContainer.ModuleErrorType_ERROR_IN_DEFAULTS: raise e logger.error(str(e)) class DisableSubCommand(SubCommand): aliases = ('disable',) summary = _('disable a module with all its streams') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True demands.resolving = True demands.root_user = True def run_on_module(self): try: self.module_base.disable(self.opts.module_spec) except dnf.exceptions.MarkingErrors as e: if self.base.conf.strict: if e.no_match_group_specs or e.error_group_specs: raise e if e.module_depsolv_errors and e.module_depsolv_errors[1] != \ libdnf.module.ModulePackageContainer.ModuleErrorType_ERROR_IN_DEFAULTS: raise e logger.error(str(e)) class ResetSubCommand(SubCommand): aliases = ('reset',) summary = _('reset a module') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True demands.resolving = True demands.root_user = True def run_on_module(self): try: self.module_base.reset(self.opts.module_spec) except dnf.exceptions.MarkingErrors as e: if self.base.conf.strict: if e.no_match_group_specs: raise e logger.error(str(e)) class InstallSubCommand(SubCommand): aliases = ('install',) summary = _('install a module profile including its packages') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True demands.resolving = True demands.root_user = True def run_on_module(self): try: self.module_base.install(self.opts.module_spec, self.base.conf.strict) except dnf.exceptions.MarkingErrors as e: if self.base.conf.strict: if e.no_match_group_specs or e.error_group_specs: raise e logger.error(str(e)) class UpdateSubCommand(SubCommand): aliases = ('update',) summary = _('update packages associated with an active stream') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True demands.resolving = True demands.root_user = True def run_on_module(self): module_specs = self.module_base.upgrade(self.opts.module_spec) if module_specs: raise NoModuleException(", ".join(module_specs)) class RemoveSubCommand(SubCommand): aliases = ('remove', 'erase',) summary = _('remove installed module profiles and their packages') def configure(self): demands = self.cli.demands demands.allow_erasing = True demands.available_repos = True demands.fresh_metadata = False demands.resolving = True demands.root_user = True demands.sack_activation = True def run_on_module(self): skipped_groups = self.module_base.remove(self.opts.module_spec) if self.opts.all: modules_from_specs = self._get_modules_from_name_stream_specs() remove_names_from_spec, __ = self._get_module_artifact_names( modules_from_specs, set()) keep_names, __ = self._get_module_artifact_names( self.base._moduleContainer.getModulePackages(), modules_from_specs) remove_query = self.base.sack.query().installed().filterm( name=remove_names_from_spec) keep_query = self.base.sack.query().installed().filterm(name=keep_names) for pkg in remove_query: if pkg in keep_query: msg = _("Package {} belongs to multiple modules, skipping").format(pkg) logger.info(msg) else: self.base.goal.erase( pkg, clean_deps=self.base.conf.clean_requirements_on_remove) if not skipped_groups: return logger.error(dnf.exceptions.MarkingErrors(no_match_group_specs=skipped_groups)) class SwitchToSubCommand(SubCommand): aliases = ('switch-to',) summary = _('switch a module to a stream and distrosync rpm packages') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True demands.resolving = True demands.root_user = True self.base.conf.module_stream_switch = True def run_on_module(self): try: self.module_base.switch_to(self.opts.module_spec, strict=self.base.conf.strict) except dnf.exceptions.MarkingErrors as e: if self.base.conf.strict: if e.no_match_group_specs or e.error_group_specs: raise e logger.error(str(e)) class ProvidesSubCommand(SubCommand): aliases = ("provides", ) summary = _('list modular packages') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True def run_on_module(self): output = self.module_base._what_provides(self.opts.module_spec) if output: print(output) class RepoquerySubCommand(SubCommand): aliases = ("repoquery", ) summary = _('list packages belonging to a module') def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True def run_on_module(self): modules_from_specs = set() for module_spec in self.opts.module_spec: modules, __ = self.module_base._get_modules(module_spec) modules_from_specs.update(modules) names_from_spec, spec_artifacts = self._get_module_artifact_names( modules_from_specs, set()) package_strings = set() if self.opts.available or not self.opts.installed: query = self.base.sack.query().available().filterm(nevra_strict=spec_artifacts) for pkg in query: package_strings.add(str(pkg)) if self.opts.installed: query = self.base.sack.query().installed().filterm(name=names_from_spec) for pkg in query: package_strings.add(str(pkg)) output = "\n".join(sorted(package_strings)) print(output) SUBCMDS = {ListSubCommand, InfoSubCommand, EnableSubCommand, DisableSubCommand, ResetSubCommand, InstallSubCommand, UpdateSubCommand, RemoveSubCommand, SwitchToSubCommand, ProvidesSubCommand, RepoquerySubCommand} SUBCMDS_NOT_REQUIRED_ARG = {ListSubCommand} aliases = ("module",) summary = _("Interact with Modules.") def __init__(self, cli): super(ModuleCommand, self).__init__(cli) subcmd_objs = (subcmd(cli) for subcmd in self.SUBCMDS) self.subcmd = None self._subcmd_name2obj = { alias: subcmd for subcmd in subcmd_objs for alias in subcmd.aliases} def set_argparser(self, parser): narrows = parser.add_mutually_exclusive_group() narrows.add_argument('--enabled', dest='enabled', action='store_true', help=_("show only enabled modules")) narrows.add_argument('--disabled', dest='disabled', action='store_true', help=_("show only disabled modules")) narrows.add_argument('--installed', dest='installed', action='store_true', help=_("show only installed modules or packages")) narrows.add_argument('--profile', dest='profile', action='store_true', help=_("show profile content")) parser.add_argument('--available', dest='available', action='store_true', help=_("show only available packages")) narrows.add_argument('--all', dest='all', action='store_true', help=_("remove all modular packages")) subcommand_choices = [] subcommand_help = [] for subcmd in sorted(self.SUBCMDS, key=lambda x: x.aliases[0]): subcommand_choices.append(subcmd.aliases[0]) subcommand_help.append('{}: {}'.format(subcmd.aliases[0], subcmd.summary or '')) parser.add_argument('subcmd', nargs=1, choices=subcommand_choices, metavar='<modular command>', help='\n'.join(subcommand_help)) parser.add_argument('module_spec', metavar='module-spec', nargs='*', help=_("Module specification")) def configure(self): try: self.subcmd = self._subcmd_name2obj[self.opts.subcmd[0]] except (CliError, KeyError): self.cli.optparser.print_usage() raise CliError self.subcmd.opts = self.opts self.subcmd.configure() def run(self): self.check_required_argument() self.subcmd.run_on_module() def check_required_argument(self): not_required_argument = [alias for subcmd in self.SUBCMDS_NOT_REQUIRED_ARG for alias in subcmd.aliases] if self.opts.subcmd[0] not in not_required_argument: if not self.opts.module_spec: raise CliError( _("{} {} {}: too few arguments").format(dnf.util.MAIN_PROG, self.opts.command, self.opts.subcmd[0]))	16.46 KB	2022-09-09 11:21
📄 reinstall.py # reinstall.py # Reinstall CLI command. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.cli import commands from dnf.cli.option_parser import OptionParser from dnf.i18n import _ import dnf.exceptions import logging logger = logging.getLogger('dnf') class ReinstallCommand(commands.Command): """A class containing methods needed by the cli to execute the reinstall command. """ aliases = ('reinstall', 'rei') summary = _('reinstall a package') @staticmethod def set_argparser(parser): parser.add_argument('packages', nargs='+', help=_('Package to reinstall'), action=OptionParser.ParseSpecGroupFileCallback, metavar=_('PACKAGE')) def configure(self): """Verify that conditions are met so that this command can run. These include that the program is being run by the root user, that there are enabled repositories with gpg keys, and that this command is called with appropriate arguments. """ demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True commands._checkGPGKey(self.base, self.cli) if not self.opts.filenames: commands._checkEnabledRepo(self.base) def run(self): # Reinstall files. done = False for pkg in self.base.add_remote_rpms(self.opts.filenames, strict=False, progress=self.base.output.progress): try: self.base.package_reinstall(pkg) except dnf.exceptions.MarkingError: logger.info(_('No match for argument: %s'), self.base.output.term.bold(pkg.location)) else: done = True # Reinstall packages. for pkg_spec in self.opts.pkg_specs + ['@' + x for x in self.opts.grp_specs]: try: self.base.reinstall(pkg_spec) except dnf.exceptions.PackagesNotInstalledError as err: for pkg in err.packages: logger.info(_('Package %s available, but not installed.'), self.output.term.bold(pkg.name)) break logger.info(_('No match for argument: %s'), self.base.output.term.bold(pkg_spec)) except dnf.exceptions.PackagesNotAvailableError as err: for pkg in err.packages: xmsg = '' pkgrepo = self.base.history.repo(pkg) if pkgrepo: xmsg = _(' (from %s)') % pkgrepo msg = _('Installed package %s%s not available.') logger.info(msg, self.base.output.term.bold(pkg), xmsg) except dnf.exceptions.MarkingError: assert False, 'Only the above marking errors are expected.' else: done = True if not done: raise dnf.exceptions.Error(_('No packages marked for reinstall.'))	4.09 KB	2022-09-09 11:21
📄 remove.py # remove_command.py # Remove CLI command. # # Copyright (C) 2012-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.cli import commands from dnf.i18n import _ from dnf.cli.option_parser import OptionParser import dnf.base import argparse import hawkey import dnf.exceptions import logging logger = logging.getLogger("dnf") class RemoveCommand(commands.Command): """Remove command.""" nevra_forms = {'remove-n': hawkey.FORM_NAME, 'remove-na': hawkey.FORM_NA, 'remove-nevra': hawkey.FORM_NEVRA, 'erase-n': hawkey.FORM_NAME, 'erase-na': hawkey.FORM_NA, 'erase-nevra': hawkey.FORM_NEVRA} aliases = ('remove', 'erase', 'rm') + tuple(nevra_forms.keys()) summary = _('remove a package or packages from your system') @staticmethod def set_argparser(parser): mgroup = parser.add_mutually_exclusive_group() mgroup.add_argument('--duplicates', action='store_true', dest='duplicated', help=_('remove duplicated packages')) mgroup.add_argument('--duplicated', action='store_true', help=argparse.SUPPRESS) mgroup.add_argument('--oldinstallonly', action='store_true', help=_( 'remove installonly packages over the limit')) parser.add_argument('packages', nargs='*', help=_('Package to remove'), action=OptionParser.ParseSpecGroupFileCallback, metavar=_('PACKAGE')) def configure(self): demands = self.cli.demands # disable all available repos to delete whole dependency tree # instead of replacing removable package with available packages demands.resolving = True demands.root_user = True demands.sack_activation = True if self.opts.duplicated: demands.available_repos = True elif dnf.base.WITH_MODULES and self.opts.grp_specs: demands.available_repos = True demands.fresh_metadata = False demands.allow_erasing = True else: demands.allow_erasing = True demands.available_repos = False def run(self): forms = [] if self.opts.command in self.nevra_forms: forms = [self.nevra_forms[self.opts.command]] # local pkgs not supported in erase command self.opts.pkg_specs += self.opts.filenames done = False if self.opts.duplicated: q = self.base.sack.query() instonly = self.base._get_installonly_query(q.installed()) dups = q.duplicated().difference(instonly) if not dups: logger.info(_('No duplicated packages found for removal.')) return for (name, arch), pkgs_list in dups._na_dict().items(): if len(pkgs_list) < 2: continue pkgs_list.sort(reverse=True) try: self.base.reinstall(str(pkgs_list[0])) except dnf.exceptions.PackagesNotAvailableError: xmsg = '' msg = _('Installed package %s%s not available.') logger.warning(msg, self.base.output.term.bold(str(pkgs_list[0])), xmsg) for pkg in pkgs_list[1:]: self.base.package_remove(pkg) return if self.opts.oldinstallonly: q = self.base.sack.query() instonly = self.base._get_installonly_query(q.installed()).latest(-1) # also remove running kernel from the set kernel = self.base.sack.get_running_kernel() if kernel is not None: running_installonly = instonly.filter( epoch=kernel.epoch, version=kernel.version, release=kernel.release) if running_installonly: instonly = instonly.difference(running_installonly) if instonly: for pkg in instonly: self.base.package_remove(pkg) else: logger.info(_('No old installonly packages found for removal.')) return # Remove groups. if self.opts.grp_specs and forms: for grp_spec in self.opts.grp_specs: msg = _('Not a valid form: %s') logger.warning(msg, self.base.output.term.bold(grp_spec)) elif self.opts.grp_specs: if dnf.base.WITH_MODULES: module_base = dnf.module.module_base.ModuleBase(self.base) skipped_grps = module_base.remove(self.opts.grp_specs) if len(self.opts.grp_specs) != len(skipped_grps): done = True else: skipped_grps = self.opts.grp_specs if skipped_grps: for group in skipped_grps: try: if self.base.env_group_remove([group]): done = True except dnf.exceptions.Error: pass for pkg_spec in self.opts.pkg_specs: try: self.base.remove(pkg_spec, forms=forms) except dnf.exceptions.MarkingError as e: msg = '{}: {}'.format(e.value, self.base.output.term.bold(pkg_spec)) logger.info(msg) else: done = True if not done: logger.warning(_('No packages marked for removal.'))	6.45 KB	2025-09-22 11:27
📄 repolist.py # repolist.py # repolist CLI command. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.cli import commands from dnf.i18n import _, ucd, fill_exact_width, exact_width from dnf.cli.option_parser import OptionParser import dnf.cli.format import dnf.pycomp import dnf.util import fnmatch import hawkey import logging import operator logger = logging.getLogger('dnf') def _expire_str(repo, md): last = dnf.util.normalize_time(repo._repo.getTimestamp()) if md else _("unknown") if repo.metadata_expire <= -1: return _("Never (last: %s)") % last elif not repo.metadata_expire: return _("Instant (last: %s)") % last else: num = _num2ui_num(repo.metadata_expire) return _("%s second(s) (last: %s)") % (num, last) def _num2ui_num(num): return ucd(dnf.pycomp.format("%d", num, True)) def _repo_match(repo, patterns): rid = repo.id.lower() rnm = repo.name.lower() for pat in patterns: if fnmatch.fnmatch(rid, pat): return True if fnmatch.fnmatch(rnm, pat): return True return False def _repo_size(sack, repo): ret = 0 for pkg in sack.query(flags=hawkey.IGNORE_EXCLUDES).filterm(reponame__eq=repo.id): ret += pkg._size return dnf.cli.format.format_number(ret) class RepoListCommand(commands.Command): """A class containing methods needed by the cli to execute the repolist command. """ aliases = ('repolist', 'repoinfo') summary = _('display the configured software repositories') @staticmethod def set_argparser(parser): repolimit = parser.add_mutually_exclusive_group() repolimit.add_argument('--all', dest='_repos_action', action='store_const', const='all', default=None, help=_("show all repos")) repolimit.add_argument('--enabled', dest='_repos_action', action='store_const', const='enabled', help=_("show enabled repos (default)")) repolimit.add_argument('--disabled', dest='_repos_action', action='store_const', const='disabled', help=_("show disabled repos")) parser.add_argument('repos', nargs='*', default='enabled-default', metavar="REPOSITORY", choices=['all', 'enabled', 'disabled'], action=OptionParser.PkgNarrowCallback, help=_("Repository specification")) def pre_configure(self): if not self.opts.quiet: self.cli.redirect_logger(stdout=logging.WARNING, stderr=logging.INFO) def configure(self): if not self.opts.quiet: self.cli.redirect_repo_progress() demands = self.cli.demands if self.base.conf.verbose or self.opts.command == 'repoinfo': demands.available_repos = True demands.sack_activation = True if self.opts._repos_action: self.opts.repos_action = self.opts._repos_action def run(self): arg = self.opts.repos_action extcmds = [x.lower() for x in self.opts.repos] verbose = self.base.conf.verbose repos = list(self.base.repos.values()) repos.sort(key=operator.attrgetter('id')) term = self.output.term on_ehibeg = term.FG_COLOR['green'] + term.MODE['bold'] on_dhibeg = term.FG_COLOR['red'] on_hiend = term.MODE['normal'] tot_num = 0 cols = [] if not repos: logger.warning(_('No repositories available')) return include_status = arg == 'all' or (arg == 'enabled-default' and extcmds) repoinfo_output = [] for repo in repos: if len(extcmds) and not _repo_match(repo, extcmds): continue (ehibeg, dhibeg, hiend) = '', '', '' ui_enabled = '' ui_endis_wid = 0 ui_excludes_num = '' if include_status: (ehibeg, dhibeg, hiend) = (on_ehibeg, on_dhibeg, on_hiend) if repo.enabled: enabled = True if arg == 'disabled': continue if include_status or verbose or self.opts.command == 'repoinfo': ui_enabled = ehibeg + _('enabled') + hiend ui_endis_wid = exact_width(_('enabled')) if verbose or self.opts.command == 'repoinfo': ui_size = _repo_size(self.base.sack, repo) else: enabled = False if arg == 'enabled' or (arg == 'enabled-default' and not extcmds): continue ui_enabled = dhibeg + _('disabled') + hiend ui_endis_wid = exact_width(_('disabled')) if not (verbose or self.opts.command == 'repoinfo'): rid = ucd(repo.id) cols.append((rid, repo.name, (ui_enabled, ui_endis_wid))) else: if enabled: md = repo.metadata else: md = None out = [self.output.fmtKeyValFill(_("Repo-id : "), repo.id), self.output.fmtKeyValFill(_("Repo-name : "), repo.name)] if include_status: out += [self.output.fmtKeyValFill(_("Repo-status : "), ui_enabled)] if md and repo._repo.getRevision(): out += [self.output.fmtKeyValFill(_("Repo-revision : "), repo._repo.getRevision())] if md and repo._repo.getContentTags(): tags = repo._repo.getContentTags() out += [self.output.fmtKeyValFill(_("Repo-tags : "), ", ".join(sorted(tags)))] if md and repo._repo.getDistroTags(): distroTagsDict = {k: v for (k, v) in repo._repo.getDistroTags()} for (distro, tags) in distroTagsDict.items(): out += [self.output.fmtKeyValFill( _("Repo-distro-tags : "), "[%s]: %s" % (distro, ", ".join(sorted(tags))))] if md: num = len(self.base.sack.query(flags=hawkey.IGNORE_EXCLUDES).filterm( reponame__eq=repo.id)) num_available = len(self.base.sack.query().filterm(reponame__eq=repo.id)) ui_num = _num2ui_num(num) ui_num_available = _num2ui_num(num_available) tot_num += num out += [ self.output.fmtKeyValFill( _("Repo-updated : "), dnf.util.normalize_time(repo._repo.getMaxTimestamp())), self.output.fmtKeyValFill(_("Repo-pkgs : "), ui_num), self.output.fmtKeyValFill(_("Repo-available-pkgs: "), ui_num_available), self.output.fmtKeyValFill(_("Repo-size : "), ui_size)] if repo.metalink: out += [self.output.fmtKeyValFill(_("Repo-metalink : "), repo.metalink)] if enabled: ts = repo._repo.getTimestamp() out += [self.output.fmtKeyValFill( _(" Updated : "), dnf.util.normalize_time(ts))] elif repo.mirrorlist: out += [self.output.fmtKeyValFill(_("Repo-mirrors : "), repo.mirrorlist)] baseurls = repo.baseurl if baseurls: out += [self.output.fmtKeyValFill(_("Repo-baseurl : "), ", ".join(baseurls))] elif enabled: mirrors = repo._repo.getMirrors() if mirrors: url = "%s (%d more)" % (mirrors[0], len(mirrors) - 1) out += [self.output.fmtKeyValFill(_("Repo-baseurl : "), url)] expire = _expire_str(repo, md) out += [self.output.fmtKeyValFill(_("Repo-expire : "), expire)] if repo.excludepkgs: # TRANSLATORS: Packages that are excluded - their names like (dnf systemd) out += [self.output.fmtKeyValFill(_("Repo-exclude : "), ", ".join(repo.excludepkgs))] if repo.includepkgs: out += [self.output.fmtKeyValFill(_("Repo-include : "), ", ".join(repo.includepkgs))] if ui_excludes_num: # TRANSLATORS: Number of packages that where excluded (5) out += [self.output.fmtKeyValFill(_("Repo-excluded : "), ui_excludes_num)] if repo.repofile: out += [self.output.fmtKeyValFill(_("Repo-filename : "), repo.repofile)] repoinfo_output.append("\n".join(map(ucd, out))) if repoinfo_output: print("\n\n".join(repoinfo_output)) if not verbose and cols: # Work out the first (id) and last (enabled/disabled/count), # then chop the middle (name)... id_len = exact_width(_('repo id')) nm_len = 0 st_len = 0 for (rid, rname, (ui_enabled, ui_endis_wid)) in cols: if id_len < exact_width(rid): id_len = exact_width(rid) if nm_len < exact_width(rname): nm_len = exact_width(rname) if st_len < ui_endis_wid: st_len = ui_endis_wid # Need this as well as above for: fill_exact_width() if include_status: if exact_width(_('status')) > st_len: left = term.columns - (id_len + len(_('status')) + 2) else: left = term.columns - (id_len + st_len + 2) else: # Don't output a status column. left = term.columns - (id_len + 1) if left < nm_len: # Name gets chopped nm_len = left else: # Share the extra... left -= nm_len id_len += left // 2 nm_len += left - (left // 2) txt_rid = fill_exact_width(_('repo id'), id_len) if include_status: txt_rnam = fill_exact_width(_('repo name'), nm_len, nm_len) else: txt_rnam = _('repo name') if not include_status: # Don't output a status column. print("%s %s" % (txt_rid, txt_rnam)) else: print("%s %s %s" % (txt_rid, txt_rnam, _('status'))) for (rid, rname, (ui_enabled, ui_endis_wid)) in cols: if not include_status: # Don't output a status column. print("%s %s" % (fill_exact_width(rid, id_len), rname)) continue print("%s %s %s" % (fill_exact_width(rid, id_len), fill_exact_width(rname, nm_len, nm_len), ui_enabled)) if verbose or self.opts.command == 'repoinfo': msg = _('Total packages: {}') print(msg.format(_num2ui_num(tot_num)))	12.62 KB	2022-09-09 11:21
📄 repoquery.py # # Copyright (C) 2014 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import print_function from __future__ import unicode_literals from dnf.i18n import _ from dnf.cli import commands from dnf.cli.option_parser import OptionParser import argparse import datetime import logging import re import sys import dnf import dnf.cli import dnf.exceptions import dnf.subject import dnf.util import hawkey logger = logging.getLogger('dnf') QFORMAT_DEFAULT = '%{name}-%{epoch}:%{version}-%{release}.%{arch}' # matches %[-][dd]{attr} QFORMAT_MATCH = re.compile(r'%(-?\d?){([:.\w]+?)}') QUERY_TAGS = """\ name, arch, epoch, version, release, reponame (repoid), from_repo, evr, debug_name, source_name, source_debug_name, installtime, buildtime, size, downloadsize, installsize, provides, requires, obsoletes, conflicts, sourcerpm, description, summary, license, url, reason""" OPTS_MAPPING = { 'conflicts': 'conflicts', 'enhances': 'enhances', 'obsoletes': 'obsoletes', 'provides': 'provides', 'recommends': 'recommends', 'requires': 'requires', 'requires-pre': 'requires_pre', 'suggests': 'suggests', 'supplements': 'supplements' } def rpm2py_format(queryformat): """Convert a rpm like QUERYFMT to an python .format() string.""" def fmt_repl(matchobj): fill = matchobj.groups()[0] key = matchobj.groups()[1] if fill: if fill[0] == '-': fill = '>' + fill[1:] else: fill = '<' + fill fill = ':' + fill return '{0.' + key.lower() + fill + "}" def brackets(txt): return txt.replace('{', '{{').replace('}', '}}') queryformat = queryformat.replace("\\n", "\n").replace("\\t", "\t") for key, value in OPTS_MAPPING.items(): queryformat = queryformat.replace(key, value) fmt = "" spos = 0 for item in QFORMAT_MATCH.finditer(queryformat): fmt += brackets(queryformat[spos:item.start()]) fmt += fmt_repl(item) spos = item.end() fmt += brackets(queryformat[spos:]) return fmt class _CommaSplitCallback(OptionParser._SplitCallback): SPLITTER = r'\s,\s' class RepoQueryCommand(commands.Command): """A class containing methods needed by the cli to execute the repoquery command. """ nevra_forms = {'repoquery-n': hawkey.FORM_NAME, 'repoquery-na': hawkey.FORM_NA, 'repoquery-nevra': hawkey.FORM_NEVRA} aliases = ('repoquery', 'rq') + tuple(nevra_forms.keys()) summary = _('search for packages matching keyword') @staticmethod def filter_repo_arch(opts, query): """Filter query by repoid and arch options""" if opts.repo: query.filterm(reponame=opts.repo) if opts.arches: query.filterm(arch=opts.arches) return query @staticmethod def set_argparser(parser): parser.add_argument('-a', '--all', dest='queryall', action='store_true', help=_("Query all packages (shorthand for repoquery '' " "or repoquery without argument)")) parser.add_argument('--show-duplicates', action='store_true', help=_("Query all versions of packages (default)")) parser.add_argument('--arch', '--archlist', dest='arches', default=[], action=_CommaSplitCallback, metavar='ARCH', help=_('show only results from this ARCH')) parser.add_argument('-f', '--file', metavar='FILE', nargs='+', help=_('show only results that owns FILE')) parser.add_argument('--whatconflicts', default=[], action=_CommaSplitCallback, metavar='REQ', help=_('show only results that conflict REQ')) parser.add_argument('--whatdepends', default=[], action=_CommaSplitCallback, metavar='REQ', help=_('shows results that requires, suggests, supplements, enhances, ' 'or recommends package provides and files REQ')) parser.add_argument('--whatobsoletes', default=[], action=_CommaSplitCallback, metavar='REQ', help=_('show only results that obsolete REQ')) parser.add_argument('--whatprovides', default=[], action=_CommaSplitCallback, metavar='REQ', help=_('show only results that provide REQ')) parser.add_argument('--whatrequires', default=[], action=_CommaSplitCallback, metavar='REQ', help=_('shows results that requires package provides and files REQ')) parser.add_argument('--whatrecommends', default=[], action=_CommaSplitCallback, metavar='REQ', help=_('show only results that recommend REQ')) parser.add_argument('--whatenhances', default=[], action=_CommaSplitCallback, metavar='REQ', help=_('show only results that enhance REQ')) parser.add_argument('--whatsuggests', default=[], action=_CommaSplitCallback, metavar='REQ', help=_('show only results that suggest REQ')) parser.add_argument('--whatsupplements', default=[], action=_CommaSplitCallback, metavar='REQ', help=_('show only results that supplement REQ')) whatrequiresform = parser.add_mutually_exclusive_group() whatrequiresform.add_argument("--alldeps", action="store_true", help=_("check non-explicit dependencies (files and Provides); default")) whatrequiresform.add_argument("--exactdeps", action="store_true", help=_('check dependencies exactly as given, opposite of --alldeps')) parser.add_argument("--recursive", action="store_true", help=_( 'used with --whatrequires, and --requires --resolve, query packages recursively.')) parser.add_argument('--deplist', action='store_true', help=_( "show a list of all dependencies and what packages provide them")) parser.add_argument('--resolve', action='store_true', help=_('resolve capabilities to originating package(s)')) parser.add_argument("--tree", action="store_true", help=_('show recursive tree for package(s)')) parser.add_argument('--srpm', action='store_true', help=_('operate on corresponding source RPM')) parser.add_argument("--latest-limit", dest='latest_limit', type=int, help=_('show N latest packages for a given name.arch' ' (or latest but N if N is negative)')) parser.add_argument("--disable-modular-filtering", action="store_true", help=_("list also packages of inactive module streams")) outform = parser.add_mutually_exclusive_group() outform.add_argument('-i', "--info", dest='queryinfo', default=False, action='store_true', help=_('show detailed information about the package')) outform.add_argument('-l', "--list", dest='queryfilelist', default=False, action='store_true', help=_('show list of files in the package')) outform.add_argument('-s', "--source", dest='querysourcerpm', default=False, action='store_true', help=_('show package source RPM name')) outform.add_argument('--changelogs', dest='querychangelogs', default=False, action='store_true', help=_('show changelogs of the package')) outform.add_argument('--qf', "--queryformat", dest='queryformat', default=QFORMAT_DEFAULT, help=_('display format for listing packages: ' '"%%{name} %%{version} ...", ' 'use --querytags to view full tag list')) parser.add_argument('--querytags', action='store_true', help=_('show available tags to use with ' '--queryformat')) outform.add_argument("--nevra", dest='queryformat', const=QFORMAT_DEFAULT, action='store_const', help=_('use name-epoch:version-release.architecture format for ' 'displaying found packages (default)')) outform.add_argument("--nvr", dest='queryformat', const='%{name}-%{version}-%{release}', action='store_const', help=_('use name-version-release format for ' 'displaying found packages ' '(rpm query default)')) outform.add_argument("--envra", dest='queryformat', const='%{epoch}:%{name}-%{version}-%{release}.%{arch}', action='store_const', help=_('use epoch:name-version-release.architecture format for ' 'displaying found packages')) outform.add_argument('--groupmember', action="store_true", help=_( 'Display in which comps groups are presented selected packages')) pkgfilter = parser.add_mutually_exclusive_group() pkgfilter.add_argument("--duplicates", dest='pkgfilter', const='duplicated', action='store_const', help=_('limit the query to installed duplicate ' 'packages')) pkgfilter.add_argument("--duplicated", dest='pkgfilter', const='duplicated', action='store_const', help=argparse.SUPPRESS) pkgfilter.add_argument("--installonly", dest='pkgfilter', const='installonly', action='store_const', help=_('limit the query to installed installonly packages')) pkgfilter.add_argument("--unsatisfied", dest='pkgfilter', const='unsatisfied', action='store_const', help=_('limit the query to installed packages with unsatisfied dependencies')) parser.add_argument('--location', action='store_true', help=_('show a location from where packages can be downloaded')) package_attribute = parser.add_mutually_exclusive_group() help_msgs = { 'conflicts': _('Display capabilities that the package conflicts with.'), 'depends': _('Display capabilities that the package can depend on, enhance, recommend,' ' suggest, and supplement.'), 'enhances': _('Display capabilities that the package can enhance.'), 'provides': _('Display capabilities provided by the package.'), 'recommends': _('Display capabilities that the package recommends.'), 'requires': _('Display capabilities that the package depends on.'), 'requires-pre': _('If the package is not installed display capabilities that it depends on for ' 'running %%pre and %%post scriptlets. If the package is installed display ' 'capabilities that is depends for %%pre, %%post, %%preun and %%postun.'), 'suggests': _('Display capabilities that the package suggests.'), 'supplements': _('Display capabilities that the package can supplement.') } for arg, help_msg in help_msgs.items(): name = '--%s' % arg package_attribute.add_argument(name, dest='packageatr', action='store_const', const=arg, help=help_msg) parser.add_argument('--available', action="store_true", help=_('Display only available packages.')) help_list = { 'installed': _('Display only installed packages.'), 'extras': _('Display only packages that are not present in any of available repositories.'), 'upgrades': _('Display only packages that provide an upgrade for some already installed package.'), 'unneeded': _('Display only packages that can be removed by "{prog} autoremove" ' 'command.').format(prog=dnf.util.MAIN_PROG), 'userinstalled': _('Display only packages that were installed by user.') } list_group = parser.add_mutually_exclusive_group() for list_arg, help_arg in help_list.items(): switch = '--%s' % list_arg list_group.add_argument(switch, dest='list', action='store_const', const=list_arg, help=help_arg) # make --autoremove hidden compatibility alias for --unneeded list_group.add_argument( '--autoremove', dest='list', action='store_const', const="unneeded", help=argparse.SUPPRESS) parser.add_argument('--recent', action="store_true", help=_('Display only recently edited packages')) parser.add_argument('key', nargs='', metavar="KEY", help=_('the key to search for')) def pre_configure(self): if not self.opts.quiet: self.cli.redirect_logger(stdout=logging.WARNING, stderr=logging.INFO) def configure(self): if not self.opts.quiet: self.cli.redirect_repo_progress() demands = self.cli.demands if self.opts.obsoletes: if self.opts.packageatr: self.cli._option_conflict("--obsoletes", "--" + self.opts.packageatr) else: self.opts.packageatr = "obsoletes" if self.opts.querytags: return if self.opts.resolve and not self.opts.packageatr: raise dnf.cli.CliError( _("Option '--resolve' has to be used together with one of the " "'--conflicts', '--depends', '--enhances', '--provides', '--recommends', " "'--requires', '--requires-pre', '--suggests' or '--supplements' options")) if self.opts.recursive: if self.opts.exactdeps: self.cli._option_conflict("--recursive", "--exactdeps") if not any([self.opts.whatrequires, (self.opts.packageatr == "requires" and self.opts.resolve)]): raise dnf.cli.CliError( _("Option '--recursive' has to be used with '--whatrequires <REQ>' " "(optionally with '--alldeps', but not with '--exactdeps'), or with " "'--requires <REQ> --resolve'")) if self.opts.alldeps or self.opts.exactdeps: if not (self.opts.whatrequires or self.opts.whatdepends): raise dnf.cli.CliError( _("argument {} requires --whatrequires or --whatdepends option".format( '--alldeps' if self.opts.alldeps else '--exactdeps'))) if self.opts.srpm: self.base.repos.enable_source_repos() if (self.opts.list not in ["installed", "userinstalled"] and self.opts.pkgfilter != "installonly") or self.opts.available: demands.available_repos = True demands.sack_activation = True if self.opts.querychangelogs: demands.changelogs = True def build_format_fn(self, opts, pkg): if opts.querychangelogs: out = [] out.append('Changelog for %s' % str(pkg)) for chlog in pkg.changelogs: dt = chlog['timestamp'] out.append(' %s %s\n%s\n' % (dt.strftime("%a %b %d %Y"), dnf.i18n.ucd(chlog['author']), dnf.i18n.ucd(chlog['text']))) return '\n'.join(out) try: po = PackageWrapper(pkg) if opts.queryinfo: return self.base.output.infoOutput(pkg) elif opts.queryfilelist: filelist = po.files if not filelist: print(_('Package {} contains no files').format(pkg), file=sys.stderr) return filelist elif opts.querysourcerpm: return po.sourcerpm else: return rpm2py_format(opts.queryformat).format(po) except AttributeError as e: # catch that the user has specified attributes # there don't exist on the dnf Package object. raise dnf.exceptions.Error(str(e)) def _resolve_nevras(self, nevras, base_query): resolved_nevras_query = self.base.sack.query().filterm(empty=True) for nevra in nevras: resolved_nevras_query = resolved_nevras_query.union(base_query.intersection( dnf.subject.Subject(nevra).get_best_query( self.base.sack, with_provides=False, with_filenames=False ) )) return resolved_nevras_query def _do_recursive_deps(self, query_in, query_select, done=None): done = done if done else query_select query_required = query_in.filter(requires=query_select) query_select = query_required.difference(done) done = query_required.union(done) if query_select: done = self._do_recursive_deps(query_in, query_select, done=done) return done def by_all_deps(self, names, query, all_dep_types=False): # in case of arguments being NEVRAs, resolve them to packages resolved_nevras_query = self._resolve_nevras(names, query) # filter the arguments directly as reldeps depquery = query.filter(requires__glob=names) # filter the resolved NEVRAs as packages depquery = depquery.union(query.filter(requires=resolved_nevras_query)) if all_dep_types: # TODO this is very inefficient, as it resolves the `names` glob to # reldeps four more times, which in a reasonably wide glob like # `dnf repoquery --whatdepends "libdnf"` can take roughly 50% of # the total execution time. depquery = depquery.union(query.filter(recommends__glob=names)) depquery = depquery.union(query.filter(enhances__glob=names)) depquery = depquery.union(query.filter(supplements__glob=names)) depquery = depquery.union(query.filter(suggests__glob=names)) depquery = depquery.union(query.filter(recommends=resolved_nevras_query)) depquery = depquery.union(query.filter(enhances=resolved_nevras_query)) depquery = depquery.union(query.filter(supplements=resolved_nevras_query)) depquery = depquery.union(query.filter(suggests=resolved_nevras_query)) if self.opts.recursive: depquery = self._do_recursive_deps(query, depquery) return depquery def _get_recursive_providers_query(self, query_in, providers, done=None): done = done if done else self.base.sack.query().filterm(empty=True) t = self.base.sack.query().filterm(empty=True) for pkg in providers.run(): t = t.union(query_in.filter(provides=pkg.requires)) query_select = t.difference(done) if query_select: done = self._get_recursive_providers_query(query_in, query_select, done=t.union(done)) return t.union(done) def _add_add_remote_packages(self): rpmnames = [] remote_packages = [] for key in self.opts.key: schemes = dnf.pycomp.urlparse.urlparse(key)[0] if key.endswith('.rpm'): rpmnames.append(key) elif schemes and schemes in ('http', 'ftp', 'file', 'https'): rpmnames.append(key) if rpmnames: remote_packages = self.base.add_remote_rpms( rpmnames, strict=False, progress=self.base.output.progress) return remote_packages def run(self): if self.opts.querytags: print(QUERY_TAGS) return self.cli._populate_update_security_filter(self.opts) q = self.base.sack.query( flags=hawkey.IGNORE_MODULAR_EXCLUDES if self.opts.disable_modular_filtering else hawkey.APPLY_EXCLUDES ) if self.opts.key: remote_packages = self._add_add_remote_packages() kwark = {'with_provides': False} if self.opts.command in self.nevra_forms: kwark["forms"] = [self.nevra_forms[self.opts.command]] kwark['with_filenames'] = False pkgs = [] query_results = q.filter(empty=True) if remote_packages: query_results = query_results.union( self.base.sack.query().filterm(pkg=remote_packages)) for key in self.opts.key: query_results = query_results.union( dnf.subject.Subject(key, ignore_case=True).get_best_query( self.base.sack, query=q, *kwark)) q = query_results if self.opts.recent: q = q._recent(self.base.conf.recent) if self.opts.available: if self.opts.list and self.opts.list != "installed": print(self.cli.optparser.print_usage()) raise dnf.exceptions.Error(_("argument {}: not allowed with argument {}".format( "--available", "--" + self.opts.list))) elif self.opts.list == "unneeded": q = q._unneeded(self.base.history.swdb) elif self.opts.list and self.opts.list != 'userinstalled': q = getattr(q, self.opts.list)() if self.opts.pkgfilter == "duplicated": installonly = self.base._get_installonly_query(q) q = q.difference(installonly).duplicated() elif self.opts.pkgfilter == "installonly": q = self.base._get_installonly_query(q) elif self.opts.pkgfilter == "unsatisfied": rpmdb = dnf.sack.rpmdb_sack(self.base) rpmdb._configure(self.base.conf.installonlypkgs, self.base.conf.installonly_limit) goal = dnf.goal.Goal(rpmdb) goal.protect_running_kernel = False solved = goal.run(verify=True) if not solved: print(dnf.util._format_resolve_problems(goal.problem_rules())) return elif not self.opts.list: # do not show packages from @System repo q = q.available() # filter repo and arch q = self.filter_repo_arch(self.opts, q) orquery = q if self.opts.file: q.filterm(file__glob=self.opts.file) if self.opts.whatconflicts: rels = q.filter(conflicts__glob=self.opts.whatconflicts) q = rels.union(q.filter(conflicts=self._resolve_nevras(self.opts.whatconflicts, q))) if self.opts.whatobsoletes: q.filterm(obsoletes=self.opts.whatobsoletes) if self.opts.whatprovides: query_for_provide = q.filter(provides__glob=self.opts.whatprovides) if query_for_provide: q = query_for_provide else: q.filterm(file__glob=self.opts.whatprovides) if self.opts.whatrequires: if (self.opts.exactdeps): q.filterm(requires__glob=self.opts.whatrequires) else: q = self.by_all_deps(self.opts.whatrequires, q) if self.opts.whatdepends: if (self.opts.exactdeps): dependsquery = q.filter(requires__glob=self.opts.whatdepends) dependsquery = dependsquery.union(q.filter(recommends__glob=self.opts.whatdepends)) dependsquery = dependsquery.union(q.filter(enhances__glob=self.opts.whatdepends)) dependsquery = dependsquery.union(q.filter(supplements__glob=self.opts.whatdepends)) q = dependsquery.union(q.filter(suggests__glob=self.opts.whatdepends)) else: q = self.by_all_deps(self.opts.whatdepends, q, True) if self.opts.whatrecommends: rels = q.filter(recommends__glob=self.opts.whatrecommends) q = rels.union(q.filter(recommends=self._resolve_nevras(self.opts.whatrecommends, q))) if self.opts.whatenhances: rels = q.filter(enhances__glob=self.opts.whatenhances) q = rels.union(q.filter(enhances=self._resolve_nevras(self.opts.whatenhances, q))) if self.opts.whatsupplements: rels = q.filter(supplements__glob=self.opts.whatsupplements) q = rels.union(q.filter(supplements=self._resolve_nevras(self.opts.whatsupplements, q))) if self.opts.whatsuggests: rels = q.filter(suggests__glob=self.opts.whatsuggests) q = rels.union(q.filter(suggests=self._resolve_nevras(self.opts.whatsuggests, q))) if self.opts.latest_limit: q = q.latest(self.opts.latest_limit) # reduce a query to security upgrades if they are specified q = self.base._merge_update_filters(q, warning=False) if self.opts.srpm: pkg_list = [] for pkg in q: srcname = pkg.source_name if srcname is not None: tmp_query = self.base.sack.query().filterm(name=srcname, evr=pkg.evr, arch='src') pkg_list += tmp_query.run() q = self.base.sack.query().filterm(pkg=pkg_list) if self.opts.tree: if not self.opts.whatrequires and self.opts.packageatr not in ( 'conflicts', 'enhances', 'obsoletes', 'provides', 'recommends', 'requires', 'suggests', 'supplements'): raise dnf.exceptions.Error( _("No valid switch specified\nusage: {prog} repoquery [--conflicts\|" "--enhances\|--obsoletes\|--provides\|--recommends\|--requires\|" "--suggest\|--supplements\|--whatrequires] [key] [--tree]\n\n" "description:\n For the given packages print a tree of the " "packages.").format(prog=dnf.util.MAIN_PROG)) self.tree_seed(q, orquery, self.opts) return pkgs = set() if self.opts.packageatr: rels = set() for pkg in q.run(): if self.opts.list != 'userinstalled' or self.base.history.user_installed(pkg): if self.opts.packageatr == 'depends': rels.update(pkg.requires + pkg.enhances + pkg.suggests + pkg.supplements + pkg.recommends) else: rels.update(getattr(pkg, OPTS_MAPPING[self.opts.packageatr])) if self.opts.resolve: # find the providing packages and show them if self.opts.list == "installed": query = self.filter_repo_arch(self.opts, self.base.sack.query()) else: query = self.filter_repo_arch(self.opts, self.base.sack.query().available()) providers = query.filter(provides=rels) if self.opts.recursive: providers = providers.union( self._get_recursive_providers_query(query, providers)) pkgs = set() for pkg in providers.latest().run(): pkgs.add(self.build_format_fn(self.opts, pkg)) else: pkgs.update(str(rel) for rel in rels) elif self.opts.location: for pkg in q.run(): location = pkg.remote_location() if location is not None: pkgs.add(location) elif self.opts.deplist: pkgs = [] for pkg in sorted(set(q.run())): if self.opts.list != 'userinstalled' or self.base.history.user_installed(pkg): deplist_output = [] deplist_output.append('package: ' + str(pkg)) for req in sorted([str(req) for req in pkg.requires]): deplist_output.append(' dependency: ' + req) subject = dnf.subject.Subject(req) query = subject.get_best_query(self.base.sack) query = self.filter_repo_arch( self.opts, query.available()) if not self.opts.verbose: query = query.latest() for provider in query.run(): deplist_output.append(' provider: ' + str(provider)) pkgs.append('\n'.join(deplist_output)) if pkgs: print('\n\n'.join(pkgs)) return elif self.opts.groupmember: self._group_member_report(q) return else: for pkg in q.run(): if self.opts.list != 'userinstalled' or self.base.history.user_installed(pkg): pkgs.add(self.build_format_fn(self.opts, pkg)) if pkgs: if self.opts.queryinfo: print("\n\n".join(sorted(pkgs))) else: print("\n".join(sorted(pkgs))) def _group_member_report(self, query): package_conf_dict = {} for group in self.base.comps.groups: package_conf_dict[group.id] = set([pkg.name for pkg in group.packages_iter()]) group_package_dict = {} pkg_not_in_group = [] for pkg in query.run(): group_id_list = [] for group_id, package_name_set in package_conf_dict.items(): if pkg.name in package_name_set: group_id_list.append(group_id) if group_id_list: group_package_dict.setdefault( '$'.join(sorted(group_id_list)), []).append(str(pkg)) else: pkg_not_in_group.append(str(pkg)) output = [] for key, package_list in sorted(group_package_dict.items()): output.append( '\n'.join(sorted(package_list) + sorted([' @' + id for id in key.split('$')]))) output.append('\n'.join(sorted(pkg_not_in_group))) if output: print('\n'.join(output)) def grow_tree(self, level, pkg, opts): pkg_string = self.build_format_fn(opts, pkg) if level == -1: print(pkg_string) return spacing = " " for x in range(0, level): spacing += "\| " requires = [] for requirepkg in pkg.requires: requires.append(str(requirepkg)) reqstr = "[" + str(len(requires)) + ": " + ", ".join(requires) + "]" print(spacing + r"\_ " + pkg_string + " " + reqstr) def tree_seed(self, query, aquery, opts, level=-1, usedpkgs=None): for pkg in sorted(set(query.run()), key=lambda p: p.name): usedpkgs = set() if usedpkgs is None or level == -1 else usedpkgs if pkg.name.startswith("rpmlib") or pkg.name.startswith("solvable"): return self.grow_tree(level, pkg, opts) if pkg not in usedpkgs: usedpkgs.add(pkg) if opts.packageatr: strpkg = getattr(pkg, opts.packageatr) ar = {} for name in set(strpkg): pkgquery = self.base.sack.query().filterm(provides=name) for querypkg in pkgquery: ar[querypkg.name + "." + querypkg.arch] = querypkg pkgquery = self.base.sack.query().filterm(pkg=list(ar.values())) else: pkgquery = self.by_all_deps((pkg.name, ), aquery) if opts.alldeps \ else aquery.filter(requires__glob=pkg.name) self.tree_seed(pkgquery, aquery, opts, level + 1, usedpkgs) class PackageWrapper(object): """Wrapper for dnf.package.Package, so we can control formatting.""" def __init__(self, pkg): self._pkg = pkg def __getattr__(self, attr): atr = getattr(self._pkg, attr) if atr is None: return "(none)" if isinstance(atr, list): return '\n'.join(sorted({dnf.i18n.ucd(reldep) for reldep in atr})) return dnf.i18n.ucd(atr) @staticmethod def _get_timestamp(timestamp): if timestamp > 0: dt = datetime.datetime.utcfromtimestamp(timestamp) return dt.strftime("%Y-%m-%d %H:%M") else: return '' @property def buildtime(self): return self._get_timestamp(self._pkg.buildtime) @property def installtime(self): return self._get_timestamp(self._pkg.installtime)	33.76 KB	2025-09-22 11:27
📄 search.py # search.py # Search CLI command. # # Copyright (C) 2012-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import print_function from __future__ import unicode_literals import collections from dnf.cli import commands from dnf.cli.option_parser import OptionParser from dnf.i18n import ucd, _, C_ import dnf.i18n import dnf.match_counter import dnf.util import hawkey import logging logger = logging.getLogger('dnf') class SearchCommand(commands.Command): """A class containing methods needed by the cli to execute the search command. """ aliases = ('search', 'se') summary = _('search package details for the given string') @staticmethod def set_argparser(parser): parser.add_argument('--all', action='store_true', help=_("search also package description and URL")) parser.add_argument('query_string', nargs='+', metavar=_('KEYWORD'), choices=['all'], default=None, action=OptionParser.PkgNarrowCallback, help=_("Keyword to search for")) def _search(self, args): """Search for simple text tags in a package object.""" TRANS_TBL = collections.OrderedDict(( ('name', C_('long', 'Name')), ('summary', C_('long', 'Summary')), ('description', C_('long', 'Description')), ('url', _('URL')), )) def _translate_attr(attr): try: return TRANS_TBL[attr] except: return attr def _print_section_header(exact_match, attrs, keys): trans_attrs = map(_translate_attr, attrs) # TRANSLATORS: separator used between package attributes (eg. Name & Summary & URL) trans_attrs_str = _(' & ').join(trans_attrs) if exact_match: # TRANSLATORS: %s - translated package attributes, # %%s - found keys (in listed attributes) section_text = _('%s Exactly Matched: %%s') % trans_attrs_str else: # TRANSLATORS: %s - translated package attributes, # %%s - found keys (in listed attributes) section_text = _('%s Matched: %%s') % trans_attrs_str formatted = self.base.output.fmtSection(section_text % ", ".join(keys)) print(ucd(formatted)) counter = dnf.match_counter.MatchCounter() for arg in args: self._search_counted(counter, 'name', arg) self._search_counted(counter, 'summary', arg) if self.opts.all: for arg in args: self._search_counted(counter, 'description', arg) self._search_counted(counter, 'url', arg) else: needles = len(args) pkgs = list(counter.keys()) for pkg in pkgs: if len(counter.matched_needles(pkg)) != needles: del counter[pkg] used_attrs = None matched_needles = None exact_match = False print_section_header = False limit = None if not self.base.conf.showdupesfromrepos: limit = self.base.sack.query().filterm(pkg=counter.keys()).latest() seen = set() for pkg in counter.sorted(reverse=True, limit_to=limit): if not self.base.conf.showdupesfromrepos: if pkg.name + pkg.arch in seen: continue seen.add(pkg.name + pkg.arch) if used_attrs != counter.matched_keys(pkg): used_attrs = counter.matched_keys(pkg) print_section_header = True if matched_needles != counter.matched_needles(pkg): matched_needles = counter.matched_needles(pkg) print_section_header = True if exact_match != (counter.matched_haystacks(pkg) == matched_needles): exact_match = counter.matched_haystacks(pkg) == matched_needles print_section_header = True if print_section_header: _print_section_header(exact_match, used_attrs, matched_needles) print_section_header = False self.base.output.matchcallback(pkg, counter.matched_haystacks(pkg), args) if len(counter) == 0: logger.info(_('No matches found.')) def _search_counted(self, counter, attr, needle): fdict = {'%s__substr' % attr : needle} if dnf.util.is_glob_pattern(needle): fdict = {'%s__glob' % attr : needle} q = self.base.sack.query().filterm(hawkey.ICASE, **fdict) for pkg in q.run(): counter.add(pkg, attr, needle) return counter def pre_configure(self): if not self.opts.quiet: self.cli.redirect_logger(stdout=logging.WARNING, stderr=logging.INFO) def configure(self): if not self.opts.quiet: self.cli.redirect_repo_progress() demands = self.cli.demands demands.available_repos = True demands.fresh_metadata = False demands.sack_activation = True self.opts.all = self.opts.all or self.opts.query_string_action def run(self): logger.debug(_('Searching Packages: ')) return self._search(self.opts.query_string)	6.16 KB	2022-09-09 11:21
📄 shell.py # shell.py # Shell CLI command. # # Copyright (C) 2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from dnf.cli import commands from dnf.i18n import _, ucd import dnf.util import cmd import copy import dnf import logging import shlex import sys logger = logging.getLogger('dnf') # only demands we'd like to override class ShellDemandSheet(object): available_repos = True resolving = True root_user = True sack_activation = True class ShellCommand(commands.Command, cmd.Cmd): aliases = ('shell', 'sh') summary = _('run an interactive {prog} shell').format(prog=dnf.util.MAIN_PROG_UPPER) MAPPING = {'repo': 'repo', 'repository': 'repo', 'exit': 'quit', 'quit': 'quit', 'run': 'ts_run', 'ts': 'transaction', 'transaction': 'transaction', 'config': 'config', 'resolvedep': 'resolve', 'help': 'help' } def __init__(self, cli): commands.Command.__init__(self, cli) cmd.Cmd.__init__(self) self.prompt = '> ' @staticmethod def set_argparser(parser): parser.add_argument('script', nargs='?', metavar=_('SCRIPT'), help=_('Script to run in {prog} shell').format( prog=dnf.util.MAIN_PROG_UPPER)) def configure(self): # append to ShellDemandSheet missing demands from # dnf.cli.demand.DemandSheet with their default values. default_demands = self.cli.demands self.cli.demands = ShellDemandSheet() for attr in dir(default_demands): if attr.startswith('__'): continue try: getattr(self.cli.demands, attr) except AttributeError: setattr(self.cli.demands, attr, getattr(default_demands, attr)) def run(self): if self.opts.script: self._run_script(self.opts.script) else: self.cmdloop() def _clean(self): self.base._finalize_base() self.base._transaction = None self.base.fill_sack() def onecmd(self, line): if not line or line == '\n': return if line == 'EOF': line = 'quit' try: s_line = shlex.split(line) except: self._help() return # reset option parser before each command, keep usage information self.cli.optparser.__init__(reset_usage=False) opts = self.cli.optparser.parse_main_args(s_line) # Disable shell recursion. if opts.command == 'shell': return if opts.command in self.MAPPING: getattr(self, '_' + self.MAPPING[opts.command])(s_line[1::]) else: cmd_cls = self.cli.cli_commands.get(opts.command) if cmd_cls is not None: cmd = cmd_cls(self.cli) try: opts = self.cli.optparser.parse_command_args(cmd, s_line) except SystemExit: # argparse.ArgumentParser prints usage information and executes # sys.exit() on problems with parsing command line arguments return try: cmd.cli.demands = copy.deepcopy(self.cli.demands) cmd.configure() cmd.run() except dnf.exceptions.Error as e: logger.error(_("Error:") + " " + ucd(e)) return else: self._help() def _config(self, args=None): def print_or_set(key, val, conf): if val: setattr(conf, key, val) else: try: print('{}: {}'.format(key, getattr(conf, str(key)))) except: logger.warning(_('Unsupported key value.')) if not args or len(args) > 2: self._help('config') return key = args[0] val = args[1] if len(args) == 2 else None period = key.find('.') if period != -1: repo_name = key[:period] key = key[period+1:] repos = self.base.repos.get_matching(repo_name) for repo in repos: print_or_set(key, val, repo) if not repos: logger.warning(_('Could not find repository: %s'), repo_name) else: print_or_set(key, val, self.base.conf) def _help(self, args=None): """Output help information. :param args: the command to output help information about. If args is an empty, general help will be output. """ arg = args[0] if isinstance(args, list) and len(args) > 0 else args msg = None if arg: if arg == 'config': msg = _("""{} arg [value] arg: debuglevel, errorlevel, obsoletes, gpgcheck, assumeyes, exclude, repo_id.gpgcheck, repo_id.exclude If no value is given it prints the current value. If value is given it sets that value.""").format(arg) elif arg == 'help': msg = _("""{} [command] print help""").format(arg) elif arg in ['repo', 'repository']: msg = _("""{} arg [option] list: lists repositories and their status. option = [all \| id \| glob] enable: enable repositories. option = repository id disable: disable repositories. option = repository id""").format(arg) elif arg == 'resolvedep': msg = _("""{} resolve the transaction set""").format(arg) elif arg in ['transaction', 'ts']: msg = _("""{} arg list: lists the contents of the transaction reset: reset (zero-out) the transaction run: run the transaction""").format(arg) elif arg == 'run': msg = _("""{} run the transaction""").format(arg) elif arg in ['exit', 'quit']: msg = _("""{} exit the shell""").format(arg) if not msg: self.cli.optparser.print_help() msg = _("""Shell specific arguments: config set config options help print help repository (or repo) enable, disable or list repositories resolvedep resolve the transaction set transaction (or ts) list, reset or run the transaction set run resolve and run the transaction set exit (or quit) exit the shell""") print('\n' + msg) def _repo(self, args=None): cmd = args[0] if args else None if cmd in ['list', None]: self.onecmd('repolist ' + ' '.join(args[1:])) elif cmd in ['enable', 'disable']: repos = self.cli.base.repos fill_sack = False for repo in args[1::]: r = repos.get_matching(repo) if r: getattr(r, cmd)() fill_sack = True else: logger.critical(_("Error:") + " " + _("Unknown repo: '%s'"), self.base.output.term.bold(repo)) if fill_sack: self.base.fill_sack() # reset base._comps, as it has changed due to changing the repos self.base._comps = None else: self._help('repo') def _resolve(self, args=None): try: self.cli.base.resolve(self.cli.demands.allow_erasing) except dnf.exceptions.DepsolveError as e: print(e) def _run_script(self, file): try: with open(file, 'r') as fd: lines = fd.readlines() for line in lines: if not line.startswith('#'): self.onecmd(line) except IOError: logger.info(_('Error: Cannot open %s for reading'), self.base.output.term.bold(file)) sys.exit(1) def _transaction(self, args=None): cmd = args[0] if args else None if cmd == 'reset': self._clean() return self._resolve() if cmd in ['list', None]: if self.base._transaction: out = self.base.output.list_transaction(self.base._transaction) logger.info(out) elif cmd == 'run': try: self.base.do_transaction() except dnf.exceptions.Error as e: logger.error(_("Error:") + " " + ucd(e)) else: logger.info(_("Complete!")) self._clean() else: self._help('transaction') def _ts_run(self, args=None): self._transaction(['run']) def _quit(self, args=None): logger.info(_('Leaving Shell')) sys.exit(0)	9.61 KB	2022-09-09 11:21
📄 swap.py # # Copyright (C) 2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import _ from dnf.cli import commands import dnf.util import logging logger = logging.getLogger("dnf") class SwapCommand(commands.Command): """A class containing methods needed by the cli to execute the swap command. """ aliases = ('swap',) summary = _('run an interactive {prog} mod for remove and install one spec').format( prog=dnf.util.MAIN_PROG_UPPER) @staticmethod def set_argparser(parser): parser.add_argument('remove_spec', action="store", help=_('The specs that will be removed')) parser.add_argument('install_spec', action="store", help=_( 'The specs that will be installed')) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True commands._checkGPGKey(self.base, self.cli) commands._checkEnabledRepo(self.base, [self.opts.install_spec]) def _perform(self, cmd_str, spec): cmd_cls = self.cli.cli_commands.get(cmd_str) if cmd_cls is not None: cmd = cmd_cls(self.cli) self.cli.optparser.parse_command_args(cmd, [cmd_str, spec]) cmd.run() def run(self): # The install part must be performed before the remove one because it can # operate on local rpm files. Command line packages cannot be added # to the sack once the goal is created. self._perform('install', self.opts.install_spec) self._perform('remove', self.opts.remove_spec)	2.56 KB	2022-09-09 11:21
📄 updateinfo.py # updateinfo.py # UpdateInfo CLI command. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # """UpdateInfo CLI command.""" from __future__ import absolute_import from __future__ import print_function from __future__ import unicode_literals import collections import fnmatch import hawkey from dnf.cli import commands from dnf.cli.option_parser import OptionParser from dnf.i18n import _, exact_width from dnf.pycomp import unicode def _maxlen(iterable): """Return maximum length of items in a non-empty iterable.""" return max(exact_width(item) for item in iterable) class UpdateInfoCommand(commands.Command): """Implementation of the UpdateInfo command.""" TYPE2LABEL = {hawkey.ADVISORY_BUGFIX: _('bugfix'), hawkey.ADVISORY_ENHANCEMENT: _('enhancement'), hawkey.ADVISORY_SECURITY: _('security'), hawkey.ADVISORY_UNKNOWN: _('unknown'), hawkey.ADVISORY_NEWPACKAGE: _('newpackage')} SECURITY2LABEL = {'Critical': _('Critical/Sec.'), 'Important': _('Important/Sec.'), 'Moderate': _('Moderate/Sec.'), 'Low': _('Low/Sec.')} direct_commands = {'list-updateinfo' : 'list', 'list-security' : 'list', 'list-sec' : 'list', 'info-updateinfo' : 'info', 'info-security' : 'info', 'info-sec' : 'info', 'summary-updateinfo' : 'summary'} aliases = ['updateinfo', 'upif'] + list(direct_commands.keys()) summary = _('display advisories about packages') availability_default = 'available' availabilities = ['installed', 'updates', 'all', availability_default] def __init__(self, cli): """Initialize the command.""" super(UpdateInfoCommand, self).__init__(cli) self._installed_query = None @staticmethod def set_argparser(parser): availability = parser.add_mutually_exclusive_group() availability.add_argument( "--available", dest='_availability', const='available', action='store_const', help=_("advisories about newer versions of installed packages (default)")) availability.add_argument( "--installed", dest='_availability', const='installed', action='store_const', help=_("advisories about equal and older versions of installed packages")) availability.add_argument( "--updates", dest='_availability', const='updates', action='store_const', help=_("advisories about newer versions of those installed packages " "for which a newer version is available")) availability.add_argument( "--all", dest='_availability', const='all', action='store_const', help=_("advisories about any versions of installed packages")) cmds = ['summary', 'list', 'info'] output_format = parser.add_mutually_exclusive_group() output_format.add_argument("--summary", dest='_spec_action', const='summary', action='store_const', help=_('show summary of advisories (default)')) output_format.add_argument("--list", dest='_spec_action', const='list', action='store_const', help=_('show list of advisories')) output_format.add_argument("--info", dest='_spec_action', const='info', action='store_const', help=_('show info of advisories')) parser.add_argument("--with-cve", dest='with_cve', default=False, action='store_true', help=_('show only advisories with CVE reference')) parser.add_argument("--with-bz", dest='with_bz', default=False, action='store_true', help=_('show only advisories with bugzilla reference')) parser.add_argument('spec', nargs='', metavar='SPEC', choices=cmds, default=cmds[0], action=OptionParser.PkgNarrowCallback, help=_("Package specification")) def configure(self): """Do any command-specific configuration based on command arguments.""" self.cli.demands.available_repos = True self.cli.demands.sack_activation = True if self.opts.command in self.direct_commands: # we were called with direct command self.opts.spec_action = self.direct_commands[self.opts.command] else: if self.opts._spec_action: self.opts.spec_action = self.opts._spec_action if self.opts._availability: self.opts.availability = self.opts._availability else: # yum compatibility - search for all\|available\|installed\|updates in spec[0] if not self.opts.spec or self.opts.spec[0] not in self.availabilities: self.opts.availability = self.availability_default else: self.opts.availability = self.opts.spec.pop(0) # filtering by advisory types (security/bugfix/enhancement/newpackage) self.opts._advisory_types = set() if self.opts.bugfix: self.opts._advisory_types.add(hawkey.ADVISORY_BUGFIX) if self.opts.enhancement: self.opts._advisory_types.add(hawkey.ADVISORY_ENHANCEMENT) if self.opts.newpackage: self.opts._advisory_types.add(hawkey.ADVISORY_NEWPACKAGE) if self.opts.security: self.opts._advisory_types.add(hawkey.ADVISORY_SECURITY) # yum compatibility - yum accepts types also as positional arguments if self.opts.spec: spec = self.opts.spec.pop(0) if spec == 'bugfix': self.opts._advisory_types.add(hawkey.ADVISORY_BUGFIX) elif spec == 'enhancement': self.opts._advisory_types.add(hawkey.ADVISORY_ENHANCEMENT) elif spec in ('security', 'sec'): self.opts._advisory_types.add(hawkey.ADVISORY_SECURITY) elif spec == 'newpackage': self.opts._advisory_types.add(hawkey.ADVISORY_NEWPACKAGE) elif spec in ('bugzillas', 'bzs'): self.opts.with_bz = True elif spec == 'cves': self.opts.with_cve = True else: self.opts.spec.insert(0, spec) if self.opts.advisory: self.opts.spec.extend(self.opts.advisory) def run(self): """Execute the command with arguments.""" if self.opts.availability == 'installed': apkg_adv_insts = self.installed_apkg_adv_insts(self.opts.spec) description = _('installed') elif self.opts.availability == 'updates': apkg_adv_insts = self.updating_apkg_adv_insts(self.opts.spec) description = _('updates') elif self.opts.availability == 'all': apkg_adv_insts = self.all_apkg_adv_insts(self.opts.spec) description = _('all') else: apkg_adv_insts = self.available_apkg_adv_insts(self.opts.spec) description = _('available') if self.opts.spec_action == 'list': self.display_list(apkg_adv_insts) elif self.opts.spec_action == 'info': self.display_info(apkg_adv_insts) else: self.display_summary(apkg_adv_insts, description) def _newer_equal_installed(self, apackage): if self._installed_query is None: self._installed_query = self.base.sack.query().installed().apply() q = self._installed_query.filter(name=apackage.name, evr__gte=apackage.evr) return len(q) > 0 def _advisory_matcher(self, advisory): if not self.opts._advisory_types \ and not self.opts.spec \ and not self.opts.severity \ and not self.opts.bugzilla \ and not self.opts.cves \ and not self.opts.with_cve \ and not self.opts.with_bz: return True if advisory.type in self.opts._advisory_types: return True if any(fnmatch.fnmatchcase(advisory.id, pat) for pat in self.opts.spec): return True if self.opts.severity and advisory.severity in self.opts.severity: return True if self.opts.bugzilla and any([advisory.match_bug(bug) for bug in self.opts.bugzilla]): return True if self.opts.cves and any([advisory.match_cve(cve) for cve in self.opts.cves]): return True if self.opts.with_cve: if any([ref.type == hawkey.REFERENCE_CVE for ref in advisory.references]): return True if self.opts.with_bz: if any([ref.type == hawkey.REFERENCE_BUGZILLA for ref in advisory.references]): return True return False def _apackage_advisory_installed(self, pkgs_query, cmptype, specs): """Return (adv. package, advisory, installed) triplets.""" for apackage in pkgs_query.get_advisory_pkgs(cmptype): advisory = apackage.get_advisory(self.base.sack) advisory_match = self._advisory_matcher(advisory) apackage_match = any(fnmatch.fnmatchcase(apackage.name, pat) for pat in self.opts.spec) if advisory_match or apackage_match: installed = self._newer_equal_installed(apackage) yield apackage, advisory, installed def running_kernel_pkgs(self): """Return query containing packages of currently running kernel""" sack = self.base.sack q = sack.query().filterm(empty=True) kernel = sack.get_running_kernel() if kernel: q = q.union(sack.query().filterm(sourcerpm=kernel.sourcerpm)) return q def available_apkg_adv_insts(self, specs): """Return available (adv. package, adv., inst.) triplets""" # check advisories for the latest installed packages q = self.base.sack.query().installed().latest(1) # plus packages of the running kernel q = q.union(self.running_kernel_pkgs().installed()) return self._apackage_advisory_installed(q, hawkey.GT, specs) def installed_apkg_adv_insts(self, specs): """Return installed (adv. package, adv., inst.) triplets""" return self._apackage_advisory_installed( self.base.sack.query().installed(), hawkey.LT \| hawkey.EQ, specs) def updating_apkg_adv_insts(self, specs): """Return updating (adv. package, adv., inst.) triplets""" return self._apackage_advisory_installed( self.base.sack.query().filterm(upgradable=True), hawkey.GT, specs) def all_apkg_adv_insts(self, specs): """Return installed (adv. package, adv., inst.) triplets""" return self._apackage_advisory_installed( self.base.sack.query().installed(), hawkey.LT \| hawkey.EQ \| hawkey.GT, specs) def _summary(self, apkg_adv_insts): """Make the summary of advisories.""" # Remove duplicate advisory IDs. We assume that the ID is unique within # a repository and two advisories with the same IDs in different # repositories must have the same type. id2type = {} for (apkg, advisory, installed) in apkg_adv_insts: id2type[advisory.id] = advisory.type if advisory.type == hawkey.ADVISORY_SECURITY: id2type[(advisory.id, advisory.severity)] = (advisory.type, advisory.severity) return collections.Counter(id2type.values()) def display_summary(self, apkg_adv_insts, description): """Display the summary of advisories.""" typ2cnt = self._summary(apkg_adv_insts) if typ2cnt: print(_('Updates Information Summary: ') + description) # Convert types to strings and order the entries. label_counts = [ (0, _('New Package notice(s)'), typ2cnt[hawkey.ADVISORY_NEWPACKAGE]), (0, _('Security notice(s)'), typ2cnt[hawkey.ADVISORY_SECURITY]), (1, _('Critical Security notice(s)'), typ2cnt[(hawkey.ADVISORY_SECURITY, 'Critical')]), (1, _('Important Security notice(s)'), typ2cnt[(hawkey.ADVISORY_SECURITY, 'Important')]), (1, _('Moderate Security notice(s)'), typ2cnt[(hawkey.ADVISORY_SECURITY, 'Moderate')]), (1, _('Low Security notice(s)'), typ2cnt[(hawkey.ADVISORY_SECURITY, 'Low')]), (1, _('Unknown Security notice(s)'), typ2cnt[(hawkey.ADVISORY_SECURITY, None)]), (0, _('Bugfix notice(s)'), typ2cnt[hawkey.ADVISORY_BUGFIX]), (0, _('Enhancement notice(s)'), typ2cnt[hawkey.ADVISORY_ENHANCEMENT]), (0, _('other notice(s)'), typ2cnt[hawkey.ADVISORY_UNKNOWN])] width = _maxlen(unicode(v[2]) for v in label_counts if v[2]) for indent, label, count in label_counts: if not count: continue print(' %s %s' % (width + 4 * indent, unicode(count), label)) if self.base.conf.autocheck_running_kernel: self.cli._check_running_kernel() def display_list(self, apkg_adv_insts): """Display the list of advisories.""" def inst2mark(inst): if not self.opts.availability == 'all': return '' elif inst: return 'i ' else: return ' ' def type2label(typ, sev): if typ == hawkey.ADVISORY_SECURITY: return self.SECURITY2LABEL.get(sev, _('Unknown/Sec.')) else: return self.TYPE2LABEL.get(typ, _('unknown')) nevra_inst_dict = dict() for apkg, advisory, installed in apkg_adv_insts: nevra = '%s-%s.%s' % (apkg.name, apkg.evr, apkg.arch) if self.opts.with_cve or self.opts.with_bz: for ref in advisory.references: if ref.type == hawkey.REFERENCE_BUGZILLA and not self.opts.with_bz: continue elif ref.type == hawkey.REFERENCE_CVE and not self.opts.with_cve: continue nevra_inst_dict.setdefault((nevra, installed, advisory.updated), dict())[ref.id] = ( advisory.type, advisory.severity) else: nevra_inst_dict.setdefault((nevra, installed, advisory.updated), dict())[advisory.id] = ( advisory.type, advisory.severity) advlist = [] # convert types to labels, find max len of advisory IDs and types idw = tlw = nw = 0 for (nevra, inst, aupdated), id2type in sorted(nevra_inst_dict.items(), key=lambda x: x[0]): nw = max(nw, len(nevra)) for aid, atypesev in id2type.items(): idw = max(idw, len(aid)) label = type2label(atypesev) tlw = max(tlw, len(label)) advlist.append((inst2mark(inst), aid, label, nevra, aupdated)) for (inst, aid, label, nevra, aupdated) in advlist: if self.base.conf.verbose: print('%s%-s %-s %-s %s' % (inst, idw, aid, tlw, label, nw, nevra, aupdated)) else: print('%s%-s %-s %s' % (inst, idw, aid, tlw, label, nevra)) def display_info(self, apkg_adv_insts): """Display the details about available advisories.""" arches = self.base.sack.list_arches() verbose = self.base.conf.verbose labels = (_('Update ID'), _('Type'), _('Updated'), _('Bugs'), _('CVEs'), _('Description'), _('Severity'), _('Rights'), _('Files'), _('Installed')) def advisory2info(advisory, installed): attributes = [ [advisory.id], [self.TYPE2LABEL.get(advisory.type, _('unknown'))], [unicode(advisory.updated)], [], [], (advisory.description or '').splitlines(), [advisory.severity], (advisory.rights or '').splitlines(), sorted(set(pkg.filename for pkg in advisory.packages if pkg.arch in arches)), None] for ref in advisory.references: if ref.type == hawkey.REFERENCE_BUGZILLA: attributes[3].append('{} - {}'.format(ref.id, ref.title or '')) elif ref.type == hawkey.REFERENCE_CVE: attributes[4].append(ref.id) attributes[3].sort() attributes[4].sort() if not verbose: attributes[7] = None attributes[8] = None if self.opts.availability == 'all': attributes[9] = [_('true') if installed else _('false')] width = _maxlen(labels) lines = [] lines.append('=' * 79) lines.append(' ' + advisory.title) lines.append('=' * 79) for label, atr_lines in zip(labels, attributes): if atr_lines in (None, [None]): continue for i, line in enumerate(atr_lines): key = label if i == 0 else '' key_padding = width - exact_width(key) lines.append('%*s%s: %s' % (key_padding, "", key, line)) return '\n'.join(lines) advisories = set() for apkg, advisory, installed in apkg_adv_insts: advisories.add(advisory2info(advisory, installed)) print("\n\n".join(sorted(advisories, key=lambda x: x.lower())))	18.56 KB	2022-09-09 11:21
📄 upgrade.py # upgrade.py # Upgrade CLI command. # # Copyright (C) 2014-2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals import logging import dnf.exceptions import dnf.base from dnf.cli import commands from dnf.cli.option_parser import OptionParser from dnf.i18n import _ logger = logging.getLogger('dnf') class UpgradeCommand(commands.Command): """A class containing methods needed by the cli to execute the update command. """ aliases = ('upgrade', 'update', 'upgrade-to', 'update-to', 'localupdate', 'up') summary = _('upgrade a package or packages on your system') @staticmethod def set_argparser(parser): parser.add_argument('packages', nargs='*', help=_('Package to upgrade'), action=OptionParser.ParseSpecGroupFileCallback, metavar=_('PACKAGE')) def configure(self): """Verify that conditions are met so that this command can run. These include that there are enabled repositories with gpg keys, and that this command is being run by the root user. """ demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.resolving = True demands.root_user = True commands._checkGPGKey(self.base, self.cli) if not self.opts.filenames: commands._checkEnabledRepo(self.base) self.upgrade_minimal = None self.all_security = None self.skipped_grp_specs = None def run(self): cmp_type = "eq" if self.upgrade_minimal else "gte" self.cli._populate_update_security_filter(self.opts, cmp_type=cmp_type, all=self.all_security) if self.opts.filenames or self.opts.pkg_specs or self.opts.grp_specs: result = False result \|= self._update_modules() result \|= self._update_files() result \|= self._update_packages() result \|= self._update_groups() if result: return else: self.base.upgrade_all() return raise dnf.exceptions.Error(_('No packages marked for upgrade.')) def _update_modules(self): group_specs_num = len(self.opts.grp_specs) if dnf.base.WITH_MODULES: module_base = dnf.module.module_base.ModuleBase(self.base) self.skipped_grp_specs = module_base.upgrade(self.opts.grp_specs) else: self.skipped_grp_specs = self.opts.grp_specs return len(self.skipped_grp_specs) != group_specs_num def _update_files(self): success = False if self.opts.filenames: for pkg in self.base.add_remote_rpms(self.opts.filenames, strict=False, progress=self.base.output.progress): try: self.base.package_upgrade(pkg) success = True except dnf.exceptions.MarkingError as e: logger.info(_('No match for argument: %s'), self.base.output.term.bold(pkg.location)) return success def _update_packages(self): success = False for pkg_spec in self.opts.pkg_specs: try: self.base.upgrade(pkg_spec) success = True except dnf.exceptions.MarkingError as e: logger.info(_('No match for argument: %s'), self.base.output.term.bold(pkg_spec)) return success def _update_groups(self): if self.skipped_grp_specs: self.base.env_group_upgrade(self.skipped_grp_specs) return True return False	4.62 KB	2022-09-09 11:21
📄 upgrademinimal.py # # Copyright (C) 2016 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General # Public License for more details. You should have received a copy of the # GNU General Public License along with this program; if not, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import _ from dnf.cli.commands.upgrade import UpgradeCommand class UpgradeMinimalCommand(UpgradeCommand): """A class containing methods needed by the cli to execute the check command. """ aliases = ('upgrade-minimal', 'update-minimal', 'up-min') summary = _("upgrade, but only 'newest' package match which fixes a problem" " that affects your system") def configure(self): UpgradeCommand.configure(self) self.upgrade_minimal = True if not any([self.opts.bugfix, self.opts.enhancement, self.opts.newpackage, self.opts.security, self.opts.advisory, self.opts.bugzilla, self.opts.cves, self.opts.severity]): self.all_security = True	1.76 KB	2022-09-09 11:21

Edit File:

Terminal