JFIF$        dd7 

Viewing File: /usr/lib/python3.9/site-packages/up2date_client/rpmUtils.py

# some high level utility stuff for rpm handling

# Client code for Update Agent
# Copyright (c) 1999--2016 Red Hat, Inc.  Distributed under GPLv2.
#
# Author: Preston Brown <pbrown@redhat.com>
#         Adrian Likins <alikins@redhat.com>
#


#
#  FIXME: Some exceptions in here are currently in up2date.py
#         fix by moving to up2dateErrors.py and importing from there
#
#

import os
import rpm
from rhn.i18n import sstr
from up2date_client import transaction

import gettext
t = gettext.translation('rhn-client-tools', fallback=True)
# Python 3 translations don't have a ugettext method
if not hasattr(t, 'ugettext'):
    t.ugettext = t.gettext
_ = t.ugettext

def installedHeaderByKeyword(**kwargs):
    """ just cause this is such a potentially useful looking method... """
    _ts = transaction.initReadOnlyTransaction()
    mi = _ts.dbMatch()
    for keyword in kwargs.keys():
        mi.pattern(keyword, rpm.RPMMIRE_GLOB, kwargs[keyword])
        # we really shouldnt be getting multiples here, but what the heck
    headerList = []
    for h in mi:
        headerList.append(h)

    return headerList

def verifyPackages(packages):
    """ given a list of package labels, run rpm -V on them
        and return a dict keyed off that data
    """
    data = {}
    missing_packages = []
    # data structure is keyed off package
    # label, with value being an array of each
    # line of the output from -V


    retlist = []
    for package in packages:
        # we have to have at least name...

        # Note: we cant reliable match on epoch, so just
        # skip it... two packages that only diff by epoch is
        # way broken anyway
        keywords = {'name': package[0],
                    'version': package[1],
                    'release': package[2],
                    # we left our epoch
                    'arch': package[4]
                    }
        for key in (keywords.keys()):
            if (keywords[key] == None) or (keywords[key] == ""):
                del(keywords[key])

        headers = installedHeaderByKeyword(**keywords)
        if len(headers) == 0:
            missing_packages.append(package)

        for header in headers:
            epoch = header['epoch']
            if epoch == None:
                epoch = ""
            # gpg-pubkey "packages" can have an arch of None, see bz #162701
            arch = header["arch"]
            if arch == None:
                arch = ""

            pkg = (header['name'], header['version'],
                   header['release'], epoch,
                   arch)

            # dont include arch in the label if it's a None arch, #162701
            if header["arch"] == "":
                packageLabel = "%s-%s-%s" % (pkg[0], pkg[1], pkg[2])
            else:
                packageLabel = "%s-%s-%s.%s" % (pkg[0], pkg[1], pkg[2], pkg[4])

            verifystring = "/usr/bin/rpmverify -V %s" % packageLabel

            fd = os.popen(verifystring)
            res = fd.readlines()
            fd.close()

            reslist = []
            for line in res:
                reslist.append(line.strip())
            retlist.append([pkg, reslist])

    return retlist, missing_packages

#FIXME: this looks like a good candidate for caching, since it takes a second
# or two to run, and I can call it a couple of times
def getInstalledPackageList(msgCallback = None, progressCallback = None,
                            getArch=None, getInfo = None):
    """ Return list of packages. Package is hash with keys name, epoch,
        version, release and optionaly arch and cookie
    """
    pkg_list = []

    if msgCallback != None:
        msgCallback(_("Getting list of packages installed on the system"))

    _ts = transaction.initReadOnlyTransaction()
    count = 0
    total = 0

    for h in _ts.dbMatch():
        if h == None:
            break
        count = count + 1

    total = count

    count = 0
    for h in _ts.dbMatch():
        if h == None:
            break
        package = {
            'name': sstr(h['name']),
            'epoch': h['epoch'],
            'version': sstr(h['version']),
            'release': sstr(h['release']),
            'installtime': h['installtime']
        }
        if package['epoch'] == None:
            package['epoch'] = ""
        else: # convert it to string
            package['epoch'] = "%s" % package['epoch']
        if getArch:
            package['arch'] = h['arch']
            # the arch on gpg-pubkeys is "None"...
            if package['arch']:
                package['arch'] = sstr(package['arch'])
                pkg_list.append(package)
        elif getInfo:
            if h['arch']:
                package['arch'] = sstr(h['arch'])
            if h['cookie']:
                package['cookie'] = sstr(h['cookie'])
            pkg_list.append(package)
        else:
            pkg_list.append(package)

        if progressCallback != None:
            progressCallback(count, total)
        count = count + 1

    pkg_list.sort(key=lambda x:(x['name'], x['epoch'], x['version'], x['release']))
    return pkg_list

def setDebugVerbosity():
    """Set rpm's verbosity mode
    """
    try:
        rpm.setVerbosity(rpm.RPMLOG_DEBUG)
    except AttributeError:
        print("extra verbosity not supported in this version of rpm")
Back to Directory  nL+D550H?Mx ,D"v]qv;6*Zqn)ZP0!1 A "#a$2Qr D8 a Ri[f\mIykIw0cuFcRı?lO7к_f˓[C$殷WF<_W ԣsKcëIzyQy/_LKℂ;C",pFA:/]=H  ~,ls/9ć:[=/#f;)x{ٛEQ )~ =𘙲r*2~ a _V=' kumFD}KYYC)({ *g&f`툪ry`=^cJ.I](*`wq1dđ#̩͑0;H]u搂@:~וKL Nsh}OIR*8:2 !lDJVo(3=M(zȰ+i*NAr6KnSl)!JJӁ* %݉?|D}d5:eP0R;{$X'xF@.ÊB {,WJuQɲRI;9QE琯62fT.DUJ;*cP A\ILNj!J۱+O\͔]ޒS߼Jȧc%ANolՎprULZԛerE2=XDXgVQeӓk yP7U*omQIs,K`)6\G3t?pgjrmۛجwluGtfh9uyP0D;Uڽ"OXlif$)&|ML0Zrm1[HXPlPR0'G=i2N+0e2]]9VTPO׮7h(F*癈'=QVZDF,d߬~TX G[`le69CR(!S2!P <0x<!1AQ "Raq02Br#SCTb ?Ζ"]mH5WR7k.ۛ!}Q~+yԏz|@T20S~Kek *zFf^2X*(@8r?CIuI|֓>^ExLgNUY+{.RѪ τV׸YTD I62'8Y27'\TP.6d&˦@Vqi|8-OΕ]ʔ U=TL8=;6c| !qfF3aů&~$l}'NWUs$Uk^SV:U# 6w++s&r+nڐ{@29 gL u"TÙM=6(^"7r}=6YݾlCuhquympǦ GjhsǜNlɻ}o7#S6aw4!OSrD57%|?x>L |/nD6?/8w#[)L7+6〼T ATg!%5MmZ/c-{1_Je"|^$'O&ޱմTrb$w)R$& N1EtdU3Uȉ1pM"N*(DNyd96.(jQ)X 5cQɎMyW?Q*!R>6=7)Xj5`J]e8%t!+'!1Q5 !1 AQaqё#2"0BRb?Gt^## .llQT $v,,m㵜5ubV =sY+@d{N! dnO<.-B;_wJt6;QJd.Qc%p{ 1,sNDdFHI0ГoXшe黅XۢF:)[FGXƹ/w_cMeD,ʡcc.WDtA$j@:) -# u c1<@ۗ9F)KJ-hpP]_x[qBlbpʖw q"LFGdƶ*s+ډ_Zc"?%t[IP 6J]#=ɺVvvCGsGh1 >)6|ey?Lӣm,4GWUi`]uJVoVDG< SB6ϏQ@ TiUlyOU0kfV~~}SZ@*WUUi##; s/[=!7}"WN]'(L! ~y5g9T̅JkbM' +s:S +B)v@Mj e Cf jE 0Y\QnzG1д~Wo{T9?`Rmyhsy3!HAD]mc1~2LSu7xT;j$`}4->L#vzŏILS ֭T{rjGKC;bpU=-`BsK.SFw4Mq]ZdHS0)tLg