# This file is a portion of the Red Hat Network Panel Applet # # Copyright (C) 1999-2002 Red Hat, Inc. All Rights Reserved. # Distributed under GPL version 2. # # Author: Chip Turner # # $Id: rhn_applet_model.py,v 1.24 2004/01/09 15:54:29 veillard Exp $ import string import os import time import gettext import rhn_utils from rhn_applet_rpm import rhnAppletRPM from rhn_applet_rpc import rhnAppletRPC from rhn_applet_yum import rhnAppletYum from rhn_applet_apt import rhnAppletApt from rhn_utils import rhnAppletException from rhn_sources import getSources, changedSources GETTEXT_DOMAIN = "rhn-applet" gettext.bindtextdomain (GETTEXT_DOMAIN, "/usr/share/locale") gettext.textdomain (GETTEXT_DOMAIN) try: gettext.install(GETTEXT_DOMAIN, "/usr/share/locale", 1) except IOError: import __builtin__ __builtin__.__dict__['_'] = unicode except AttributeError: # older gettext modules don't have an install() method from gettext import _ class rhnAppletModel: def __init__(self, refresh_callback, url, uuid, mode = "gui", cert = None): self.__refresh_callback__ = refresh_callback self.__url__ = url self.__uuid__ = uuid self.__proxy__ = '' self.__proxy_username__ = '' self.__proxy_password__ = '' self.__arch__ = os.uname()[4] self.__needed_packages__ = [] self.__ignored_packages__ = {} self.__release__ = None self.__kernel_running__ = None self.__sources__ = None self.__sources_list__ = getSources() self.__rpmdb__ = None self.__last_rpm_change_number__ = 0 self.__last_sources_change_number__ = 0 # public data member self.change_number = 0 if not mode in ('gui', 'tui'): raise _("invalid rhnAppletModel mode %s") % mode self.__mode__ = mode self.__cert__ = cert def __build_sources(self): if self.__sources_list__ == None: self.__sources_list__ = getSources() if self.__sources_list__ == None: self.__sources__ = None return self.__sources__ = [] has_up2date = 0 for source in self.__sources_list__: try: if source['type'] == 'up2date': if has_up2date == 0: rpc_server = rhnAppletRPC(self.__url__, self.__refresh_callback__, self.__uuid__, self.__release__, self.__arch__, self.__cert__, self.__proxy__, self.__proxy_username__, self.__proxy_password__) if rpc_server != None: self.__sources__.append(rpc_server) has_up2date = 1 elif source['type'] == 'yum': src = rhnAppletYum(source['url'], self.__refresh_callback__, source['label'], self.__release__, self.__arch__, self.__proxy__, self.__proxy_username__, self.__proxy_password__) if src != None: rhn_utils.log_debug("adding yum source %s" % (source['label'])) self.__sources__.append(src) elif source['type'] == 'apt': src = rhnAppletApt(source['url'], self.__refresh_callback__, source['label'], source['dist'], self.__release__, self.__arch__, self.__proxy__, self.__proxy_username__, self.__proxy_password__) if src != None: rhn_utils.log_debug("adding apt source %s" % (source['label'])) self.__sources__.append(src) else: rhn_utils.log_debug("Unsupported source type %s for %s" % (source['type'], source['url'])) except: import sys rhn_utils.log_debug("Failed to initialize source %s for %s : %s %s" % ( source['type'], source['url'], sys.exc_type, sys.exc_value)) # # Depending on the set of source used the usage of the applet may or # may not require the user to consent the RHN licence usage. This # routine walks the list of sources to check if any such consent is # actually needed. # def need_consent(self): try: if self.__sources_list__ == None: self.__sources_list__ = getSources() if self.__sources_list__ == None: return 0 for source in self.__sources_list__: if source['type'] == 'up2date': rhn_utils.log_debug("Applet uses RHN as source") return 1 except: rhn_utils.log_debug("Exception checking if RHN is in sources") return 0 rhn_utils.log_debug("Applet does not uses RHN as source") return 0 def set_proxy(self, proxy, u, p): self.__proxy__ = proxy self.__proxy_username__ = u self.__proxy_password__ = p if self.__sources__ != None: for source in self.__sources__: source.set_proxy(proxy, u, p) def is_package_ignored(self, name): if self.__ignored_packages__.has_key(name): return 1 else: return 0 def add_ignored_package(self, name): self.__ignored_packages__[name] = 1 def remove_ignored_package(self, name): if self.__ignored_packages__.has_key(name): del self.__ignored_packages__[name] def ignored_package_list(self): return self.__ignored_packages__.keys() def installed_package(self, name): rpm = self.__rpmdb__ return rpm.find_latest_by_name(name) def needed_packages(self): return self.__needed_packages__, self.__ignored_packages__.keys() def alerts(self): if not self.fully_initialized(): return [] ret = [] # find the latest version of the package that has the same # name as the package that the currently running kernel # belongs to if len(self.__kernel_running__): latest_kernel = self.__rpmdb__.find_latest_by_name(self.__kernel_running__["name"]) if not latest_kernel["epoch"]: latest_kernel["epoch"] = None if latest_kernel != self.__kernel_running__: a = rhnAppletAlert('kernel', self.__kernel_running__, latest_kernel) ret.append(a) else: # not running a kernel in RPM format? not much we can do a = rhnAppletAlert('unsupported-kernel', os.uname()[2]) ret.append(a) pass needed_up2date = None for pkg in self.__needed_packages__: if pkg["name"] != "up2date": continue needed_up2date = pkg break if needed_up2date: a = rhnAppletAlert('up2date', self.installed_package("up2date"), needed_up2date) ret.append(a) sources_errors = "" for source in self.__sources__: if source.in_error(): if sources_errors == "": sources_errors = source.name() else: sources_errors = sources_errors + ", %s" %(source.name()) if sources_errors != "": a = rhnAppletAlert('source-error', sources_errors) ret.append(a) return ret # # Check for upgrade packages from a source # def __check_source_refresh(self, install, avail): rpm = self.__rpmdb__ A = avail # format of elements of A is dict with keys: # nvre, name, version, release, epoch, errata_advisory, errata_id I = install # I is a hash of installed packages. keys are package # names, values is LATEST version of package as a dict: # nvre, name, version, version, epoch for latest in A: if not I.has_key(latest['name']): # package not on this system? next! continue installed = I[latest["name"]] cmp = rpm.compare_vre(latest["version"], latest["release"], latest["epoch"], installed["version"], installed["release"], installed["epoch"]) if cmp <= 0: # we have the same or a more recent continue self.__needed_packages__.append(latest) # if we're not initialized, checkin simply returns, letting # defaults control behavior def refresh(self, force=0): if not self.fully_initialized(): return if changedSources(): self.__sources__ = None self.__sources_list__ = None self.__build_sources() rpm = self.__rpmdb__ installed = rpm.latest_installed_packages() self.__needed_packages__ = [] if self.__sources__ != None: for source in self.__sources__: try: avail = source.latest_packages() self.__check_source_refresh(installed, avail) except: import sys rhn_utils.log_debug("failed to check source: %s %s" % (sys.exc_type, sys.exc_value)); self.change_number = self.change_number + 1 def fully_initialized(self): rpm = self.__rpmdb__ if not self.__release__: releases = rpm.find_provides("redhat-release") if releases and len(releases): self.__release__ = releases[0]["version"] if self.__kernel_running__ == None: kernel_image = "/boot/vmlinuz-" + os.uname()[2] kernel_running = rpm.find_provides_by_file(kernel_image) if kernel_running and len(kernel_running): self.__kernel_running__ = {} for i in ("name", "version", "release", "epoch"): self.__kernel_running__[i] = kernel_running[0][i] else: # no idea what kernel you're running rhn_utils.log_debug("could not figure out kernel") self.__kernel_running__ = [ ] if self.__release__ and not self.__sources__: self.__build_sources() if self.__kernel_running__ != None and self.__sources__ and rpm.latest_installed_packages(): return 1 return 0 # change numbers are monotonically increasing values; idea # borrowed from the database world. if something changed, we mark # it. higher level components can watch for changes and act # accordingly. fits in well with GUIs so we don't redraw too # often. def poll_change_numbers(self): changed = 0 if self.__last_rpm_change_number__ != self.__rpmdb__.change_number: self.__last_rpm_change_number__ = self.__rpmdb__.change_number self.change_number = self.change_number + 1 changed = 1 if changedSources(): changed = 1 if self.__sources__ != None: val = 0 for source in self.__sources__: val = val + source.change_number() if val != self.__last_sources_change_number__: self.__last_sources_change_number__ = val; return changed def pkg_as_nvr(self, pkg): return string.join((pkg["name"], pkg["version"], pkg["release"]), "-") class rhnAppletModelTUI(rhnAppletModel): def __init__(self, refresh_callback, url, uuid, cert = None): rhnAppletModel.__init__(self, refresh_callback, url, uuid, 'tui', cert) self.__rpmdb__ = rhnAppletRPM(fatal = 1) self.__rpmdb__.refresh() while not self.fully_initialized(): pass if self.__sources__ != None: for source in self.__sources__: if source != None: source.refresh(1) self.change_number = 1 self.poll_change_numbers() rhnAppletModel.refresh(self) class rhnAppletModelGUI(rhnAppletModel): def __init__(self, refresh_callback, url, uuid, cert = None): rhnAppletModel.__init__(self, refresh_callback, url, uuid, 'gui', cert) self.__rpmdb__ = rhnAppletRPM(fatal = 0) def refresh(self, force=0): newconf = 0 self.__rpmdb__.refresh(force) if not self.fully_initialized(): print "ain't there yet" return if changedSources(): # # the sources configuration changed, update the server list # newconf = 1 if self.__sources__ != None: for source in self.__sources__: if source != None: source.refresh(force) if self.poll_change_numbers() or newconf == 1: rhnAppletModel.refresh(self) class rhnAppletAlert: def __init__(self, type, *data): self.__type__ = type self.__data__ = data def is_unsupported_kernel_alert(self): if self.__type__ == 'unsupported-kernel': return 1 return 0 def is_kernel_alert(self): if self.__type__ == 'kernel': return 1 return 0 def is_up2date_alert(self): if self.__type__ == 'up2date': return 1 return 0 def is_source_alert(self): if self.__type__ == 'source-error': return 1 return 0 def alert_data(self): return self.__data__