python3-distupgrade 1:0.220.10 / usr / lib / python3 / dist-packages / DistUpgrade / DistUpgradeViewNonInteractive.py

# DistUpgradeView.py 
#  
#  Copyright (c) 2004,2005 Canonical
#  
#  Author: Michael Vogt <michael.vogt@ubuntu.com>
# 
#  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, write to the Free Software
#  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
#  USA

from __future__ import absolute_import, print_function

import apt
import apt_pkg
import logging
import time
import sys
import os
import pty
import select
import subprocess
import copy
import apt.progress

try:
    from configparser import NoSectionError, NoOptionError
except ImportError:
    from ConfigParser import NoSectionError, NoOptionError
from subprocess import PIPE, Popen

from .DistUpgradeView import DistUpgradeView, InstallProgress, AcquireProgress
from .DistUpgradeConfigParser import DistUpgradeConfig

class NonInteractiveAcquireProgress(AcquireProgress):
    def update_status(self, uri, descr, shortDescr, status):
        AcquireProgress.update_status(self, uri, descr, shortDescr, status)
        #logging.debug("Fetch: updateStatus %s %s" % (uri, status))
        if status == apt_pkg.STAT_DONE:
            print("fetched %s (%.2f/100) at %sb/s" % (
                uri, self.percent, apt_pkg.size_to_str(int(self.current_cps))))
            if sys.stdout.isatty():
                sys.stdout.flush()
        

class NonInteractiveInstallProgress(InstallProgress):
    """ 
    Non-interactive version of the install progress class
    
    This ensures that conffile prompts are handled and that
    hanging scripts are killed after a (long) timeout via ctrl-c
    """

    def __init__(self, logdir):
        InstallProgress.__init__(self)
        logging.debug("setting up environ for non-interactive use")
        if "DEBIAN_FRONTEND" not in os.environ:
            os.environ["DEBIAN_FRONTEND"] = "noninteractive"
        os.environ["APT_LISTCHANGES_FRONTEND"] = "none"
        os.environ["RELEASE_UPRADER_NO_APPORT"] = "1"
        self.config = DistUpgradeConfig(".")
        self.logdir = logdir
        self.install_run_number = 0
        try:
            if self.config.getWithDefault("NonInteractive","ForceOverwrite", False):
                apt_pkg.config.set("DPkg::Options::","--force-overwrite")
        except (NoSectionError, NoOptionError):
            pass
        # more debug
        #apt_pkg.config.set("Debug::pkgOrderList","true")
        #apt_pkg.config.set("Debug::pkgDPkgPM","true")
        # default to 2400 sec timeout
        self.timeout = 2400
        try:
            self.timeout = self.config.getint("NonInteractive","TerminalTimeout")
        except Exception:
            pass

    def error(self, pkg, errormsg):
        logging.error("got a error from dpkg for pkg: '%s': '%s'" % (pkg, errormsg))
        # check if re-run of maintainer script is requested
        if not self.config.getWithDefault(
            "NonInteractive","DebugBrokenScripts", False):
            return
        # re-run maintainer script with sh -x/perl debug to get a better 
        # idea what went wrong
        # 
        # FIXME: this is just a approximation for now, we also need
        #        to pass:
        #        - a version after remove (if upgrade to new version)
        #
        #        not everything is a shell or perl script
        #
        # if the new preinst fails, its not yet in /var/lib/dpkg/info
        # so this is inaccurate as well
        environ = copy.copy(os.environ)
        environ["PYCENTRAL"] = "debug"
        cmd = []

        # find what maintainer script failed
        if "post-installation" in errormsg:
            prefix = "/var/lib/dpkg/info/"
            name = "postinst"
            argument = "configure"
            maintainer_script = "%s/%s.%s" % (prefix, pkg, name)
        elif "pre-installation" in errormsg:
            prefix = "/var/lib/dpkg/tmp.ci/"
            #prefix = "/var/lib/dpkg/info/"
            name = "preinst"
            argument = "install"
            maintainer_script = "%s/%s" % (prefix, name)
        elif "pre-removal" in errormsg:
            prefix = "/var/lib/dpkg/info/"
            name = "prerm"
            argument = "remove"
            maintainer_script = "%s/%s.%s" % (prefix, pkg, name)
        elif "post-removal" in errormsg:
            prefix = "/var/lib/dpkg/info/"
            name = "postrm"
            argument = "remove"
            maintainer_script = "%s/%s.%s" % (prefix, pkg, name)
        else:
            print("UNKNOWN (trigger?) dpkg/script failure for %s (%s) " % (pkg, errormsg))
            return

        # find out about the interpreter
        if not os.path.exists(maintainer_script):
            logging.error("can not find failed maintainer script '%s' " % maintainer_script)
            return
        interp = open(maintainer_script).readline()[2:].strip().split()[0]
        if ("bash" in interp) or ("/bin/sh" in interp):
            debug_opts = ["-ex"]
        elif ("perl" in interp):
            debug_opts = ["-d"]
            environ["PERLDB_OPTS"] = "AutoTrace NonStop"
        else:
            logging.warning("unknown interpreter: '%s'" % interp)

        # check if debconf is used and fiddle a bit more if it is
        if ". /usr/share/debconf/confmodule" in open(maintainer_script).read():
            environ["DEBCONF_DEBUG"] = "developer"
            environ["DEBIAN_HAS_FRONTEND"] = "1"
            interp = "/usr/share/debconf/frontend"
            debug_opts = ["sh","-ex"]

        # build command
        cmd.append(interp)
        cmd.extend(debug_opts)
        cmd.append(maintainer_script)
        cmd.append(argument)

        # check if we need to pass a version
        if name == "postinst":
            version = Popen("dpkg-query -s %s|grep ^Config-Version" % pkg,
                            shell=True, stdout=PIPE,
                            universal_newlines=True).communicate()[0]
            if version:
                cmd.append(version.split(":",1)[1].strip())
        elif name == "preinst":
            pkg = os.path.basename(pkg)
            pkg = pkg.split("_")[0]
            version = Popen("dpkg-query -s %s|grep ^Version" % pkg,
                            shell=True, stdout=PIPE,
                            universal_newlines=True).communicate()[0]
            if version:
                cmd.append(version.split(":",1)[1].strip())

        logging.debug("re-running '%s' (%s)" % (cmd, environ))
        ret = subprocess.call(cmd, env=environ)
        logging.debug("%s script returned: %s" % (name,ret))
        
    def conffile(self, current, new):
        logging.warning("got a conffile-prompt from dpkg for file: '%s'" % current)
        # looks like we have a race here *sometimes*
        time.sleep(5)
        try:
          # don't overwrite
          os.write(self.master_fd,"n\n")
        except Exception as e:
          logging.error("error '%s' when trying to write to the conffile"%e)

    def start_update(self):
        InstallProgress.start_update(self)
        self.last_activity = time.time()
        progress_log = self.config.getWithDefault("NonInteractive","DpkgProgressLog", False)
        if progress_log:
            fullpath = os.path.join(self.logdir, "dpkg-progress.%s.log" % self.install_run_number)
            logging.debug("writing dpkg progress log to '%s'" % fullpath)
            self.dpkg_progress_log = open(fullpath, "w")
        else:
            self.dpkg_progress_log = open(os.devnull, "w")
        self.dpkg_progress_log.write("%s: Start\n" % time.time())
    def finish_update(self):
        InstallProgress.finish_update(self)
        self.dpkg_progress_log.write("%s: Finished\n" % time.time())
        self.dpkg_progress_log.close()
        self.install_run_number += 1
    def status_change(self, pkg, percent, status_str):
        self.dpkg_progress_log.write("%s:%s:%s:%s\n" % (time.time(),
                                                        percent,
                                                        pkg,
                                                        status_str))
    def update_interface(self):
        InstallProgress.update_interface(self)
        if self.statusfd == None:
            return
        if (self.last_activity + self.timeout) < time.time():
            logging.warning("no activity %s seconds (%s) - sending ctrl-c" % (
                    self.timeout, self.status))
            # ctrl-c
            os.write(self.master_fd,chr(3))
        # read master fd and write to stdout so that terminal output
        # actualy works
        res = select.select([self.master_fd],[],[],0.1)
        while len(res[0]) > 0:
           self.last_activity = time.time()
           try:
               s = os.read(self.master_fd, 1)
               sys.stdout.write("%s" % s)
           except OSError:
               # happens after we are finished because the fd is closed
               return
           res = select.select([self.master_fd],[],[],0.1)
        sys.stdout.flush()
    

    def fork(self):
        logging.debug("doing a pty.fork()")
        # some maintainer scripts fail without
        os.environ["TERM"] = "dumb"
        # unset PAGER so that we can do "diff" in the dpkg prompt
        os.environ["PAGER"] = "true"
        (self.pid, self.master_fd) = pty.fork()
        if self.pid != 0:
            logging.debug("pid is: %s" % self.pid)
        return self.pid

class DistUpgradeViewNonInteractive(DistUpgradeView):
    " non-interactive version of the upgrade view "
    def __init__(self, datadir=None, logdir=None):
        DistUpgradeView.__init__(self)
        self.config = DistUpgradeConfig(".")
        self._acquireProgress = NonInteractiveAcquireProgress()
        self._installProgress = NonInteractiveInstallProgress(logdir)
        self._opProgress = apt.progress.base.OpProgress()
        sys.__excepthook__ = self.excepthook
    def excepthook(self, type, value, traceback):
        " on uncaught exceptions -> print error and reboot "
        logging.exception("got exception '%s': %s " % (type, value))
        #sys.excepthook(type, value, traceback)
        self.confirmRestart()
    def getOpCacheProgress(self):
        " return a OpProgress() subclass for the given graphic"
        return self._opProgress
    def getAcquireProgress(self):
        " return an acquire progress object "
        return self._acquireProgress
    def getInstallProgress(self, cache=None):
        " return a install progress object "
        return self._installProgress
    def updateStatus(self, msg):
        """ update the current status of the distUpgrade based
            on the current view
        """
        pass
    def setStep(self, step):
        """ we have 5 steps current for a upgrade:
        1. Analyzing the system
        2. Updating repository information
        3. Performing the upgrade
        4. Post upgrade stuff
        5. Complete
        """
        pass
    def confirmChanges(self, summary, changes, demotions, downloadSize,
                       actions=None, removal_bold=True):
        DistUpgradeView.confirmChanges(self, summary, changes, demotions, 
                                       downloadSize, actions)
        logging.debug("toinstall: '%s'" % [p.name for p in self.toInstall])
        logging.debug("toupgrade: '%s'" % [p.name for p in self.toUpgrade])
        logging.debug("toremove: '%s'" % [p.name for p in self.toRemove])
        return True
    def askYesNoQuestion(self, summary, msg, default='No'):
        " ask a Yes/No question and return True on 'Yes' "
        # if this gets enabled upgrades over ssh with the non-interactive
        # frontend will no longer work
        #if default.lower() == "no":
        #    return False
        return True
    def confirmRestart(self):
        " generic ask about the restart, can be overridden "
        logging.debug("confirmRestart() called")
        # rebooting here makes sense if we run e.g. in qemu
        return self.config.getWithDefault("NonInteractive","RealReboot", False)
    def error(self, summary, msg, extended_msg=None):
        " display a error "
        logging.error("%s %s (%s)" % (summary, msg, extended_msg))
    def abort(self):
        logging.error("view.abort called")


if __name__ == "__main__":

  view = DistUpgradeViewNonInteractive()
  ap = NonInteractiveAcquireProgress()
  ip = NonInteractiveInstallProgress()

  #ip.error("linux-image-2.6.17-10-generic","post-installation script failed")
  ip.error("xserver-xorg","pre-installation script failed")

  cache = apt.Cache()
  for pkg in sys.argv[1:]:
    #if cache[pkg].is_installed:
    #  cache[pkg].mark_delete()
    #else:
    cache[pkg].mark_install()
  cache.commit(ap, ip)
  time.sleep(2)
  sys.exit(0)