�ɲɾ�����ӯ�����һ��ˣ��������С���˴��ͣ�������P���ҹ��ñ˽��ά�Բ��������˸߸ԣ�������ơ��ҹ��ñ�����ά�Բ���ˡ���˳^�ӣ������ӡ�
���ͯj�ӣ��ƺ���ӣ�
? PNG ?%k25u25%fgd5n!? PNG ?%k25u25%fgd5n!? PNG ?%k25u25%fgd5n!? PNG ?%k25u25%fgd5n!repodiff.py 0000644 00000026323 15252533450 0006726 0 ustar 00 # repodiff.py
# DNF plugin adding a command to show differencies between two sets
# of repositories.
#
# 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 unicode_literals
import dnf.cli
from dnf.cli.option_parser import OptionParser
import hawkey
from dnfpluginscore import _
class RepoDiff(dnf.Plugin):
name = "repodiff"
def __init__(self, base, cli):
super(RepoDiff, self).__init__(base, cli)
if cli is None:
return
cli.register_command(RepoDiffCommand)
class RepoDiffCommand(dnf.cli.Command):
aliases = ("repodiff",)
summary = _("List differences between two sets of repositories")
@staticmethod
def set_argparser(parser):
# I'd like to use --old and --new options like Yum did.
# But ability to disable abbreviated long options is added
# only in Python >= 3.5
# So in command arguments we are not able to use arguments,
# which are prefixes of main arguments (i.w. --new would be
# treated as --newpackage). This is because we run .parse_args
# two times - for main and then for command arguments.
# https://stackoverflow.com/questions/33900846
parser.add_argument("--repo-old", "-o", default=[], action="append", dest="old",
help=_("Specify old repository, can be used multiple times"))
parser.add_argument("--repo-new", "-n", default=[], action="append", dest="new",
help=_("Specify new repository, can be used multiple times"))
parser.add_argument("--arch", "--archlist", "-a", default=[],
action=OptionParser._SplitCallback, dest="arches",
help=_("Specify architectures to compare, can be used "
"multiple times. By default, only source rpms are "
"compared."))
parser.add_argument("--size", "-s", action="store_true",
help=_("Output additional data about the size of the changes."))
parser.add_argument("--compare-arch", action="store_true",
help=_("Compare packages also by arch. By default "
"packages are compared just by name."))
parser.add_argument("--simple", action="store_true",
help=_("Output a simple one line message for modified packages."))
parser.add_argument("--downgrade", action="store_true",
help=_("Split the data for modified packages between "
"upgraded and downgraded packages."))
def configure(self):
demands = self.cli.demands
demands.sack_activation = True
demands.available_repos = True
demands.changelogs = True
self.base.conf.disable_excludes = ["all"]
# TODO yum was able to handle mirrorlist in --new/--old arguments
# Can be resolved by improving --repofrompath option
if not self.opts.new or not self.opts.old:
msg = _("Both old and new repositories must be set.")
raise dnf.exceptions.Error(msg)
for repo in self.base.repos.all():
if repo.id in self.opts.new + self.opts.old:
repo.enable()
else:
repo.disable()
if not self.opts.arches:
self.opts.arches = ['src']
def _pkgkey(self, pkg):
if self.opts.compare_arch:
return (pkg.name, pkg.arch)
return pkg.name
def _repodiff(self, old, new):
'''compares packagesets old and new, returns dictionary with packages:
added: only in new set
removed: only in old set
upgraded: in both old and new, new has bigger evr
downgraded: in both old and new, new has lower evr
obsoletes: dictionary of which old package is obsoleted by which new
'''
old_d = dict([(self._pkgkey(p), p) for p in old])
old_keys = set(old_d.keys())
new_d = dict([(self._pkgkey(p), p) for p in new])
new_keys = set(new_d.keys())
# mapping obsoleted_package_from_old: obsoleted_by_package_from_new
obsoletes = dict()
for obsoleter in new.filter(obsoletes=old):
for obsoleted in old.filter(provides=obsoleter.obsoletes):
obsoletes[self._pkgkey(obsoleted)] = obsoleter
evr_cmp = self.base.sack.evr_cmp
repodiff = dict(
added=[new_d[k] for k in new_keys - old_keys],
removed=[old_d[k] for k in old_keys - new_keys],
obsoletes=obsoletes,
upgraded=[],
downgraded=[])
for k in old_keys.intersection(new_keys):
pkg_old = old_d[k]
pkg_new = new_d[k]
if pkg_old.evr == pkg_new.evr:
continue
if evr_cmp(pkg_old.evr, pkg_new.evr) > 0:
repodiff['downgraded'].append((pkg_old, pkg_new))
else:
repodiff['upgraded'].append((pkg_old, pkg_new))
return repodiff
def _report(self, repodiff):
def pkgstr(pkg):
if self.opts.compare_arch:
return str(pkg)
return "%s-%s" % (pkg.name, pkg.evr)
def sizestr(num):
msg = str(num)
if num > 0:
msg += " ({})".format(dnf.cli.format.format_number(num).strip())
elif num < 0:
msg += " (-{})".format(dnf.cli.format.format_number(-num).strip())
return msg
def report_modified(pkg_old, pkg_new):
msgs = []
if self.opts.simple:
msgs.append("%s -> %s" % (pkgstr(pkg_old), pkgstr(pkg_new)))
else:
msgs.append('')
msgs.append("%s -> %s" % (pkgstr(pkg_old), pkgstr(pkg_new)))
msgs.append('-' * len(msgs[-1]))
if pkg_old.changelogs:
old_chlog = pkg_old.changelogs[0]
else:
old_chlog = None
for chlog in pkg_new.changelogs:
if old_chlog:
if chlog['timestamp'] < old_chlog['timestamp']:
break
elif (chlog['timestamp'] == old_chlog['timestamp'] and
chlog['author'] == old_chlog['author'] and
chlog['text'] == old_chlog['text']):
break
msgs.append('* %s %s\n%s' % (
chlog['timestamp'].strftime("%a %b %d %Y"),
dnf.i18n.ucd(chlog['author']),
dnf.i18n.ucd(chlog['text'])))
if self.opts.size:
msgs.append(_("Size change: {} bytes").format(
pkg_new.size - pkg_old.size))
print('\n'.join(msgs))
sizes = dict(added=0, removed=0, upgraded=0, downgraded=0)
for pkg in sorted(repodiff['added']):
print(_("Added package : {}").format(pkgstr(pkg)))
sizes['added'] += pkg.size
for pkg in sorted(repodiff['removed']):
print(_("Removed package: {}").format(pkgstr(pkg)))
obsoletedby = repodiff['obsoletes'].get(self._pkgkey(pkg))
if obsoletedby:
print(_("Obsoleted by : {}").format(pkgstr(obsoletedby)))
sizes['removed'] += pkg.size
if self.opts.downgrade:
if repodiff['upgraded']:
print(_("\nUpgraded packages"))
for (pkg_old, pkg_new) in sorted(repodiff['upgraded']):
sizes['upgraded'] += (pkg_new.size - pkg_old.size)
report_modified(pkg_old, pkg_new)
if repodiff['downgraded']:
print(_("\nDowngraded packages"))
for (pkg_old, pkg_new) in sorted(repodiff['downgraded']):
sizes['downgraded'] += (pkg_new.size - pkg_old.size)
report_modified(pkg_old, pkg_new)
else:
modified = repodiff['upgraded'] + repodiff['downgraded']
if modified:
print(_("\nModified packages"))
for (pkg_old, pkg_new) in sorted(modified):
sizes['upgraded'] += (pkg_new.size - pkg_old.size)
report_modified(pkg_old, pkg_new)
print(_("\nSummary"))
print(_("Added packages: {}").format(len(repodiff['added'])))
print(_("Removed packages: {}").format(len(repodiff['removed'])))
if self.opts.downgrade:
print(_("Upgraded packages: {}").format(len(repodiff['upgraded'])))
print(_("Downgraded packages: {}").format(len(repodiff['downgraded'])))
else:
print(_("Modified packages: {}").format(
len(repodiff['upgraded']) + len(repodiff['downgraded'])))
if self.opts.size:
print(_("Size of added packages: {}").format(sizestr(sizes['added'])))
print(_("Size of removed packages: {}").format(sizestr(sizes['removed'])))
if not self.opts.downgrade:
print(_("Size of modified packages: {}").format(
sizestr(sizes['upgraded'] + sizes['downgraded'])))
else:
print(_("Size of upgraded packages: {}").format(
sizestr(sizes['upgraded'])))
print(_("Size of downgraded packages: {}").format(
sizestr(sizes['downgraded'])))
print(_("Size change: {}").format(
sizestr(sizes['added'] + sizes['upgraded'] + sizes['downgraded'] -
sizes['removed'])))
def run(self):
# prepare old and new packagesets based by given arguments
q_new = self.base.sack.query(hawkey.IGNORE_EXCLUDES).filter(
reponame=self.opts.new)
q_old = self.base.sack.query(hawkey.IGNORE_EXCLUDES).filter(
reponame=self.opts.old)
if self.opts.arches and '*' not in self.opts.arches:
q_new.filterm(arch=self.opts.arches)
q_old.filterm(arch=self.opts.arches)
if self.opts.compare_arch:
q_new.filterm(latest_per_arch=1)
q_old.filterm(latest_per_arch=1)
else:
q_new.filterm(latest=1)
q_old.filterm(latest=1)
q_new.apply()
q_old.apply()
self._report(self._repodiff(q_old, q_new))
system_upgrade.py 0000644 00000064251 15252533450 0010165 0 ustar 00 # -*- coding: utf-8 -*-
#
# Copyright (c) 2015-2020 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
#
# Author(s): Will Woods
"""system_upgrade.py - DNF plugin to handle major-version system upgrades."""
from subprocess import call, Popen, check_output, CalledProcessError
import json
import os
import os.path
import re
import sys
import uuid
from systemd import journal
from dnfpluginscore import _, logger
import dnf
import dnf.cli
from dnf.cli import CliError
from dnf.i18n import ucd
import dnf.transaction
from dnf.transaction_sr import serialize_transaction, TransactionReplay
import libdnf.conf
# Translators: This string is only used in unit tests.
_("the color of the sky")
DOWNLOAD_FINISHED_ID = uuid.UUID('9348174c5cc74001a71ef26bd79d302e')
REBOOT_REQUESTED_ID = uuid.UUID('fef1cc509d5047268b83a3a553f54b43')
UPGRADE_STARTED_ID = uuid.UUID('3e0a5636d16b4ca4bbe5321d06c6aa62')
UPGRADE_FINISHED_ID = uuid.UUID('8cec00a1566f4d3594f116450395f06c')
ID_TO_IDENTIFY_BOOTS = UPGRADE_STARTED_ID
PLYMOUTH = '/usr/bin/plymouth'
RELEASEVER_MSG = _(
"Need a --releasever greater than the current system version.")
DOWNLOAD_FINISHED_MSG = _( # Translators: do not change "reboot" here
"Download complete! Use 'dnf {command} reboot' to start the upgrade.\n"
"To remove cached metadata and transaction use 'dnf {command} clean'")
CANT_RESET_RELEASEVER = _(
"Sorry, you need to use 'download --releasever' instead of '--network'")
STATE_VERSION = 2
# --- Miscellaneous helper functions ------------------------------------------
def reboot():
if os.getenv("DNF_SYSTEM_UPGRADE_NO_REBOOT", default=False):
logger.info(_("Reboot turned off, not rebooting."))
else:
Popen(["systemctl", "reboot"])
def get_url_from_os_release():
key = "UPGRADE_GUIDE_URL="
for path in ["/etc/os-release", "/usr/lib/os-release"]:
try:
with open(path) as release_file:
for line in release_file:
line = line.strip()
if line.startswith(key):
return line[len(key):].strip('"')
except IOError:
continue
return None
# DNF-FIXME: dnf.util.clear_dir() doesn't delete regular files :/
def clear_dir(path, ignore=[]):
if not os.path.isdir(path):
return
for entry in os.listdir(path):
fullpath = os.path.join(path, entry)
if fullpath in ignore:
continue
try:
if os.path.isdir(fullpath):
dnf.util.rm_rf(fullpath)
else:
os.unlink(fullpath)
except OSError:
pass
def check_release_ver(conf, target=None):
if dnf.rpm.detect_releasever(conf.installroot) == conf.releasever:
raise CliError(RELEASEVER_MSG)
if target and target != conf.releasever:
# it's too late to set releasever here, so this can't work.
# (see https://bugzilla.redhat.com/show_bug.cgi?id=1212341)
raise CliError(CANT_RESET_RELEASEVER)
def disable_blanking():
try:
tty = open('/dev/tty0', 'wb')
tty.write(b'\33[9;0]')
except Exception as e:
print(_("Screen blanking can't be disabled: %s") % e)
# --- State object - for tracking upgrade state between runs ------------------
# DNF-INTEGRATION-NOTE: basically the same thing as dnf.persistor.JSONDB
class State(object):
def __init__(self, statefile):
self.statefile = statefile
self._data = {}
self._read()
def _read(self):
try:
with open(self.statefile) as fp:
self._data = json.load(fp)
except IOError:
self._data = {}
except ValueError:
self._data = {}
logger.warning(_("Failed loading state file: %s, continuing with "
"empty state."), self.statefile)
def write(self):
dnf.util.ensure_dir(os.path.dirname(self.statefile))
with open(self.statefile, 'w') as outf:
json.dump(self._data, outf, indent=4, sort_keys=True)
def clear(self):
if os.path.exists(self.statefile):
os.unlink(self.statefile)
self._read()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type is None:
self.write()
# helper function for creating properties. pylint: disable=protected-access
def _prop(option): # pylint: disable=no-self-argument
def setprop(self, value):
self._data[option] = value
def getprop(self):
return self._data.get(option)
return property(getprop, setprop)
# !!! Increase STATE_VERSION for any changes in data structure like a new property or a new
# data structure !!!
state_version = _prop("state_version")
download_status = _prop("download_status")
destdir = _prop("destdir")
target_releasever = _prop("target_releasever")
system_releasever = _prop("system_releasever")
gpgcheck = _prop("gpgcheck")
# list of repos with gpgcheck=True
gpgcheck_repos = _prop("gpgcheck_repos")
# list of repos with repo_gpgcheck=True
repo_gpgcheck_repos = _prop("repo_gpgcheck_repos")
upgrade_status = _prop("upgrade_status")
upgrade_command = _prop("upgrade_command")
distro_sync = _prop("distro_sync")
enable_disable_repos = _prop("enable_disable_repos")
module_platform_id = _prop("module_platform_id")
# --- Plymouth output helpers -------------------------------------------------
class PlymouthOutput(object):
"""A plymouth output helper class.
Filters duplicate calls, and stops calling the plymouth binary if we
fail to contact it.
"""
def __init__(self):
self.alive = True
self._last_args = dict()
self._last_msg = None
def _plymouth(self, cmd, *args):
dupe_cmd = (args == self._last_args.get(cmd))
if (self.alive and not dupe_cmd) or cmd == '--ping':
try:
self.alive = (call((PLYMOUTH, cmd) + args) == 0)
except OSError:
self.alive = False
self._last_args[cmd] = args
return self.alive
def ping(self):
return self._plymouth("--ping")
def message(self, msg):
if self._last_msg and self._last_msg != msg:
self._plymouth("hide-message", "--text", self._last_msg)
self._last_msg = msg
return self._plymouth("display-message", "--text", msg)
def set_mode(self):
mode = 'updates'
try:
s = check_output([PLYMOUTH, '--help'])
if re.search('--system-upgrade', ucd(s)):
mode = 'system-upgrade'
except (CalledProcessError, OSError):
pass
return self._plymouth("change-mode", "--" + mode)
def progress(self, percent):
return self._plymouth("system-update", "--progress", str(percent))
# A single PlymouthOutput instance for us to use within this module
Plymouth = PlymouthOutput()
# A TransactionProgress class that updates plymouth for us.
class PlymouthTransactionProgress(dnf.callback.TransactionProgress):
# pylint: disable=too-many-arguments
def progress(self, package, action, ti_done, ti_total, ts_done, ts_total):
self._update_plymouth(package, action, ts_done, ts_total)
def _update_plymouth(self, package, action, current, total):
# Prevents quick jumps of progressbar when pretrans scriptlets
# and TRANS_PREPARATION are reported as 1/1
if total == 1:
return
# Verification goes through all the packages again,
# which resets the "current" param value, this prevents
# resetting of the progress bar as well. (Rhbug:1809096)
if action != dnf.callback.PKG_VERIFY:
Plymouth.progress(int(90.0 * current / total))
else:
Plymouth.progress(90 + int(10.0 * current / total))
Plymouth.message(self._fmt_event(package, action, current, total))
def _fmt_event(self, package, action, current, total):
action = dnf.transaction.ACTIONS.get(action, action)
return "[%d/%d] %s %s..." % (current, total, action, package)
# --- journal helpers -------------------------------------------------
def find_boots(message_id):
"""Find all boots with this message id.
Returns the entries of all found boots.
"""
j = journal.Reader()
j.add_match(MESSAGE_ID=message_id.hex, # identify the message
_UID=0) # prevent spoofing of logs
oldboot = None
for entry in j:
boot = entry['_BOOT_ID']
if boot == oldboot:
continue
oldboot = boot
yield entry
def list_logs():
print(_('The following boots appear to contain upgrade logs:'))
n = -1
for n, entry in enumerate(find_boots(ID_TO_IDENTIFY_BOOTS)):
print('{} / {.hex}: {:%Y-%m-%d %H:%M:%S} {}→{}'.format(
n + 1,
entry['_BOOT_ID'],
entry['__REALTIME_TIMESTAMP'],
entry.get('SYSTEM_RELEASEVER', '??'),
entry.get('TARGET_RELEASEVER', '??')))
if n == -1:
print(_('-- no logs were found --'))
def pick_boot(message_id, n):
boots = list(find_boots(message_id))
# Positive indices index all found boots starting with 1 and going forward,
# zero is the current boot, and -1, -2, -3 are previous going backwards.
# This is the same as journalctl.
try:
if n == 0:
raise IndexError
if n > 0:
n -= 1
return boots[n]['_BOOT_ID']
except IndexError:
raise CliError(_("Cannot find logs with this index."))
def show_log(n):
boot_id = pick_boot(ID_TO_IDENTIFY_BOOTS, n)
process = Popen(['journalctl', '--boot', boot_id.hex])
process.wait()
rc = process.returncode
if rc == 1:
raise dnf.exceptions.Error(_("Unable to match systemd journal entry"))
CMDS = ['download', 'clean', 'reboot', 'upgrade', 'log']
# --- The actual Plugin and Command objects! ----------------------------------
class SystemUpgradePlugin(dnf.Plugin):
name = 'system-upgrade'
def __init__(self, base, cli):
super(SystemUpgradePlugin, self).__init__(base, cli)
if cli:
cli.register_command(SystemUpgradeCommand)
cli.register_command(OfflineUpgradeCommand)
cli.register_command(OfflineDistrosyncCommand)
class SystemUpgradeCommand(dnf.cli.Command):
aliases = ('system-upgrade', 'fedup',)
summary = _("Prepare system for upgrade to a new release")
DATADIR = 'var/lib/dnf/system-upgrade'
def __init__(self, cli):
super(SystemUpgradeCommand, self).__init__(cli)
self.datadir = os.path.join(cli.base.conf.installroot, self.DATADIR)
self.transaction_file = os.path.join(self.datadir, 'system-upgrade-transaction.json')
self.magic_symlink = os.path.join(cli.base.conf.installroot, 'system-update')
self.state = State(os.path.join(self.datadir, 'system-upgrade-state.json'))
@staticmethod
def set_argparser(parser):
parser.add_argument("--no-downgrade", dest='distro_sync',
action='store_false',
help=_("keep installed packages if the new "
"release's version is older"))
parser.add_argument('tid', nargs=1, choices=CMDS,
metavar="[%s]" % "|".join(CMDS))
parser.add_argument('--number', type=int, help=_('which logs to show'))
def log_status(self, message, message_id):
"""Log directly to the journal."""
journal.send(message,
MESSAGE_ID=message_id,
PRIORITY=journal.LOG_NOTICE,
SYSTEM_RELEASEVER=self.state.system_releasever,
TARGET_RELEASEVER=self.state.target_releasever,
DNF_VERSION=dnf.const.VERSION)
def pre_configure(self):
self._call_sub("check")
self._call_sub("pre_configure")
def configure(self):
self._call_sub("configure")
def run(self):
self._call_sub("run")
def run_transaction(self):
self._call_sub("transaction")
def run_resolved(self):
self._call_sub("resolved")
def _call_sub(self, name):
subfunc = getattr(self, name + '_' + self.opts.tid[0], None)
if callable(subfunc):
subfunc()
def _check_state_version(self, command):
if self.state.state_version != STATE_VERSION:
msg = _("Incompatible version of data. Rerun 'dnf {command} download [OPTIONS]'"
"").format(command=command)
raise CliError(msg)
def _set_cachedir(self):
# set download directories from json state file
self.base.conf.cachedir = self.datadir
self.base.conf.destdir = self.state.destdir if self.state.destdir else None
def _get_forward_reverse_pkg_reason_pairs(self):
"""
forward = {repoid:{pkg_nevra: {tsi.action: tsi.reason}}
reverse = {pkg_nevra: {tsi.action: tsi.reason}}
:return: forward, reverse
"""
backward_action = set(dnf.transaction.BACKWARD_ACTIONS + [libdnf.transaction.TransactionItemAction_REINSTALLED])
forward_actions = set(dnf.transaction.FORWARD_ACTIONS)
forward = {}
reverse = {}
for tsi in self.cli.base.transaction:
if tsi.action in forward_actions:
pkg = tsi.pkg
forward.setdefault(pkg.repo.id, {}).setdefault(
str(pkg), {})[tsi.action] = tsi.reason
elif tsi.action in backward_action:
reverse.setdefault(str(tsi.pkg), {})[tsi.action] = tsi.reason
return forward, reverse
# == pre_configure_*: set up action-specific demands ==========================
def pre_configure_download(self):
# only download subcommand accepts --destdir command line option
self.base.conf.cachedir = self.datadir
self.base.conf.destdir = self.opts.destdir if self.opts.destdir else None
if 'offline-distrosync' == self.opts.command and not self.opts.distro_sync:
raise CliError(
_("Command 'offline-distrosync' cannot be used with --no-downgrade option"))
elif 'offline-upgrade' == self.opts.command:
self.opts.distro_sync = False
def pre_configure_reboot(self):
self._set_cachedir()
def pre_configure_upgrade(self):
self._set_cachedir()
if self.state.enable_disable_repos:
self.opts.repos_ed = self.state.enable_disable_repos
self.base.conf.releasever = self.state.target_releasever
def pre_configure_clean(self):
self._set_cachedir()
# == configure_*: set up action-specific demands ==========================
def configure_download(self):
if 'system-upgrade' == self.opts.command or 'fedup' == self.opts.command:
logger.warning(_('WARNING: this operation is not supported on the RHEL distribution. '
'Proceed at your own risk.'))
help_url = get_url_from_os_release()
if help_url:
msg = _('Additional information for System Upgrade: {}')
logger.info(msg.format(ucd(help_url)))
if self.base._promptWanted():
msg = _('Before you continue ensure that your system is fully upgraded by running '
'"dnf --refresh upgrade". Do you want to continue')
if self.base.conf.assumeno or not self.base.output.userconfirm(
msg='{} [y/N]: '.format(msg), defaultyes_msg='{} [Y/n]: '.format(msg)):
logger.error(_("Operation aborted."))
sys.exit(1)
check_release_ver(self.base.conf, target=self.opts.releasever)
elif 'offline-upgrade' == self.opts.command:
self.cli._populate_update_security_filter(self.opts)
self.cli.demands.root_user = True
self.cli.demands.resolving = True
self.cli.demands.available_repos = True
self.cli.demands.sack_activation = True
self.cli.demands.freshest_metadata = True
# We want to do the depsolve / download / transaction-test, but *not*
# run the actual RPM transaction to install the downloaded packages.
# Setting the "test" flag makes the RPM transaction a test transaction,
# so nothing actually gets installed.
# (It also means that we run two test transactions in a row, which is
# kind of silly, but that's something for DNF to fix...)
self.base.conf.tsflags += ["test"]
def configure_reboot(self):
# FUTURE: add a --debug-shell option to enable debug shell:
# systemctl add-wants system-update.target debug-shell.service
self.cli.demands.root_user = True
def configure_upgrade(self):
# same as the download, but offline and non-interactive. so...
self.cli.demands.root_user = True
self.cli.demands.resolving = True
self.cli.demands.available_repos = True
self.cli.demands.sack_activation = True
# use the saved value for --allowerasing, etc.
self.opts.distro_sync = self.state.distro_sync
if self.state.gpgcheck is not None:
self.base.conf.gpgcheck = self.state.gpgcheck
if self.state.gpgcheck_repos is not None:
for repo in self.base.repos.values():
repo.gpgcheck = repo.id in self.state.gpgcheck_repos
if self.state.repo_gpgcheck_repos is not None:
for repo in self.base.repos.values():
repo.repo_gpgcheck = repo.id in self.state.repo_gpgcheck_repos
self.base.conf.module_platform_id = self.state.module_platform_id
# don't try to get new metadata, 'cuz we're offline
self.cli.demands.cacheonly = True
# and don't ask any questions (we confirmed all this beforehand)
self.base.conf.assumeyes = True
self.cli.demands.transaction_display = PlymouthTransactionProgress()
# upgrade operation already removes all element that must be removed. Additional removal
# could trigger unwanted changes in transaction.
self.base.conf.clean_requirements_on_remove = False
self.base.conf.install_weak_deps = False
def configure_clean(self):
self.cli.demands.root_user = True
def configure_log(self):
pass
# == check_*: do any action-specific checks ===============================
def check_reboot(self):
if not self.state.download_status == 'complete':
raise CliError(_("system is not ready for upgrade"))
self._check_state_version(self.opts.command)
if self.state.upgrade_command != self.opts.command:
msg = _("the transaction was not prepared for '{command}'. "
"Rerun 'dnf {command} download [OPTIONS]'").format(command=self.opts.command)
raise CliError(msg)
if os.path.lexists(self.magic_symlink):
raise CliError(_("upgrade is already scheduled"))
dnf.util.ensure_dir(self.datadir)
# FUTURE: checkRPMDBStatus(self.state.download_transaction_id)
def check_upgrade(self):
if not os.path.lexists(self.magic_symlink):
logger.info(_("trigger file does not exist. exiting quietly."))
raise SystemExit(0)
if os.readlink(self.magic_symlink) != self.datadir:
logger.info(_("another upgrade tool is running. exiting quietly."))
raise SystemExit(0)
# Delete symlink ASAP to avoid reboot loops
dnf.yum.misc.unlink_f(self.magic_symlink)
command = self.state.upgrade_command
if not command:
command = self.opts.command
self._check_state_version(command)
if not self.state.upgrade_status == 'ready':
msg = _("use 'dnf {command} reboot' to begin the upgrade").format(command=command)
raise CliError(msg)
# == run_*: run the action/prep the transaction ===========================
def run_prepare(self):
# make the magic symlink
os.symlink(self.datadir, self.magic_symlink)
# set upgrade_status so that the upgrade can run
with self.state as state:
state.upgrade_status = 'ready'
def run_reboot(self):
self.run_prepare()
if not self.opts.tid[0] == "reboot":
return
self.log_status(_("Rebooting to perform upgrade."),
REBOOT_REQUESTED_ID)
reboot()
def run_download(self):
# Mark everything in the world for upgrade/sync
if self.opts.distro_sync:
self.base.distro_sync()
else:
self.base.upgrade_all()
if self.opts.command not in ['offline-upgrade', 'offline-distrosync']:
# Mark all installed groups and environments for upgrade
self.base.read_comps()
installed_groups = [g.id for g in self.base.comps.groups if self.base.history.group.get(g.id)]
if installed_groups:
self.base.env_group_upgrade(installed_groups)
installed_environments = [g.id for g in self.base.comps.environments if self.base.history.env.get(g.id)]
if installed_environments:
self.base.env_group_upgrade(installed_environments)
with self.state as state:
state.download_status = 'downloading'
state.target_releasever = self.base.conf.releasever
state.destdir = self.base.conf.destdir
def run_upgrade(self):
# change the upgrade status (so we can detect crashed upgrades later)
command = ''
with self.state as state:
state.upgrade_status = 'incomplete'
command = state.upgrade_command
if command == 'offline-upgrade':
msg = _("Starting offline upgrade. This will take a while.")
elif command == 'offline-distrosync':
msg = _("Starting offline distrosync. This will take a while.")
else:
msg = _("Starting system upgrade. This will take a while.")
self.log_status(msg, UPGRADE_STARTED_ID)
# reset the splash mode and let the user know we're running
Plymouth.set_mode()
Plymouth.progress(0)
Plymouth.message(msg)
# disable screen blanking
disable_blanking()
self.replay = TransactionReplay(self.base, self.transaction_file)
self.replay.run()
def run_clean(self):
logger.info(_("Cleaning up downloaded data..."))
# Don't delete persistor, it contains paths for downloaded packages
# that are used by dnf during finalizing base to clean them up
clear_dir(self.base.conf.cachedir,
[dnf.persistor.TempfilePersistor(self.base.conf.cachedir).db_path])
with self.state as state:
state.download_status = None
state.state_version = None
state.upgrade_status = None
state.upgrade_command = None
state.destdir = None
def run_log(self):
if self.opts.number:
show_log(self.opts.number)
else:
list_logs()
# == resolved_*: do staff after succesful resolvement =====================
def resolved_upgrade(self):
"""Adjust transaction reasons according to stored values"""
self.replay.post_transaction()
# == transaction_*: do stuff after a successful transaction ===============
def transaction_download(self):
transaction = self.base.history.get_current()
if not transaction.packages():
logger.info(_("The system-upgrade transaction is empty, your system is already up-to-date."))
return
data = serialize_transaction(transaction)
try:
with open(self.transaction_file, "w") as f:
json.dump(data, f, indent=4, sort_keys=True)
f.write("\n")
print(_("Transaction saved to {}.").format(self.transaction_file))
except OSError as e:
raise dnf.cli.CliError(_('Error storing transaction: {}').format(str(e)))
# Okay! Write out the state so the upgrade can use it.
system_ver = dnf.rpm.detect_releasever(self.base.conf.installroot)
with self.state as state:
state.download_status = 'complete'
state.state_version = STATE_VERSION
state.distro_sync = self.opts.distro_sync
state.gpgcheck = self.base.conf.gpgcheck
state.gpgcheck_repos = [
repo.id for repo in self.base.repos.values() if repo.gpgcheck]
state.repo_gpgcheck_repos = [
repo.id for repo in self.base.repos.values() if repo.repo_gpgcheck]
state.system_releasever = system_ver
state.target_releasever = self.base.conf.releasever
state.module_platform_id = self.base.conf.module_platform_id
state.enable_disable_repos = self.opts.repos_ed
state.destdir = self.base.conf.destdir
state.upgrade_command = self.opts.command
msg = DOWNLOAD_FINISHED_MSG.format(command=self.opts.command)
logger.info(msg)
self.log_status(_("Download finished."), DOWNLOAD_FINISHED_ID)
def transaction_upgrade(self):
Plymouth.message(_("Upgrade complete! Cleaning up and rebooting..."))
self.log_status(_("Upgrade complete! Cleaning up and rebooting..."),
UPGRADE_FINISHED_ID)
self.run_clean()
if self.opts.tid[0] == "upgrade":
reboot()
class OfflineUpgradeCommand(SystemUpgradeCommand):
aliases = ('offline-upgrade',)
summary = _("Prepare offline upgrade of the system")
class OfflineDistrosyncCommand(SystemUpgradeCommand):
aliases = ('offline-distrosync',)
summary = _("Prepare offline distrosync of the system")
changelog.py 0000644 00000011547 15252533450 0007061 0 ustar 00 # changelog.py
# DNF plugin adding a command changelog.
#
# 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 unicode_literals
import argparse
import collections
import dateutil.parser
from dnfpluginscore import _, P_, logger
import dnf
import dnf.cli
def validate_date(val):
try:
return dateutil.parser.parse(val, fuzzy=True)
except (ValueError, TypeError, OverflowError):
raise argparse.ArgumentTypeError(_('Not a valid date: "{0}".').format(val))
@dnf.plugin.register_command
class ChangelogCommand(dnf.cli.Command):
aliases = ('changelog',)
summary = _('Show changelog data of packages')
@staticmethod
def set_argparser(parser):
filter_group = parser.add_mutually_exclusive_group()
filter_group.add_argument(
'--since', metavar="DATE", default=None,
type=validate_date,
help=_('show changelog entries since DATE. To avoid ambiguosity, '
'YYYY-MM-DD format is recommended.'))
filter_group.add_argument(
'--count', default=None, type=int,
help=_('show given number of changelog entries per package'))
filter_group.add_argument(
'--upgrades', default=False, action='store_true',
help=_('show only new changelog entries for packages, that provide an '
'upgrade for some of already installed packages.'))
parser.add_argument("package", nargs='*', metavar=_('PACKAGE'))
def configure(self):
demands = self.cli.demands
demands.available_repos = True
demands.sack_activation = True
demands.changelogs = True
def query(self):
q = self.base.sack.query()
if self.opts.package:
q.filterm(empty=True)
for pkg in self.opts.package:
pkg_q = dnf.subject.Subject(pkg, ignore_case=True).get_best_query(
self.base.sack, with_nevra=True,
with_provides=False, with_filenames=False)
if self.opts.repo:
pkg_q.filterm(reponame=self.opts.repo)
if pkg_q:
q = q.union(pkg_q.latest())
else:
logger.info(_('No match for argument: %s') % pkg)
elif self.opts.repo:
q.filterm(reponame=self.opts.repo)
if self.opts.upgrades:
q = q.upgrades()
else:
q = q.available()
return q
def by_srpm(self, packages):
by_srpm = collections.OrderedDict()
for pkg in sorted(packages):
by_srpm.setdefault((pkg.source_name or pkg.name, pkg.evr), []).append(pkg)
return by_srpm
def filter_changelogs(self, package):
if self.opts.upgrades:
return self.base.latest_changelogs(package)
elif self.opts.count:
return package.changelogs[:self.opts.count]
elif self.opts.since:
return [chlog for chlog in package.changelogs
if chlog['timestamp'] >= self.opts.since.date()]
else:
return package.changelogs
def run(self):
if self.opts.since:
logger.info(_('Listing changelogs since {}').format(self.opts.since))
elif self.opts.count:
logger.info(P_('Listing only latest changelog',
'Listing {} latest changelogs',
self.opts.count).format(self.opts.count))
elif self.opts.upgrades:
logger.info(
_('Listing only new changelogs since installed version of the package'))
else:
logger.info(_('Listing all changelogs'))
by_srpm = self.by_srpm(self.query())
for name in by_srpm:
print(_('Changelogs for {}').format(
', '.join(sorted({str(pkg) for pkg in by_srpm[name]}))))
for chlog in self.filter_changelogs(by_srpm[name][0]):
print(self.base.format_changelog(chlog))
reposync.py 0000644 00000034470 15252533450 0006774 0 ustar 00 # reposync.py
# DNF plugin adding a command to download all packages from given remote repo.
#
# 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 unicode_literals
import hawkey
import os
import shutil
import types
from dnfpluginscore import _, logger
from dnf.cli.option_parser import OptionParser
import dnf
import dnf.cli
def _pkgdir(intermediate, target):
cwd = dnf.i18n.ucd(os.getcwd())
return os.path.realpath(os.path.join(cwd, intermediate, target))
class RPMPayloadLocation(dnf.repo.RPMPayload):
def __init__(self, pkg, progress, pkg_location):
super(RPMPayloadLocation, self).__init__(pkg, progress)
self.package_dir = os.path.dirname(pkg_location)
def _target_params(self):
tp = super(RPMPayloadLocation, self)._target_params()
dnf.util.ensure_dir(self.package_dir)
tp['dest'] = self.package_dir
return tp
@dnf.plugin.register_command
class RepoSyncCommand(dnf.cli.Command):
aliases = ('reposync',)
summary = _('download all packages from remote repo')
def __init__(self, cli):
super(RepoSyncCommand, self).__init__(cli)
@staticmethod
def set_argparser(parser):
parser.add_argument('-a', '--arch', dest='arches', default=[],
action=OptionParser._SplitCallback, metavar='[arch]',
help=_('download only packages for this ARCH'))
parser.add_argument('--delete', default=False, action='store_true',
help=_('delete local packages no longer present in repository'))
parser.add_argument('--download-metadata', default=False, action='store_true',
help=_('download all the metadata.'))
parser.add_argument('-g', '--gpgcheck', default=False, action='store_true',
help=_('Remove packages that fail GPG signature checking '
'after downloading'))
parser.add_argument('-m', '--downloadcomps', default=False, action='store_true',
help=_('also download and uncompress comps.xml'))
parser.add_argument('--metadata-path',
help=_('where to store downloaded repository metadata. '
'Defaults to the value of --download-path.'))
parser.add_argument('-n', '--newest-only', default=False, action='store_true',
help=_('download only newest packages per-repo'))
parser.add_argument('--norepopath', default=False, action='store_true',
help=_("Don't add the reponame to the download path."))
parser.add_argument('-p', '--download-path', default='./',
help=_('where to store downloaded repositories'))
parser.add_argument('--remote-time', default=False, action='store_true',
help=_('try to set local timestamps of local files by '
'the one on the server'))
parser.add_argument('--source', default=False, action='store_true',
help=_('download only source packages'))
parser.add_argument('-u', '--urls', default=False, action='store_true',
help=_("Just list urls of what would be downloaded, "
"don't download"))
def configure(self):
demands = self.cli.demands
demands.available_repos = True
demands.sack_activation = True
repos = self.base.repos
if self.opts.repo:
repos.all().disable()
for repoid in self.opts.repo:
try:
repo = repos[repoid]
except KeyError:
raise dnf.cli.CliError("Unknown repo: '%s'." % repoid)
repo.enable()
if self.opts.source:
repos.enable_source_repos()
if len(list(repos.iter_enabled())) > 1 and self.opts.norepopath:
raise dnf.cli.CliError(
_("Can't use --norepopath with multiple repositories"))
for repo in repos.iter_enabled():
repo._repo.expire()
repo.deltarpm = False
def run(self):
self.base.conf.keepcache = True
gpgcheck_ok = True
for repo in self.base.repos.iter_enabled():
if self.opts.remote_time:
repo._repo.setPreserveRemoteTime(True)
if self.opts.download_metadata:
if self.opts.urls:
for md_type, md_location in repo._repo.getMetadataLocations():
url = repo.remote_location(md_location)
if url:
print(url)
else:
msg = _("Failed to get mirror for metadata: %s") % md_type
logger.warning(msg)
else:
self.download_metadata(repo)
if self.opts.downloadcomps:
if self.opts.urls:
mdl = dict(repo._repo.getMetadataLocations())
group_locations = [mdl[md_type]
for md_type in ('group', 'group_gz', 'group_gz_zck')
if md_type in mdl]
if group_locations:
for group_location in group_locations:
url = repo.remote_location(group_location)
if url:
print(url)
break
else:
msg = _("Failed to get mirror for the group file.")
logger.warning(msg)
else:
self.getcomps(repo)
pkglist = self.get_pkglist(repo)
if self.opts.urls:
self.print_urls(pkglist)
else:
self.download_packages(pkglist)
if self.opts.gpgcheck:
for pkg in pkglist:
local_path = self.pkg_download_path(pkg)
# base.package_signature_check uses pkg.localPkg() to determine
# the location of the package rpm file on the disk.
# Set it to the correct download path.
pkg.localPkg = types.MethodType(
lambda s, local_path=local_path: local_path, pkg)
result, error = self.base.package_signature_check(pkg)
if result != 0:
logger.warning(_("Removing {}: {}").format(
os.path.basename(local_path), error))
os.unlink(local_path)
gpgcheck_ok = False
if self.opts.delete:
self.delete_old_local_packages(repo, pkglist)
if not gpgcheck_ok:
raise dnf.exceptions.Error(_("GPG signature check failed."))
def repo_target(self, repo):
return _pkgdir(self.opts.destdir or self.opts.download_path,
repo.id if not self.opts.norepopath else '')
def metadata_target(self, repo):
if self.opts.metadata_path:
return _pkgdir(self.opts.metadata_path, repo.id)
else:
return self.repo_target(repo)
def pkg_download_path(self, pkg):
repo_target = self.repo_target(pkg.repo)
pkg_download_path = os.path.realpath(
os.path.join(repo_target, pkg.location))
# join() ensures repo_target ends with a path separator (otherwise the
# check would pass if pkg_download_path was a "sibling" path component
# of repo_target that has the same prefix).
if not pkg_download_path.startswith(os.path.join(repo_target, '')):
raise dnf.exceptions.Error(
_("Download target '{}' is outside of download path '{}'.").format(
pkg_download_path, repo_target))
return pkg_download_path
def delete_old_local_packages(self, repo, pkglist):
# delete any *.rpm file under target path, that was not downloaded from repository
downloaded_files = set(self.pkg_download_path(pkg) for pkg in pkglist)
for dirpath, dirnames, filenames in os.walk(self.repo_target(repo)):
for filename in filenames:
path = os.path.join(dirpath, filename)
if filename.endswith('.rpm') and os.path.isfile(path):
if path not in downloaded_files:
# Delete disappeared or relocated file
try:
os.unlink(path)
logger.info(_("[DELETED] %s"), path)
except OSError:
logger.error(_("failed to delete file %s"), path)
def getcomps(self, repo):
comps_fn = repo._repo.getCompsFn()
if comps_fn:
dest_path = self.metadata_target(repo)
dnf.util.ensure_dir(dest_path)
dest = os.path.join(dest_path, 'comps.xml')
dnf.yum.misc.decompress(comps_fn, dest=dest)
logger.info(_("comps.xml for repository %s saved"), repo.id)
def download_metadata(self, repo):
repo_target = self.metadata_target(repo)
repo._repo.downloadMetadata(repo_target)
return True
def _get_latest(self, query):
"""
return union of these queries:
- the latest NEVRAs from non-modular packages
- all packages from stream version with the latest package NEVRA
(this should not be needed but the latest package NEVRAs might be
part of an older module version)
- all packages from the latest stream version
"""
if not dnf.base.WITH_MODULES:
return query.latest()
query.apply()
module_packages = self.base._moduleContainer.getModulePackages()
all_artifacts = set()
module_dict = {} # {NameStream: {Version: [modules]}}
artifact_version = {} # {artifact: {NameStream: [Version]}}
for module_package in module_packages:
artifacts = module_package.getArtifacts()
all_artifacts.update(artifacts)
module_dict.setdefault(module_package.getNameStream(), {}).setdefault(
module_package.getVersionNum(), []).append(module_package)
for artifact in artifacts:
artifact_version.setdefault(artifact, {}).setdefault(
module_package.getNameStream(), []).append(module_package.getVersionNum())
# the latest NEVRAs from non-modular packages
latest_query = query.filter(
pkg__neq=query.filter(nevra_strict=all_artifacts)).latest()
# artifacts from the newest version and those versions that contain an artifact
# with the highest NEVRA
latest_stream_artifacts = set()
for namestream, version_dict in module_dict.items():
# versions that will be synchronized
versions = set()
# add the newest stream version
versions.add(sorted(version_dict.keys(), reverse=True)[0])
# collect all artifacts in all stream versions
stream_artifacts = set()
for modules in version_dict.values():
for module in modules:
stream_artifacts.update(module.getArtifacts())
# find versions to which the packages with the highest NEVRAs belong
for latest_pkg in query.filter(nevra_strict=stream_artifacts).latest():
# here we depend on modules.yaml allways containing full NEVRA (including epoch)
nevra = "{0.name}-{0.epoch}:{0.version}-{0.release}.{0.arch}".format(latest_pkg)
# download only highest version containing the latest artifact
versions.add(max(artifact_version[nevra][namestream]))
# add all artifacts from selected versions for synchronization
for version in versions:
for module in version_dict[version]:
latest_stream_artifacts.update(module.getArtifacts())
latest_query = latest_query.union(query.filter(nevra_strict=latest_stream_artifacts))
return latest_query
def get_pkglist(self, repo):
query = self.base.sack.query(flags=hawkey.IGNORE_MODULAR_EXCLUDES).available().filterm(
reponame=repo.id)
if self.opts.newest_only:
query = self._get_latest(query)
if self.opts.source:
query.filterm(arch='src')
elif self.opts.arches:
query.filterm(arch=self.opts.arches)
return query
def download_packages(self, pkglist):
base = self.base
progress = base.output.progress
if progress is None:
progress = dnf.callback.NullDownloadProgress()
drpm = dnf.drpm.DeltaInfo(base.sack.query(flags=hawkey.IGNORE_MODULAR_EXCLUDES).installed(),
progress, 0)
payloads = [RPMPayloadLocation(pkg, progress, self.pkg_download_path(pkg))
for pkg in pkglist]
base._download_remote_payloads(payloads, drpm, progress, None, False)
def print_urls(self, pkglist):
for pkg in pkglist:
url = pkg.remote_location()
if url:
print(url)
else:
msg = _("Failed to get mirror for package: %s") % pkg.name
logger.warning(msg)
debug.py 0000644 00000030425 15252533450 0006214 0 ustar 00 #
# Copyright (C) 2015 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 ucd
from dnfpluginscore import _, logger
import dnf
import dnf.cli
import gzip
import hawkey
import os
import rpm
import sys
import time
DEBUG_VERSION = "dnf-debug-dump version 1\n"
class Debug(dnf.Plugin):
name = 'debug'
def __init__(self, base, cli):
super(Debug, self).__init__(base, cli)
self.base = base
self.cli = cli
if self.cli is not None:
self.cli.register_command(DebugDumpCommand)
self.cli.register_command(DebugRestoreCommand)
class DebugDumpCommand(dnf.cli.Command):
aliases = ("debug-dump",)
summary = _("dump information about installed rpm packages to file")
def __init__(self, cli):
super(DebugDumpCommand, self).__init__(cli)
self.dump_file = None
def configure(self):
self.cli.demands.sack_activation = True
self.cli.demands.available_repos = True
@staticmethod
def set_argparser(parser):
parser.add_argument(
"--norepos", action="store_true", default=False,
help=_("do not attempt to dump the repository contents."))
parser.add_argument(
"filename", nargs="?",
help=_("optional name of dump file"))
def run(self):
"""create debug txt file and compress it, if no filename specified
use dnf_debug_dump-.txt.gz by default"""
filename = self.opts.filename
if not filename:
now = time.strftime("%Y-%m-%d_%T", time.localtime(time.time()))
filename = "dnf_debug_dump-%s-%s.txt.gz" % (os.uname()[1], now)
filename = os.path.abspath(filename)
if filename.endswith(".gz"):
self.dump_file = gzip.GzipFile(filename, "w")
else:
self.dump_file = open(filename, "w")
self.write(DEBUG_VERSION)
self.dump_system_info()
self.dump_dnf_config_info()
self.dump_rpm_problems()
self.dump_packages(not self.opts.norepos)
self.dump_rpmdb_versions()
self.dump_file.close()
print(_("Output written to: %s") % filename)
def write(self, msg):
if dnf.pycomp.PY3 and isinstance(self.dump_file, gzip.GzipFile):
msg = bytes(msg, "utf8")
dnf.pycomp.write_to_file(self.dump_file, msg)
def dump_system_info(self):
self.write("%%%%SYSTEM INFO\n")
uname = os.uname()
self.write(" uname: %s, %s\n" % (uname[2], uname[4]))
self.write(" rpm ver: %s\n" % rpm.__version__)
self.write(" python ver: %s\n" % sys.version.replace("\n", ""))
return
def dump_dnf_config_info(self):
var = self.base.conf.substitutions
plugins = ",".join([p.name for p in self.base._plugins.plugins])
self.write("%%%%DNF INFO\n")
self.write(" arch: %s\n" % var["arch"])
self.write(" basearch: %s\n" % var["basearch"])
self.write(" releasever: %s\n" % var["releasever"])
self.write(" dnf ver: %s\n" % dnf.const.VERSION)
self.write(" enabled plugins: %s\n" % plugins)
self.write(" global excludes: %s\n" % ",".join(self.base.conf.excludepkgs))
return
def dump_rpm_problems(self):
self.write("%%%%RPMDB PROBLEMS\n")
(missing, conflicts) = rpm_problems(self.base)
self.write("".join(["Package %s requires %s\n" % (ucd(pkg), ucd(req))
for (req, pkg) in missing]))
self.write("".join(["Package %s conflicts with %s\n" % (ucd(pkg),
ucd(conf))
for (conf, pkg) in conflicts]))
def dump_packages(self, load_repos):
q = self.base.sack.query()
# packages from rpmdb
self.write("%%%%RPMDB\n")
for p in sorted(q.installed()):
self.write(" %s\n" % pkgspec(p))
if not load_repos:
return
self.write("%%%%REPOS\n")
available = q.available()
for repo in sorted(self.base.repos.iter_enabled(), key=lambda x: x.id):
try:
url = None
if repo.metalink is not None:
url = repo.metalink
elif repo.mirrorlist is not None:
url = repo.mirrorlist
elif len(repo.baseurl) > 0:
url = repo.baseurl[0]
self.write("%%%s - %s\n" % (repo.id, url))
self.write(" excludes: %s\n" % ",".join(repo.excludepkgs))
for po in sorted(available.filter(reponame=repo.id)):
self.write(" %s\n" % pkgspec(po))
except dnf.exceptions.Error as e:
self.write("Error accessing repo %s: %s\n" % (repo, str(e)))
continue
return
def dump_rpmdb_versions(self):
self.write("%%%%RPMDB VERSIONS\n")
version = self.base.sack._rpmdb_version()
self.write(" all: %s\n" % version)
return
class DebugRestoreCommand(dnf.cli.Command):
aliases = ("debug-restore",)
summary = _("restore packages recorded in debug-dump file")
def configure(self):
self.cli.demands.sack_activation = True
self.cli.demands.available_repos = True
self.cli.demands.root_user = True
if not self.opts.output:
self.cli.demands.resolving = True
@staticmethod
def set_argparser(parser):
parser.add_argument(
"--output", action="store_true",
help=_("output commands that would be run to stdout."))
parser.add_argument(
"--install-latest", action="store_true",
help=_("Install the latest version of recorded packages."))
parser.add_argument(
"--ignore-arch", action="store_true",
help=_("Ignore architecture and install missing packages matching "
"the name, epoch, version and release."))
parser.add_argument(
"--filter-types", metavar="[install, remove, replace]",
default="install, remove, replace",
help=_("limit to specified type"))
parser.add_argument(
"--remove-installonly", action="store_true",
help=_('Allow removing of install-only packages. Using this option may '
'result in an attempt to remove the running kernel.'))
parser.add_argument(
"filename", nargs=1, help=_("name of dump file"))
def run(self):
"""Execute the command action here."""
if self.opts.filter_types:
self.opts.filter_types = set(
self.opts.filter_types.replace(",", " ").split())
dump_pkgs = self.read_dump_file(self.opts.filename[0])
self.process_installed(dump_pkgs, self.opts)
self.process_dump(dump_pkgs, self.opts)
def process_installed(self, dump_pkgs, opts):
installed = self.base.sack.query().installed()
installonly_pkgs = self.base._get_installonly_query(installed)
for pkg in installed:
pkg_remove = False
spec = pkgspec(pkg)
dumped_versions = dump_pkgs.get((pkg.name, pkg.arch), None)
if dumped_versions is not None:
evr = (pkg.epoch, pkg.version, pkg.release)
if evr in dumped_versions:
# the correct version is already installed
dumped_versions[evr] = 'skip'
else:
# other version is currently installed
if pkg in installonly_pkgs:
# package is install-only, should be removed
pkg_remove = True
else:
# package should be upgraded / downgraded
if "replace" in opts.filter_types:
action = 'replace'
else:
action = 'skip'
for d_evr in dumped_versions.keys():
dumped_versions[d_evr] = action
else:
# package should not be installed
pkg_remove = True
if pkg_remove and "remove" in opts.filter_types:
if pkg not in installonly_pkgs or opts.remove_installonly:
if opts.output:
print("remove %s" % spec)
else:
self.base.package_remove(pkg)
def process_dump(self, dump_pkgs, opts):
for (n, a) in sorted(dump_pkgs.keys()):
dumped_versions = dump_pkgs[(n, a)]
for (e, v, r) in sorted(dumped_versions.keys()):
action = dumped_versions[(e, v, r)]
if action == 'skip':
continue
if opts.ignore_arch:
arch = ""
else:
arch = "." + a
if opts.install_latest and action == "install":
pkg_spec = "%s%s" % (n, arch)
else:
pkg_spec = pkgtup2spec(n, arch, e, v, r)
if action in opts.filter_types:
if opts.output:
print("%s %s" % (action, pkg_spec))
else:
try:
self.base.install(pkg_spec)
except dnf.exceptions.MarkingError:
logger.error(_("Package %s is not available"), pkg_spec)
@staticmethod
def read_dump_file(filename):
if filename.endswith(".gz"):
fobj = gzip.GzipFile(filename)
else:
fobj = open(filename)
if ucd(fobj.readline()) != DEBUG_VERSION:
logger.error(_("Bad dnf debug file: %s"), filename)
raise dnf.exceptions.Error
skip = True
pkgs = {}
for line in fobj:
line = ucd(line)
if skip:
if line == "%%%%RPMDB\n":
skip = False
continue
if not line or line[0] != " ":
break
pkg_spec = line.strip()
nevra = hawkey.split_nevra(pkg_spec)
# {(name, arch): {(epoch, version, release): action}}
pkgs.setdefault((nevra.name, nevra.arch), {})[
(nevra.epoch, nevra.version, nevra.release)] = "install"
return pkgs
def rpm_problems(base):
rpmdb = dnf.sack._rpmdb_sack(base)
allpkgs = rpmdb.query().installed()
requires = set()
conflicts = set()
for pkg in allpkgs:
requires.update([(req, pkg) for req in pkg.requires
if not str(req) == "solvable:prereqmarker"
and not str(req).startswith("rpmlib(")])
conflicts.update([(conf, pkg) for conf in pkg.conflicts])
missing_requires = [(req, pkg) for (req, pkg) in requires
if not allpkgs.filter(provides=req)]
existing_conflicts = [(conf, pkg) for (conf, pkg) in conflicts
if allpkgs.filter(provides=conf)]
return missing_requires, existing_conflicts
def pkgspec(pkg):
return pkgtup2spec(pkg.name, pkg.arch, pkg.epoch, pkg.version, pkg.release)
def pkgtup2spec(name, arch, epoch, version, release):
a = "" if not arch else ".%s" % arch.lstrip('.')
e = "" if epoch in (None, "") else "%s:" % epoch
return "%s-%s%s-%s%s" % (name, e, version, release, a)
debuginfo-install.py 0000644 00000025514 15252533450 0010537 0 ustar 00 # debuginfo-install.py
# Install the debuginfo of packages and their dependencies to debug this package.
#
# Copyright (C) 2014 Igor Gnatenko
# Copyright (C) 2014-2019 Red Hat
#
# 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 dnfpluginscore import _, logger
import dnf
from dnf.package import Package
class DebuginfoInstall(dnf.Plugin):
"""DNF plugin supplying the 'debuginfo-install' command."""
name = 'debuginfo-install'
def __init__(self, base, cli):
"""Initialize the plugin instance."""
super(DebuginfoInstall, self).__init__(base, cli)
self.base = base
self.cli = cli
if cli is not None:
cli.register_command(DebuginfoInstallCommand)
def config(self):
cp = self.read_config(self.base.conf)
autoupdate = (cp.has_section('main')
and cp.has_option('main', 'autoupdate')
and cp.getboolean('main', 'autoupdate'))
if autoupdate:
# allow update of already installed debuginfo packages
dbginfo = dnf.sack._rpmdb_sack(self.base).query().filterm(name__glob="*-debuginfo")
if len(dbginfo):
self.base.repos.enable_debug_repos()
class DebuginfoInstallCommand(dnf.cli.Command):
""" DebuginfoInstall plugin for DNF """
aliases = ("debuginfo-install",)
summary = _('install debuginfo packages')
def __init__(self, cli):
super(DebuginfoInstallCommand, self).__init__(cli)
self.available_debuginfo_missing = set()
self.available_debugsource_missing = set()
self.installed_debuginfo_missing = set()
self.installed_debugsource_missing = set()
@staticmethod
def set_argparser(parser):
parser.add_argument('package', nargs='+')
def configure(self):
demands = self.cli.demands
demands.resolving = True
demands.root_user = True
demands.sack_activation = True
demands.available_repos = True
self.base.repos.enable_debug_repos()
def run(self):
errors_spec = []
debuginfo_suffix_len = len(Package.DEBUGINFO_SUFFIX)
debugsource_suffix_len = len(Package.DEBUGSOURCE_SUFFIX)
for pkgspec in self.opts.package:
solution = dnf.subject.Subject(pkgspec).get_best_solution(self.base.sack,
with_src=False)
query = solution["query"]
if not query:
logger.info(_('No match for argument: %s'), self.base.output.term.bold(pkgspec))
errors_spec.append(pkgspec)
continue
package_dict = query.available()._name_dict()
# installed versions of packages have priority, replace / add them to the dict
package_dict.update(query.installed()._name_dict())
# Remove debuginfo packages if their base packages are in the query.
# They can get there through globs and they break the installation
# of debug packages with the same version as the installed base
# packages. If the base package of a debuginfo package is not in
# the query, the user specified a debug package on the command
# line. We don't want to ignore those, so we will install them.
# But, in this case the version will not be matched to the
# installed version of the base package, as that would require
# another query and is further complicated if the user specifies a
# version themselves etc.
for name in list(package_dict.keys()):
if name.endswith(Package.DEBUGINFO_SUFFIX):
if name[:-debuginfo_suffix_len] in package_dict:
package_dict.pop(name)
if name.endswith(Package.DEBUGSOURCE_SUFFIX):
if name[:-debugsource_suffix_len] in package_dict:
package_dict.pop(name)
# attempt to install debuginfo and debugsource for the highest
# listed version of the package (in case the package is installed,
# only the installed version is listed)
for pkgs in package_dict.values():
first_pkg = pkgs[0]
# for packages from system (installed) there can be more
# packages with different architectures listed and we want to
# install debuginfo for all of them
if first_pkg._from_system:
# we need to split them by architectures and install the
# latest version for each architecture
arch_dict = {}
for pkg in pkgs:
arch_dict.setdefault(pkg.arch, []).append(pkg)
for package_arch_list in arch_dict.values():
pkg = package_arch_list[0]
if not self._install_debug_from_system(pkg.debug_name, pkg):
if not self._install_debug_from_system(pkg.source_debug_name, pkg):
self.installed_debuginfo_missing.add(str(pkg))
if not self._install_debug_from_system(pkg.debugsource_name, pkg):
self.installed_debugsource_missing.add(str(pkg))
continue
# if the package in question is -debuginfo or -debugsource, install it directly
if first_pkg.name.endswith(Package.DEBUGINFO_SUFFIX) \
or first_pkg.name.endswith(Package.DEBUGSOURCE_SUFFIX):
self._install(pkgs) # pass all pkgs to the solver, it will pick the best one
continue
# if we have NEVRA parsed from the pkgspec, use it to install the package
if solution["nevra"] is not None:
if not self._install_debug(first_pkg.debug_name, solution["nevra"]):
if not self._install_debug(first_pkg.source_debug_name, solution["nevra"]):
self.available_debuginfo_missing.add(
"{}-{}".format(first_pkg.name, first_pkg.evr))
if not self._install_debug(first_pkg.debugsource_name, solution["nevra"]):
self.available_debugsource_missing.add(
"{}-{}".format(first_pkg.name, first_pkg.evr))
continue
# if we don't have NEVRA from the pkgspec, pass nevras from
# all packages that were found (while replacing the name with
# the -debuginfo and -debugsource variant) to the solver, which
# will pick the correct version and architecture
if not self._install_debug_no_nevra(first_pkg.debug_name, pkgs):
if not self._install_debug_no_nevra(first_pkg.source_debug_name, pkgs):
self.available_debuginfo_missing.add(
"{}-{}".format(first_pkg.name, first_pkg.evr))
if not self._install_debug_no_nevra(first_pkg.debugsource_name, pkgs):
self.available_debugsource_missing.add(
"{}-{}".format(first_pkg.name, first_pkg.evr))
if self.available_debuginfo_missing:
logger.info(
_("Could not find debuginfo package for the following available packages: %s"),
", ".join(sorted(self.available_debuginfo_missing)))
if self.available_debugsource_missing:
logger.info(
_("Could not find debugsource package for the following available packages: %s"),
", ".join(sorted(self.available_debugsource_missing)))
if self.installed_debuginfo_missing:
logger.info(
_("Could not find debuginfo package for the following installed packages: %s"),
", ".join(sorted(self.installed_debuginfo_missing)))
if self.installed_debugsource_missing:
logger.info(
_("Could not find debugsource package for the following installed packages: %s"),
", ".join(sorted(self.installed_debugsource_missing)))
if errors_spec and self.base.conf.strict:
raise dnf.exceptions.PackagesNotAvailableError(_("Unable to find a match"),
pkg_spec=' '.join(errors_spec))
def _install_debug_from_system(self, debug_name, pkg):
query = self.base.sack.query().filter(name=debug_name,
epoch=pkg.epoch,
version=pkg.version,
release=pkg.release,
arch=pkg.arch)
if query:
self._install(query)
return True
return False
def _install_debug(self, debug_name, base_nevra):
kwargs = {}
# if some part of EVRA was specified in the argument, add it as a filter
if base_nevra.epoch is not None:
kwargs["epoch__glob"] = base_nevra.epoch
if base_nevra.version is not None:
kwargs["version__glob"] = base_nevra.version
if base_nevra.release is not None:
kwargs["release__glob"] = base_nevra.release
if base_nevra.arch is not None:
kwargs["arch__glob"] = base_nevra.arch
query = self.base.sack.query().filter(name=debug_name, **kwargs)
if query:
self._install(query)
return True
return False
def _install_debug_no_nevra(self, debug_name, pkgs):
query = self.base.sack.query().filterm(
nevra_strict=["{}-{}.{}".format(debug_name, p.evr, p.arch) for p in pkgs])
if query:
self._install(query)
return True
return False
def _install(self, pkgs):
selector = dnf.selector.Selector(self.base.sack)
selector.set(pkg=pkgs)
self.base.goal.install(select=selector, optional=not self.base.conf.strict)
repomanage.py 0000644 00000024512 15252533450 0007244 0 ustar 00 # repomanage.py
# DNF plugin adding a command to manage rpm packages from given directory.
#
# Copyright (C) 2015 Igor Gnatenko
#
# 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 dnfpluginscore import _, logger
import dnf
import dnf.cli
import logging
import os
import hawkey
class RepoManage(dnf.Plugin):
name = "repomanage"
def __init__(self, base, cli):
super(RepoManage, self).__init__(base, cli)
if cli is None:
return
cli.register_command(RepoManageCommand)
class RepoManageCommand(dnf.cli.Command):
aliases = ("repomanage",)
summary = _("Manage a directory of rpm packages")
def pre_configure(self):
if not self.opts.verbose and not self.opts.quiet:
self.cli.redirect_logger(stdout=logging.WARNING, stderr=logging.INFO)
def configure(self):
if not self.opts.verbose and not self.opts.quiet:
self.cli.redirect_repo_progress()
demands = self.cli.demands
demands.sack_activation = True
def run(self):
if self.opts.new and self.opts.old:
raise dnf.exceptions.Error(_("Pass either --old or --new, not both!"))
if self.opts.new and self.opts.oldonly:
raise dnf.exceptions.Error(_("Pass either --oldonly or --new, not both!"))
if self.opts.old and self.opts.oldonly:
raise dnf.exceptions.Error(_("Pass either --old or --oldonly, not both!"))
if not self.opts.old and not self.opts.oldonly:
self.opts.new = True
verfile = {}
pkgdict = {}
module_dict = {} # {NameStream: {Version: [modules]}}
all_modular_artifacts = set()
keepnum = int(self.opts.keep) # the number of items to keep
try:
REPOMANAGE_REPOID = "repomanage_repo"
repo_conf = self.base.repos.add_new_repo(REPOMANAGE_REPOID, self.base.conf, baseurl=[self.opts.path])
# Always expire the repo, otherwise repomanage could use cached metadata and give identical results
# for multiple runs even if the actual repo changed in the meantime
repo_conf._repo.expire()
self.base._add_repo_to_sack(repo_conf)
if dnf.base.WITH_MODULES:
self.base._setup_modular_excludes()
# Prepare modules
module_packages = self.base._moduleContainer.getModulePackages()
for module_package in module_packages:
# Even though we load only REPOMANAGE_REPOID other modules can be loaded from system
# failsafe data automatically, we don't want them affecting repomanage results so ONLY
# use modules from REPOMANAGE_REPOID.
if module_package.getRepoID() == REPOMANAGE_REPOID:
all_modular_artifacts.update(module_package.getArtifacts())
module_dict.setdefault(module_package.getNameStream(), {}).setdefault(
module_package.getVersionNum(), []).append(module_package)
except dnf.exceptions.RepoError:
rpm_list = []
rpm_list = self._get_file_list(self.opts.path, ".rpm")
if len(rpm_list) == 0:
raise dnf.exceptions.Error(_("No files to process"))
self.base.reset(sack=True, repos=True)
self.base.fill_sack(load_system_repo=False, load_available_repos=False)
try:
self.base.add_remote_rpms(rpm_list, progress=self.base.output.progress)
except IOError:
logger.warning(_("Could not open {}").format(', '.join(rpm_list)))
# Prepare regular packages
query = self.base.sack.query(flags=hawkey.IGNORE_MODULAR_EXCLUDES).available()
packages = [x for x in query.filter(pkg__neq=query.filter(nevra_strict=all_modular_artifacts)).available()]
packages.sort()
for pkg in packages:
na = (pkg.name, pkg.arch)
if na in pkgdict:
if pkg not in pkgdict[na]:
pkgdict[na].append(pkg)
else:
pkgdict[na] = [pkg]
nevra = self._package_to_nevra(pkg)
if nevra in verfile:
verfile[nevra].append(self._package_to_path(pkg))
else:
verfile[nevra] = [self._package_to_path(pkg)]
outputpackages = []
# modular packages
keepnum_latest_stream_artifacts = set()
if self.opts.new:
# regular packages
for (n, a) in pkgdict.keys():
evrlist = pkgdict[(n, a)]
newevrs = evrlist[-keepnum:]
for package in newevrs:
nevra = self._package_to_nevra(package)
for fpkg in verfile[nevra]:
outputpackages.append(fpkg)
# modular packages
for streams_by_version in module_dict.values():
sorted_stream_versions = sorted(streams_by_version.keys())
new_sorted_stream_versions = sorted_stream_versions[-keepnum:]
for i in new_sorted_stream_versions:
for stream in streams_by_version[i]:
keepnum_latest_stream_artifacts.update(set(stream.getArtifacts()))
if self.opts.old:
# regular packages
for (n, a) in pkgdict.keys():
evrlist = pkgdict[(n, a)]
oldevrs = evrlist[:-keepnum]
for package in oldevrs:
nevra = self._package_to_nevra(package)
for fpkg in verfile[nevra]:
outputpackages.append(fpkg)
# modular packages
for streams_by_version in module_dict.values():
sorted_stream_versions = sorted(streams_by_version.keys())
old_sorted_stream_versions = sorted_stream_versions[:-keepnum]
for i in old_sorted_stream_versions:
for stream in streams_by_version[i]:
keepnum_latest_stream_artifacts.update(set(stream.getArtifacts()))
if self.opts.oldonly:
# regular packages
for (n, a) in pkgdict.keys():
evrlist = pkgdict[(n, a)]
oldevrs = evrlist[:-keepnum]
for package in oldevrs:
nevra = self._package_to_nevra(package)
for fpkg in verfile[nevra]:
outputpackages.append(fpkg)
# modular packages
keepnum_newer_stream_artifacts = set()
for streams_by_version in module_dict.values():
sorted_stream_versions = sorted(streams_by_version.keys())
new_sorted_stream_versions = sorted_stream_versions[-keepnum:]
for i in new_sorted_stream_versions:
for stream in streams_by_version[i]:
keepnum_newer_stream_artifacts.update(set(stream.getArtifacts()))
for streams_by_version in module_dict.values():
sorted_stream_versions = sorted(streams_by_version.keys())
old_sorted_stream_versions = sorted_stream_versions[:-keepnum]
for i in old_sorted_stream_versions:
for stream in streams_by_version[i]:
for artifact in stream.getArtifacts():
if artifact not in keepnum_newer_stream_artifacts:
keepnum_latest_stream_artifacts.add(artifact)
modular_packages = [self._package_to_path(x) for x in query.filter(pkg__eq=query.filter(nevra_strict=keepnum_latest_stream_artifacts)).available()]
outputpackages = outputpackages + modular_packages
outputpackages.sort()
if self.opts.space:
print(" ".join(outputpackages))
else:
for pkg in outputpackages:
print(pkg)
@staticmethod
def set_argparser(parser):
parser.add_argument("-o", "--old", action="store_true",
help=_("Print the older packages"))
parser.add_argument("-O", "--oldonly", action="store_true",
help=_("Print the older packages. Exclude the newest packages."))
parser.add_argument("-n", "--new", action="store_true",
help=_("Print the newest packages"))
parser.add_argument("-s", "--space", action="store_true",
help=_("Space separated output, not newline"))
parser.add_argument("-k", "--keep", action="store", metavar="KEEP",
help=_("Newest N packages to keep - defaults to 1"),
default=1, type=int)
parser.add_argument("path", action="store",
help=_("Path to directory"))
@staticmethod
def _get_file_list(path, ext):
"""Return all files in path matching ext
return list object
"""
filelist = []
for root, dirs, files in os.walk(path):
for f in files:
if os.path.splitext(f)[1].lower() == str(ext):
filelist.append(os.path.join(root, f))
return filelist
def _package_to_path(self, pkg):
if len(self.base.repos):
return os.path.join(self.opts.path, pkg.location)
else:
return pkg.location
@staticmethod
def _package_to_nevra(pkg):
return (pkg.name, pkg.epoch, pkg.version, pkg.release, pkg.arch)
builddep.py 0000644 00000022202 15252533450 0006710 0 ustar 00 # builddep.py
# Install all the deps needed to build this package.
#
# Copyright (C) 2013-2015 Red Hat, Inc.
# Copyright (C) 2015 Igor Gnatenko
#
# 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 dnfpluginscore import _, logger
import argparse
import dnf
import dnf.cli
import dnf.exceptions
import dnf.rpm.transaction
import dnf.yum.rpmtrans
import libdnf.repo
import os
import rpm
import shutil
import tempfile
@dnf.plugin.register_command
class BuildDepCommand(dnf.cli.Command):
aliases = ('builddep', 'build-dep')
msg = "Install build dependencies for package or spec file"
summary = _(msg)
usage = _("[PACKAGE|PACKAGE.spec]")
def __init__(self, cli):
super(BuildDepCommand, self).__init__(cli)
self._rpm_ts = dnf.rpm.transaction.initReadOnlyTransaction()
self.tempdirs = []
def __del__(self):
for temp_dir in self.tempdirs:
shutil.rmtree(temp_dir)
def _download_remote_file(self, pkgspec):
"""
In case pkgspec is a remote URL, download it to a temporary location
and use the temporary file instead.
"""
location = dnf.pycomp.urlparse.urlparse(pkgspec)
if location[0] in ('file', ''):
# just strip the file:// prefix
return location.path
downloader = libdnf.repo.Downloader()
temp_dir = tempfile.mkdtemp(prefix="dnf_builddep_")
temp_file = os.path.join(temp_dir, os.path.basename(pkgspec))
self.tempdirs.append(temp_dir)
temp_fo = open(temp_file, "wb+")
try:
downloader.downloadURL(self.base.conf._config, pkgspec, temp_fo.fileno())
except RuntimeError as ex:
raise
finally:
temp_fo.close()
return temp_file
@staticmethod
def set_argparser(parser):
def macro_def(arg):
arglist = arg.split(None, 1) if arg else []
if len(arglist) < 2:
msg = _("'%s' is not of the format 'MACRO EXPR'") % arg
raise argparse.ArgumentTypeError(msg)
return arglist
parser.add_argument('packages', nargs='+', metavar='package',
help=_('packages with builddeps to install'))
parser.add_argument('-D', '--define', action='append', default=[],
metavar="'MACRO EXPR'", type=macro_def,
help=_('define a macro for spec file parsing'))
parser.add_argument('--skip-unavailable', action='store_true', default=False,
help=_('skip build dependencies not available in repositories'))
ptype = parser.add_mutually_exclusive_group()
ptype.add_argument('--spec', action='store_true',
help=_('treat commandline arguments as spec files'))
ptype.add_argument('--srpm', action='store_true',
help=_('treat commandline arguments as source rpm'))
def pre_configure(self):
if not self.opts.rpmverbosity:
self.opts.rpmverbosity = 'error'
def configure(self):
demands = self.cli.demands
demands.available_repos = True
demands.resolving = True
demands.root_user = True
demands.sack_activation = True
# enable source repos only if needed
if not (self.opts.spec or self.opts.srpm):
for pkgspec in self.opts.packages:
if not (pkgspec.endswith('.src.rpm')
or pkgspec.endswith('.nosrc.rpm')
or pkgspec.endswith('.spec')):
self.base.repos.enable_source_repos()
break
def run(self):
rpmlog = dnf.yum.rpmtrans.RPMTransaction(self.base)
# Push user-supplied macro definitions for spec parsing
for macro in self.opts.define:
rpm.addMacro(macro[0], macro[1])
pkg_errors = False
for pkgspec in self.opts.packages:
pkgspec = self._download_remote_file(pkgspec)
try:
if self.opts.srpm:
self._src_deps(pkgspec)
elif self.opts.spec:
self._spec_deps(pkgspec)
elif pkgspec.endswith('.src.rpm') or pkgspec.endswith('nosrc.rpm'):
self._src_deps(pkgspec)
elif pkgspec.endswith('.spec'):
self._spec_deps(pkgspec)
else:
self._remote_deps(pkgspec)
except dnf.exceptions.Error as e:
for line in rpmlog.messages():
logger.error(_("RPM: {}").format(line))
logger.error(e)
pkg_errors = True
# Pop user macros so they don't affect future rpm calls
for macro in self.opts.define:
rpm.delMacro(macro[0])
if pkg_errors:
raise dnf.exceptions.Error(_("Some packages could not be found."))
@staticmethod
def _rpm_dep2reldep_str(rpm_dep):
return rpm_dep.DNEVR()[2:]
def _install(self, reldep_str):
# Try to find something by provides
sltr = dnf.selector.Selector(self.base.sack)
sltr.set(provides=reldep_str)
found = sltr.matches()
if not found and reldep_str.startswith("/"):
# Nothing matches by provides and since it's file, try by files
sltr = dnf.selector.Selector(self.base.sack)
sltr.set(file=reldep_str)
found = sltr.matches()
if not found and not reldep_str.startswith("("):
# No provides, no files
# Richdeps can have no matches but it could be correct (solver must decide later)
msg = _("No matching package to install: '%s'")
logger.warning(msg, reldep_str)
return self.opts.skip_unavailable is True
if found:
already_inst = self.base._sltr_matches_installed(sltr)
if already_inst:
for package in already_inst:
dnf.base._msg_installed(package)
self.base._goal.install(select=sltr, optional=False)
return True
def _src_deps(self, src_fn):
fd = os.open(src_fn, os.O_RDONLY)
try:
h = self._rpm_ts.hdrFromFdno(fd)
except rpm.error as e:
if str(e) == 'error reading package header':
e = _("Failed to open: '%s', not a valid source rpm file.") % src_fn
os.close(fd)
raise dnf.exceptions.Error(e)
os.close(fd)
ds = h.dsFromHeader('requirename')
done = True
for dep in ds:
reldep_str = self._rpm_dep2reldep_str(dep)
if reldep_str.startswith('rpmlib('):
continue
done &= self._install(reldep_str)
if not done:
err = _("Not all dependencies satisfied")
raise dnf.exceptions.Error(err)
if self.opts.define:
logger.warning(_("Warning: -D or --define arguments have no meaning "
"for source rpm packages."))
def _spec_deps(self, spec_fn):
try:
spec = rpm.spec(spec_fn)
except ValueError as ex:
msg = _("Failed to open: '%s', not a valid spec file: %s") % (
spec_fn, ex)
raise dnf.exceptions.Error(msg)
done = True
for dep in rpm.ds(spec.sourceHeader, 'requires'):
reldep_str = self._rpm_dep2reldep_str(dep)
done &= self._install(reldep_str)
if not done:
err = _("Not all dependencies satisfied")
raise dnf.exceptions.Error(err)
def _remote_deps(self, package):
available = dnf.subject.Subject(package).get_best_query(
self.base.sack).filter(arch__neq="src")
sourcenames = list({pkg.source_name for pkg in available})
pkgs = self.base.sack.query().available().filter(
name=(sourcenames + [package]), arch="src").latest().run()
if not pkgs:
raise dnf.exceptions.Error(_('no package matched: %s') % package)
done = True
for pkg in pkgs:
for req in pkg.requires:
done &= self._install(str(req))
if not done:
err = _("Not all dependencies satisfied")
raise dnf.exceptions.Error(err)
__pycache__/repograph.cpython-36.pyc 0000644 00000005342 15252533450 0013401 0 ustar 00 3
gt` @ s^ d dl mZ d dl mZ d dlmZmZ d dlZdZG dd dej Z
G dd d ejjZ
dS )
)absolute_import)unicode_literals)_loggerNzY
size="20.69,25.52";
ratio="fill";
rankdir="TB";
orientation=port;
node[style="filled"];
c s e Zd ZdZ fddZ ZS ) RepoGraph repographc s, t t| j|| |d krd S |jt d S )N)superr __init__Zregister_commandRepoGraphCommand)selfbasecli) __class__ /usr/lib/python3.6/repograph.pyr ) s zRepoGraph.__init__)__name__
__module____qualname__namer
__classcell__r r )r r r % s r c @ s<