Author: gnichols
Date: 2009-12-22 17:58:11 +0000 (Tue, 22 Dec 2009)
New Revision: 271
Added:
trunk/v7/command-popen2.py
Log:
484657 - hts uses code deprecated in python 2.6
Added: trunk/v7/command-popen2.py
===================================================================
--- trunk/v7/command-popen2.py (rev 0)
+++ trunk/v7/command-popen2.py 2009-12-22 17:58:11 UTC (rev 271)
@@ -0,0 +1,275 @@
+# Copyright (c) 2006 Red Hat, Inc. All rights reserved. This copyrighted material
+# is made available to anyone wishing to use, modify, copy, or
+# redistribute it subject to the terms and conditions of the GNU General
+# Public License v.2.
+#
+# 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+# Author: Greg Nichols
+#
+# Overview:
+# The Command class is a wrapper for shell commands that performs
+# validation and error checking. Example usage can be seen in the
+# __main__ self test function at the end of this file.
+
+
+import os,re, commands, exceptions, popen2, string, sys
+
+
+class Command:
+
+ def __init__(self, command):
+ """ Creates a Command object that wraps the shell command
+ via the supplied string, similar to os.system. The
+ constuctor does not actually execute the command.
+ """
+ self.command = command
+ self.regex = None
+ self.singleLine = True
+ self.regexGroup = None
+ self.output = None
+ self.errors = None
+ self.returnValue = 0
+ self.signal = 0
+
+ def _run(self):
+ commandPipe = popen2.Popen3(self.command, capturestderr=True)
+ self.output = commandPipe.fromchild.readlines()
+ result = commandPipe.wait()
+ self.returnValue = (result >> 8) & 0xFF
+ self.signal = result & 0xFF
+ self.errors = None
+
+ if commandPipe.childerr:
+ self.errors = commandPipe.childerr.readlines()
+ # always print error messages
+ for line in self.errors:
+ sys.stderr.write( line )
+ sys.stderr.flush()
+
+ def _checkErrors(self):
+
+ if self.errors and len(self.errors) > 0:
+ raise V7CommandException(self, "has output on stderr")
+ if self.returnValue != 0:
+ # if error returned, show stdout whether echo or run was called
+ for line in self.output:
+ sys.stdout.write( line )
+ sys.stdout.flush()
+ raise V7CommandException(self, "returned %d" % self.returnValue)
+
+ def run(self):
+ """ This method runs the command to produce an action. Any output
+ to stdout is not echoed to the caller. """
+ self._run()
+ self._checkErrors()
+ return 0
+
+
+ def echo(self):
+ """ output is equivalent to run, except that the commands' output
+ is echo'd on stdout. """
+ self._run()
+ self._checkErrors()
+ for line in self.output:
+ sys.stdout.write( line )
+ return 0
+
+ def echoIgnoreErrors(self):
+ """ like echo, except don't raise exception on errors """
+ self._run()
+ for line in self.output:
+ sys.stdout.write( line )
+ return 0
+
+ def printOutput(self):
+ for line in self.output:
+ sys.stdout.write( line )
+ sys.stdout.flush()
+
+ def printErrors(self):
+ for line in self.errors:
+ sys.stderr.write( line )
+ sys.stderr.flush()
+
+
+ def _getString(self, regex=None, regexGroup=None, singleLine=True, returnList=False, ignoreErrors=False):
+ """
+ Get the string from the command's output. With default parameters
+ it returns the command's single line of output as a string.
+
+ If singleLine is True, and multiple lines are found in the output,
+ V7CommandException is raised.
+
+ The regex parameter allows the output to be searched for a regular
+ expression match. If no regexGroup is supplied, the entire pattern
+ match is returned. The regexGroup parameter allows named
+ substrings of the match to be returned if regex has named groups
+ via the "(?P<name>)" syntax. If no match is found,
+ V7CommandException is raised.
+ """
+ self.regex = regex
+ self.singleLine = singleLine
+ self.regexGroup = regexGroup
+
+ self._run()
+
+ if self.singleLine:
+ if len(self.output) > 1:
+ raise V7CommandException(self, "Found %u lines of output, expected 1" % len(self.output))
+
+ line = self.output[0].strip()
+ if not self.regex:
+ return line
+ # otherwise, try the regex
+ pattern = re.compile(self.regex)
+ match = pattern.match(line)
+ if match:
+ if self.regexGroup:
+ return match.group(self.regexGroup)
+ # otherwise, no group, return the whole line
+ return line
+
+ # no regex match try a grep-style match
+ if not self.regexGroup:
+ match = pattern.search(line)
+ if match:
+ return match.group()
+
+ # otherwise
+ raise V7CommandException(self, "no match for regular expression %s" % self.regex)
+
+
+ #otherwise, multi-line or single-line regex
+ if not self.regex:
+ raise V7CommandError(self, "no regular expression set for multi-line command")
+ pattern = re.compile(self.regex)
+ result = None
+ if returnList:
+ result = list()
+ for line in self.output:
+ if self.regexGroup:
+ match = pattern.match(line)
+ if match:
+ if self.regexGroup:
+ if returnList:
+ result.append(match.group(self.regexGroup))
+ else:
+ return match.group(self.regexGroup)
+ else:
+ # otherwise, return the matching line
+ match = pattern.search(line)
+ if match:
+ if returnList:
+ result.append(match.group())
+ else:
+ return match.group()
+ if result:
+ return result
+
+ # otherwise
+ raise V7CommandException(self, "no match for regular expression %s" % self.regex)
+
+ def getStringList(self, regex=None, regexGroup=None, ignoreErrors=False):
+ """ like getString, except return a complete list of matches on multiple lines."""
+ result = self._getString(regex, regexGroup, singleLine=False, returnList=True)
+ if not ignoreErrors:
+ self._checkErrors()
+ return result
+
+ def getString(self, regex=None, regexGroup=None, singleLine=True, ignoreErrors=False):
+ """ like getString, except return a complete list of matches on multiple lines."""
+ result = self._getString(regex, regexGroup, singleLine, returnList=False)
+ if not ignoreErrors:
+ self._checkErrors()
+ return result
+
+
+ def getInteger(self, regex=None, regexGroup=None, singleLine=True):
+ """
+ getInteger is the same as getString, except the output is required to
+ be an Integer.
+ """
+ value = self.getString(regex, regexGroup, singleLine)
+ return string.atoi(value)
+
+class V7CommandException(exceptions.Exception):
+ def __init__(self, command, message):
+ self.message = message
+ self.command = command
+
+ def __str__(self):
+ return "\"%s\" %s" % (self.command.command, self.message)
+
+
+if __name__ == "__main__":
+
+ try:
+ # positive test: run
+ command = Command("ls -a")
+ print "ls -a:"
+ command.run()
+
+ # positive test: simple match
+ command = Command("date")
+ print "is it 2008?"
+ print command.getString("2008")
+
+ # positive test: regex on single line
+ command = Command("date")
+ print "day of the week via date: %s" % command.getString(regex="^(?P<day>[MTWF][a-z]).*$", regexGroup="day")
+
+ # positive test: regex on multiline
+ command = Command("who")
+ print "a device via who: %s" % command.getString(regex="^(?P<user>[a-z]+) (?P<device>[a-z/]+[0-9]*)[ \t]*(?P<date>2008-\d+-\d+).*$", regexGroup="device", singleLine=False)
+
+ #positive test: integer - simple match
+ command = Command("du .")
+ print "simple du: %u" % command.getInteger(regex="\d+", singleLine=False)
+
+ #positive test: integer
+ command = Command("df .")
+ print "blocks via df: %u" % command.getInteger(regex="^[ \t]+(?P<blocks>[0-9]+[ \t]+).*$", regexGroup="blocks", singleLine=False)
+
+ except V7CommandException, e:
+ print e
+
+ # negative test: fail simple match
+ try:
+ print "is it 1999?:"
+ command = Command("date")
+ command.getString(regex="1999")
+ except V7CommandException, e:
+ print e
+
+ # negative test: return value
+ try:
+ print "negative test: return value:"
+ command = Command("exit 1")
+ print "call exit(1)" % command.getString()
+ except V7CommandException, e:
+ print e
+
+ # negative test: expect single line, get multiple
+ try:
+ print "negative test: more than one line of output"
+ command = Command("who")
+ print "who: %s" % command.getString()
+ except V7CommandException, e:
+ print e
+
+ # negative test: expect single line, get multiple
+ try:
+ print "negative test: output on stderr"
+ command = Command("echo \"boguserror\" >&2; echo \"boguserror\"")
+ print "error echo: %s" % command.getString(regex="boguserror", singleLine=False)
+ except V7CommandException, e:
+ print e
+
+
