libbe/util/subproc.py

   1 # Copyright (C) 2009-2012 Chris Ball <cjb@laptop.org>
   2 #                         W. Trevor King <wking@tremily.us>
   3 #
   4 # This file is part of Bugs Everywhere.
   5 #
   6 # Bugs Everywhere is free software: you can redistribute it and/or modify it
   7 # under the terms of the GNU General Public License as published by the Free
   8 # Software Foundation, either version 2 of the License, or (at your option) any
   9 # later version.
  10 #
  11 # Bugs Everywhere is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13 # FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  14 # more details.
  15 #
  16 # You should have received a copy of the GNU General Public License along with
  17 # Bugs Everywhere.  If not, see <http://www.gnu.org/licenses/>.
  18
  19 """
  20 Functions for running external commands in subprocesses.
  21 """
  22
  23 from subprocess import Popen, PIPE
  24 import sys
  25 import types
  26
  27 import libbe
  28 from encoding import get_encoding
  29 if libbe.TESTING == True:
  30     import doctest
  31
  32 _MSWINDOWS = sys.platform == 'win32'
  33 _POSIX = not _MSWINDOWS
  34
  35 if _POSIX == True:
  36     import os
  37     import select
  38
  39 class CommandError(Exception):
  40     def __init__(self, command, status, stdout=None, stderr=None):
  41         strerror = ['Command failed (%d):\n  %s\n' % (status, stderr),
  42                     'while executing\n  %s' % str(command)]
  43         Exception.__init__(self, '\n'.join(strerror))
  44         self.command = command
  45         self.status = status
  46         self.stdout = stdout
  47         self.stderr = stderr
  48
  49 def invoke(args, stdin=None, stdout=PIPE, stderr=PIPE, expect=(0,),
  50            cwd=None, shell=None, unicode_output=True, encoding=None, **kwargs):
  51     """
  52     expect should be a tuple of allowed exit codes.  cwd should be
  53     the directory from which the command will be executed.  When
  54     unicode_output == True, convert stdout and stdin strings to
  55     unicode before returing them.
  56     """
  57     if cwd == None:
  58         cwd = '.'
  59     if isinstance(shell, types.StringTypes):
  60         list_args = ' '.split(args)  # sloppy, but just for logging
  61         str_args = args
  62     else:
  63         list_args = args
  64         str_args = ' '.join(args)  # sloppy, but just for logging
  65     libbe.LOG.debug('{0}$ {1}'.format(cwd, str_args))
  66     try :
  67         if _POSIX:
  68             if shell is None:
  69                 shell = False
  70             q = Popen(args, stdin=PIPE, stdout=stdout, stderr=stderr,
  71                       shell=shell, cwd=cwd, **kwargs)
  72         else:
  73             assert _MSWINDOWS==True, 'invalid platform'
  74             if shell is None:
  75                 shell = True
  76             # win32 don't have os.execvp() so have to run command in a shell
  77             q = Popen(args, stdin=PIPE, stdout=stdout, stderr=stderr,
  78                       shell=shell, cwd=cwd, **kwargs)
  79     except OSError, e:
  80         raise CommandError(list_args, status=e.args[0], stderr=e)
  81     stdout,stderr = q.communicate(input=stdin)
  82     status = q.wait()
  83     if unicode_output == True:
  84         if encoding == None:
  85             encoding = get_encoding()
  86         if stdout != None:
  87             stdout = unicode(stdout, encoding)
  88         if stderr != None:
  89             stderr = unicode(stderr, encoding)
  90     libbe.LOG.debug('{0}\n{1}{2}'.format(status, stdout, stderr))
  91     if status not in expect:
  92         raise CommandError(list_args, status, stdout, stderr)
  93     return status, stdout, stderr
  94
  95 if libbe.TESTING == True:
  96     suite = doctest.DocTestSuite()