[ PARENT ]

EDIT :: up2dateUtils.py

# Client code for Update Agent # Copyright (c) 1999--2018 Red Hat, Inc. Distributed under GPLv2. # # Author: Preston Brown <pbrown@redhat.com> # Adrian Likins <alikins@redhat.com> # """utility functions for up2date""" import contextlib import os import sys import re import gettext from up2date_client import up2dateErrors from up2date_client import config from up2date_client.pkgplatform import getPlatform from rhn.i18n import sstr 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 if getPlatform() == 'deb': import lsb_release def _getOSVersionAndRelease(): dist_info = lsb_release.get_distro_information() os_name = dist_info['ID'] os_version = 'n/a' if 'CODENAME' in dist_info: os_version = dist_info['CODENAME'] os_release = dist_info['RELEASE'] return os_name, os_version, os_release else: from up2date_client import transaction def _getOSVersionAndRelease(): ts = transaction.initReadOnlyTransaction() for h in ts.dbMatch('Providename', "oraclelinux-release"): SYSRELVER = 'system-release(releasever)' version = sstr(h['version']) release = sstr(h['release']) if SYSRELVER in (sstr(provide) for provide in h['providename']): provides = dict((sstr(n), sstr(v)) for n,v in zip(h['providename'], h['provideversion'])) release = '%s-%s' % (version, release) version = provides[SYSRELVER] osVersionRelease = (sstr(h['name']), version, release) return osVersionRelease else: for h in ts.dbMatch('Providename', "redhat-release"): SYSRELVER = 'system-release(releasever)' version = sstr(h['version']) release = sstr(h['release']) if SYSRELVER in (sstr(provide) for provide in h['providename']): provides = dict((sstr(n), sstr(v)) for n,v in zip(h['providename'], h['provideversion'])) release = '%s-%s' % (version, release) version = provides[SYSRELVER] osVersionRelease = (sstr(h['name']), version, release) return osVersionRelease else: for h in ts.dbMatch('Providename', "distribution-release"): osVersionRelease = (sstr(h['name']), sstr(h['version']), sstr(h['release'])) # zypper requires a exclusive lock on the rpmdb. So we need # to close it here. ts.ts.closeDB() return osVersionRelease else: raise up2dateErrors.RpmError( "Could not determine what version of CloudLinux you "\ "are running.\nIf you get this error, try running \n\n"\ "\t\trpm --rebuilddb\n\n") def getVersion(): ''' Returns the version of redhat-release rpm ''' cfg = config.initUp2dateConfig() if cfg["versionOverride"]: return str(cfg["versionOverride"]) os_release, version, release = _getOSVersionAndRelease() return version def getOSRelease(): ''' Returns the name of the redhat-release rpm ''' os_release, version, release = _getOSVersionAndRelease() return os_release def getRelease(): ''' Returns the release of the redhat-release rpm ''' os_release, version, release = _getOSVersionAndRelease() return release def getArch(): if os.access("/etc/rpm/platform", os.R_OK): fd = open("/etc/rpm/platform", "r") platform = fd.read().strip() #bz 216225 #handle some replacements.. replace = {"ia32e-redhat-linux": "x86_64-redhat-linux"} if platform in replace: platform = replace[platform] return platform arch = os.uname()[4] if getPlatform() == 'deb': # On debian we only support i386 if arch in ['i486', 'i586', 'i686']: arch = 'i386' if arch == 'x86_64': arch = 'amd64' arch += '-debian-linux' return arch def version(): # substituted to the real version by the Makefile at installation time. return "2.12.5-1.module_el8.10.0+6935+f9aadf00.cloudlinux" @contextlib.contextmanager def suppress_errors(error_patterns): ''' Context manager to suppress errors matching the specified patterns ''' read_end, write_end = os.pipe() old_stdout = os.dup(1) old_stderr = os.dup(2) os.dup2(write_end, 1) os.dup2(write_end, 2) try: yield finally: # Restore stdout and stderr os.dup2(old_stdout, 1) os.dup2(old_stderr, 2) os.close(write_end) with os.fdopen(read_end) as f: combined_pattern = re.compile('|'.join(error_patterns)) for line in f: if not combined_pattern.search(line): print(line, file=sys.stderr)

SAVE CANCEL