QMTest/TestCmd.py

   1 """
   2 TestCmd.py:  a testing framework for commands and scripts.
   3
   4 The TestCmd module provides a framework for portable automated testing
   5 of executable commands and scripts (in any language, not just Python),
   6 especially commands and scripts that require file system interaction.
   7
   8 In addition to running tests and evaluating conditions, the TestCmd
   9 module manages and cleans up one or more temporary workspace
  10 directories, and provides methods for creating files and directories in
  11 those workspace directories from in-line data, here-documents), allowing
  12 tests to be completely self-contained.
  13
  14 A TestCmd environment object is created via the usual invocation:
  15
  16     import TestCmd
  17     test = TestCmd.TestCmd()
  18
  19 There are a bunch of keyword arguments that you can use at instantiation
  20 time:
  21
  22     test = TestCmd.TestCmd(description = 'string',
  23                            program = 'program_or_script_to_test',
  24                            interpreter = 'script_interpreter',
  25                            workdir = 'prefix',
  26                            subdir = 'subdir',
  27                            verbose = Boolean,
  28                            match = default_match_function,
  29                            combine = Boolean)
  30
  31 There are a bunch of methods that let you do a bunch of different
  32 things.  Here is an overview of them:
  33
  34     test.verbose_set(1)
  35
  36     test.description_set('string')
  37
  38     test.program_set('program_or_script_to_test')
  39
  40     test.interpreter_set('script_interpreter')
  41     test.interpreter_set(['script_interpreter', 'arg'])
  42
  43     test.workdir_set('prefix')
  44     test.workdir_set('')
  45
  46     test.workpath('file')
  47     test.workpath('subdir', 'file')
  48
  49     test.subdir('subdir', ...)
  50
  51     test.rmdir('subdir', ...)
  52
  53     test.write('file', "contents\n")
  54     test.write(['subdir', 'file'], "contents\n")
  55
  56     test.read('file')
  57     test.read(['subdir', 'file'])
  58     test.read('file', mode)
  59     test.read(['subdir', 'file'], mode)
  60
  61     test.writable('dir', 1)
  62     test.writable('dir', None)
  63
  64     test.preserve(condition, ...)
  65
  66     test.cleanup(condition)
  67
  68     test.run(program = 'program_or_script_to_run',
  69              interpreter = 'script_interpreter',
  70              arguments = 'arguments to pass to program',
  71              chdir = 'directory_to_chdir_to',
  72              stdin = 'input to feed to the program\n')
  73
  74     test.pass_test()
  75     test.pass_test(condition)
  76     test.pass_test(condition, function)
  77
  78     test.fail_test()
  79     test.fail_test(condition)
  80     test.fail_test(condition, function)
  81     test.fail_test(condition, function, skip)
  82
  83     test.no_result()
  84     test.no_result(condition)
  85     test.no_result(condition, function)
  86     test.no_result(condition, function, skip)
  87
  88     test.stdout()
  89     test.stdout(run)
  90
  91     test.stderr()
  92     test.stderr(run)
  93
  94     test.symlink(target, link)
  95
  96     test.match(actual, expected)
  97
  98     test.match_exact("actual 1\nactual 2\n", "expected 1\nexpected 2\n")
  99     test.match_exact(["actual 1\n", "actual 2\n"],
 100                      ["expected 1\n", "expected 2\n"])
 101
 102     test.match_re("actual 1\nactual 2\n", regex_string)
 103     test.match_re(["actual 1\n", "actual 2\n"], list_of_regexes)
 104
 105     test.match_re_dotall("actual 1\nactual 2\n", regex_string)
 106     test.match_re_dotall(["actual 1\n", "actual 2\n"], list_of_regexes)
 107
 108     test.tempdir()
 109     test.tempdir('temporary-directory')
 110
 111     test.sleep()
 112     test.sleep(seconds)
 113
 114     test.where_is('foo')
 115     test.where_is('foo', 'PATH1:PATH2')
 116     test.where_is('foo', 'PATH1;PATH2', '.suffix3;.suffix4')
 117
 118     test.unlink('file')
 119     test.unlink('subdir', 'file')
 120
 121 The TestCmd module provides pass_test(), fail_test(), and no_result()
 122 unbound functions that report test results for use with the Aegis change
 123 management system.  These methods terminate the test immediately,
 124 reporting PASSED, FAILED, or NO RESULT respectively, and exiting with
 125 status 0 (success), 1 or 2 respectively.  This allows for a distinction
 126 between an actual failed test and a test that could not be properly
 127 evaluated because of an external condition (such as a full file system
 128 or incorrect permissions).
 129
 130     import TestCmd
 131
 132     TestCmd.pass_test()
 133     TestCmd.pass_test(condition)
 134     TestCmd.pass_test(condition, function)
 135
 136     TestCmd.fail_test()
 137     TestCmd.fail_test(condition)
 138     TestCmd.fail_test(condition, function)
 139     TestCmd.fail_test(condition, function, skip)
 140
 141     TestCmd.no_result()
 142     TestCmd.no_result(condition)
 143     TestCmd.no_result(condition, function)
 144     TestCmd.no_result(condition, function, skip)
 145
 146 The TestCmd module also provides unbound functions that handle matching
 147 in the same way as the match_*() methods described above.
 148
 149     import TestCmd
 150
 151     test = TestCmd.TestCmd(match = TestCmd.match_exact)
 152
 153     test = TestCmd.TestCmd(match = TestCmd.match_re)
 154
 155     test = TestCmd.TestCmd(match = TestCmd.match_re_dotall)
 156
 157 Lastly, the where_is() method also exists in an unbound function
 158 version.
 159
 160     import TestCmd
 161
 162     TestCmd.where_is('foo')
 163     TestCmd.where_is('foo', 'PATH1:PATH2')
 164     TestCmd.where_is('foo', 'PATH1;PATH2', '.suffix3;.suffix4')
 165 """
 166
 167 # Copyright 2000, 2001, 2002, 2003, 2004 Steven Knight
 168 # This module is free software, and you may redistribute it and/or modify
 169 # it under the same terms as Python itself, so long as this copyright message
 170 # and disclaimer are retained in their original form.
 171 #
 172 # IN NO EVENT SHALL THE AUTHOR BE LIABLE TO ANY PARTY FOR DIRECT, INDIRECT,
 173 # SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE USE OF
 174 # THIS CODE, EVEN IF THE AUTHOR HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH
 175 # DAMAGE.
 176 #
 177 # THE AUTHOR SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT
 178 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
 179 # PARTICULAR PURPOSE.  THE CODE PROVIDED HEREUNDER IS ON AN "AS IS" BASIS,
 180 # AND THERE IS NO OBLIGATION WHATSOEVER TO PROVIDE MAINTENANCE,
 181 # SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 182
 183 __author__ = "Steven Knight <knight at baldmt dot com>"
 184 __revision__ = "TestCmd.py 0.30.D001 2007/10/01 16:53:55 knight"
 185 __version__ = "0.30"
 186
 187 import os
 188 import os.path
 189 import popen2
 190 import re
 191 import shutil
 192 import stat
 193 import string
 194 import sys
 195 import tempfile
 196 import time
 197 import traceback
 198 import types
 199 import UserList
 200
 201 __all__ = [
 202     'diff_re',
 203     'fail_test',
 204     'no_result',
 205     'pass_test',
 206     'match_exact',
 207     'match_re',
 208     'match_re_dotall',
 209     'python_executable',
 210     'TestCmd'
 211 ]
 212
 213 def is_List(e):
 214     return type(e) is types.ListType \
 215         or isinstance(e, UserList.UserList)
 216
 217 try:
 218     from UserString import UserString
 219 except ImportError:
 220     class UserString:
 221         pass
 222
 223 if hasattr(types, 'UnicodeType'):
 224     def is_String(e):
 225         return type(e) is types.StringType \
 226             or type(e) is types.UnicodeType \
 227             or isinstance(e, UserString)
 228 else:
 229     def is_String(e):
 230         return type(e) is types.StringType or isinstance(e, UserString)
 231
 232 tempfile.template = 'testcmd.'
 233 if os.name in ('posix', 'nt'):
 234     tempfile.template = 'testcmd.' + str(os.getpid()) + '.'
 235 else:
 236     tempfile.template = 'testcmd.'
 237
 238 re_space = re.compile('\s')
 239
 240 _Cleanup = []
 241
 242 def _clean():
 243     global _Cleanup
 244     cleanlist = filter(None, _Cleanup)
 245     del _Cleanup[:]
 246     cleanlist.reverse()
 247     for test in cleanlist:
 248         test.cleanup()
 249
 250 sys.exitfunc = _clean
 251
 252 class Collector:
 253     def __init__(self, top):
 254         self.entries = [top]
 255     def __call__(self, arg, dirname, names):
 256         pathjoin = lambda n, d=dirname: os.path.join(d, n)
 257         self.entries.extend(map(pathjoin, names))
 258
 259 def _caller(tblist, skip):
 260     string = ""
 261     arr = []
 262     for file, line, name, text in tblist:
 263         if file[-10:] == "TestCmd.py":
 264                 break
 265         arr = [(file, line, name, text)] + arr
 266     atfrom = "at"
 267     for file, line, name, text in arr[skip:]:
 268         if name == "?":
 269             name = ""
 270         else:
 271             name = " (" + name + ")"
 272         string = string + ("%s line %d of %s%s\n" % (atfrom, line, file, name))
 273         atfrom = "\tfrom"
 274     return string
 275
 276 def fail_test(self = None, condition = 1, function = None, skip = 0):
 277     """Cause the test to fail.
 278
 279     By default, the fail_test() method reports that the test FAILED
 280     and exits with a status of 1.  If a condition argument is supplied,
 281     the test fails only if the condition is true.
 282     """
 283     if not condition:
 284         return
 285     if not function is None:
 286         function()
 287     of = ""
 288     desc = ""
 289     sep = " "
 290     if not self is None:
 291         if self.program:
 292             of = " of " + self.program
 293             sep = "\n\t"
 294         if self.description:
 295             desc = " [" + self.description + "]"
 296             sep = "\n\t"
 297
 298     at = _caller(traceback.extract_stack(), skip)
 299     sys.stderr.write("FAILED test" + of + desc + sep + at)
 300
 301     sys.exit(1)
 302
 303 def no_result(self = None, condition = 1, function = None, skip = 0):
 304     """Causes a test to exit with no valid result.
 305
 306     By default, the no_result() method reports NO RESULT for the test
 307     and exits with a status of 2.  If a condition argument is supplied,
 308     the test fails only if the condition is true.
 309     """
 310     if not condition:
 311         return
 312     if not function is None:
 313         function()
 314     of = ""
 315     desc = ""
 316     sep = " "
 317     if not self is None:
 318         if self.program:
 319             of = " of " + self.program
 320             sep = "\n\t"
 321         if self.description:
 322             desc = " [" + self.description + "]"
 323             sep = "\n\t"
 324
 325     at = _caller(traceback.extract_stack(), skip)
 326     sys.stderr.write("NO RESULT for test" + of + desc + sep + at)
 327
 328     sys.exit(2)
 329
 330 def pass_test(self = None, condition = 1, function = None):
 331     """Causes a test to pass.
 332
 333     By default, the pass_test() method reports PASSED for the test
 334     and exits with a status of 0.  If a condition argument is supplied,
 335     the test passes only if the condition is true.
 336     """
 337     if not condition:
 338         return
 339     if not function is None:
 340         function()
 341     sys.stderr.write("PASSED\n")
 342     sys.exit(0)
 343
 344 def match_exact(lines = None, matches = None):
 345     """
 346     """
 347     if not is_List(lines):
 348         lines = string.split(lines, "\n")
 349     if not is_List(matches):
 350         matches = string.split(matches, "\n")
 351     if len(lines) != len(matches):
 352         return
 353     for i in range(len(lines)):
 354         if lines[i] != matches[i]:
 355             return
 356     return 1
 357
 358 def match_re(lines = None, res = None):
 359     """
 360     """
 361     if not is_List(lines):
 362         lines = string.split(lines, "\n")
 363     if not is_List(res):
 364         res = string.split(res, "\n")
 365     if len(lines) != len(res):
 366         return
 367     for i in range(len(lines)):
 368         if not re.compile("^" + res[i] + "$").search(lines[i]):
 369             return
 370     return 1
 371
 372 def match_re_dotall(lines = None, res = None):
 373     """
 374     """
 375     if not type(lines) is type(""):
 376         lines = string.join(lines, "\n")
 377     if not type(res) is type(""):
 378         res = string.join(res, "\n")
 379     if re.compile("^" + res + "$", re.DOTALL).match(lines):
 380         return 1
 381
 382 def diff_re(a, b, fromfile='', tofile='',
 383                 fromfiledate='', tofiledate='', n=3, lineterm='\n'):
 384     """
 385     A simple "diff" of two sets of lines when the expected lines
 386     are regular expressions.  This is a really dumb thing that
 387     just compares each line in turn, so it doesn't look for
 388     chunks of matching lines and the like--but at least it lets
 389     you know exactly which line first didn't compare correctl...
 390     """
 391     result = []
 392     diff = len(a) - len(b)
 393     if diff < 0:
 394         a = a + ['']*(-diff)
 395     elif diff > 0:
 396         b = b + ['']*diff
 397     i = 0
 398     for aline, bline in zip(a, b):
 399         if not re.compile("^" + aline + "$").search(bline):
 400             result.append("%sc%s" % (i+1, i+1))
 401             result.append('< ' + repr(a[i]))
 402             result.append('---')
 403             result.append('> ' + repr(b[i]))
 404         i = i+1
 405     return result
 406
 407 if os.name == 'java':
 408
 409     python_executable = os.path.join(sys.prefix, 'jython')
 410
 411 else:
 412
 413     python_executable = sys.executable
 414
 415 if sys.platform == 'win32':
 416
 417     default_sleep_seconds = 2
 418
 419     def where_is(file, path=None, pathext=None):
 420         if path is None:
 421             path = os.environ['PATH']
 422         if is_String(path):
 423             path = string.split(path, os.pathsep)
 424         if pathext is None:
 425             pathext = os.environ['PATHEXT']
 426         if is_String(pathext):
 427             pathext = string.split(pathext, os.pathsep)
 428         for ext in pathext:
 429             if string.lower(ext) == string.lower(file[-len(ext):]):
 430                 pathext = ['']
 431                 break
 432         for dir in path:
 433             f = os.path.join(dir, file)
 434             for ext in pathext:
 435                 fext = f + ext
 436                 if os.path.isfile(fext):
 437                     return fext
 438         return None
 439
 440 else:
 441
 442     def where_is(file, path=None, pathext=None):
 443         if path is None:
 444             path = os.environ['PATH']
 445         if is_String(path):
 446             path = string.split(path, os.pathsep)
 447         for dir in path:
 448             f = os.path.join(dir, file)
 449             if os.path.isfile(f):
 450                 try:
 451                     st = os.stat(f)
 452                 except OSError:
 453                     continue
 454                 if stat.S_IMODE(st[stat.ST_MODE]) & 0111:
 455                     return f
 456         return None
 457
 458     default_sleep_seconds = 1
 459
 460 class TestCmd:
 461     """Class TestCmd
 462     """
 463
 464     def __init__(self, description = None,
 465                        program = None,
 466                        interpreter = None,
 467                        workdir = None,
 468                        subdir = None,
 469                        verbose = None,
 470                        match = None,
 471                        combine = 0,
 472                        universal_newlines = 1):
 473         self._cwd = os.getcwd()
 474         self.description_set(description)
 475         self.program_set(program)
 476         self.interpreter_set(interpreter)
 477         if verbose is None:
 478             try:
 479                 verbose = max( 0, int(os.environ.get('TESTCMD_VERBOSE', 0)) )
 480             except ValueError:
 481                 verbose = 0
 482         self.verbose_set(verbose)
 483         self.combine = combine
 484         self.universal_newlines = universal_newlines
 485         if not match is None:
 486             self.match_func = match
 487         else:
 488             self.match_func = match_re
 489         self._dirlist = []
 490         self._preserve = {'pass_test': 0, 'fail_test': 0, 'no_result': 0}
 491         if os.environ.has_key('PRESERVE') and not os.environ['PRESERVE'] is '':
 492             self._preserve['pass_test'] = os.environ['PRESERVE']
 493             self._preserve['fail_test'] = os.environ['PRESERVE']
 494             self._preserve['no_result'] = os.environ['PRESERVE']
 495         else:
 496             try:
 497                 self._preserve['pass_test'] = os.environ['PRESERVE_PASS']
 498             except KeyError:
 499                 pass
 500             try:
 501                 self._preserve['fail_test'] = os.environ['PRESERVE_FAIL']
 502             except KeyError:
 503                 pass
 504             try:
 505                 self._preserve['no_result'] = os.environ['PRESERVE_NO_RESULT']
 506             except KeyError:
 507                 pass
 508         self._stdout = []
 509         self._stderr = []
 510         self.status = None
 511         self.condition = 'no_result'
 512         self.workdir_set(workdir)
 513         self.subdir(subdir)
 514
 515     def __del__(self):
 516         self.cleanup()
 517
 518     def __repr__(self):
 519         return "%x" % id(self)
 520
 521     if os.name == 'posix':
 522
 523         def escape(self, arg):
 524             "escape shell special characters"
 525             slash = '\\'
 526             special = '"$'
 527
 528             arg = string.replace(arg, slash, slash+slash)
 529             for c in special:
 530                 arg = string.replace(arg, c, slash+c)
 531
 532             if re_space.search(arg):
 533                 arg = '"' + arg + '"'
 534             return arg
 535
 536     else:
 537
 538         # Windows does not allow special characters in file names
 539         # anyway, so no need for an escape function, we will just quote
 540         # the arg.
 541         def escape(self, arg):
 542             if re_space.search(arg):
 543                 arg = '"' + arg + '"'
 544             return arg
 545
 546     def canonicalize(self, path):
 547         if is_List(path):
 548             path = apply(os.path.join, tuple(path))
 549         if not os.path.isabs(path):
 550             path = os.path.join(self.workdir, path)
 551         return path
 552
 553     def cleanup(self, condition = None):
 554         """Removes any temporary working directories for the specified
 555         TestCmd environment.  If the environment variable PRESERVE was
 556         set when the TestCmd environment was created, temporary working
 557         directories are not removed.  If any of the environment variables
 558         PRESERVE_PASS, PRESERVE_FAIL, or PRESERVE_NO_RESULT were set
 559         when the TestCmd environment was created, then temporary working
 560         directories are not removed if the test passed, failed, or had
 561         no result, respectively.  Temporary working directories are also
 562         preserved for conditions specified via the preserve method.
 563
 564         Typically, this method is not called directly, but is used when
 565         the script exits to clean up temporary working directories as
 566         appropriate for the exit status.
 567         """
 568         if not self._dirlist:
 569             return
 570         os.chdir(self._cwd)
 571         self.workdir = None
 572         if condition is None:
 573             condition = self.condition
 574         if self._preserve[condition]:
 575             for dir in self._dirlist:
 576                 print "Preserved directory", dir
 577         else:
 578             list = self._dirlist[:]
 579             list.reverse()
 580             for dir in list:
 581                 self.writable(dir, 1)
 582                 shutil.rmtree(dir, ignore_errors = 1)
 583             self._dirlist = []
 584
 585         try:
 586             global _Cleanup
 587             _Cleanup.remove(self)
 588         except (AttributeError, ValueError):
 589             pass
 590
 591     def chmod(self, path, mode):
 592         """Changes permissions on the specified file or directory
 593         path name."""
 594         path = self.canonicalize(path)
 595         os.chmod(path, mode)
 596
 597     def description_set(self, description):
 598         """Set the description of the functionality being tested.
 599         """
 600         self.description = description
 601
 602 #    def diff(self):
 603 #        """Diff two arrays.
 604 #        """
 605
 606     def fail_test(self, condition = 1, function = None, skip = 0):
 607         """Cause the test to fail.
 608         """
 609         if not condition:
 610             return
 611         self.condition = 'fail_test'
 612         fail_test(self = self,
 613                   condition = condition,
 614                   function = function,
 615                   skip = skip)
 616
 617     def interpreter_set(self, interpreter):
 618         """Set the program to be used to interpret the program
 619         under test as a script.
 620         """
 621         self.interpreter = interpreter
 622
 623     def match(self, lines, matches):
 624         """Compare actual and expected file contents.
 625         """
 626         return self.match_func(lines, matches)
 627
 628     def match_exact(self, lines, matches):
 629         """Compare actual and expected file contents.
 630         """
 631         return match_exact(lines, matches)
 632
 633     def match_re(self, lines, res):
 634         """Compare actual and expected file contents.
 635         """
 636         return match_re(lines, res)
 637
 638     def match_re_dotall(self, lines, res):
 639         """Compare actual and expected file contents.
 640         """
 641         return match_re_dotall(lines, res)
 642
 643     def no_result(self, condition = 1, function = None, skip = 0):
 644         """Report that the test could not be run.
 645         """
 646         if not condition:
 647             return
 648         self.condition = 'no_result'
 649         no_result(self = self,
 650                   condition = condition,
 651                   function = function,
 652                   skip = skip)
 653
 654     def pass_test(self, condition = 1, function = None):
 655         """Cause the test to pass.
 656         """
 657         if not condition:
 658             return
 659         self.condition = 'pass_test'
 660         pass_test(self = self, condition = condition, function = function)
 661
 662     def preserve(self, *conditions):
 663         """Arrange for the temporary working directories for the
 664         specified TestCmd environment to be preserved for one or more
 665         conditions.  If no conditions are specified, arranges for
 666         the temporary working directories to be preserved for all
 667         conditions.
 668         """
 669         if conditions is ():
 670             conditions = ('pass_test', 'fail_test', 'no_result')
 671         for cond in conditions:
 672             self._preserve[cond] = 1
 673
 674     def program_set(self, program):
 675         """Set the executable program or script to be tested.
 676         """
 677         if program and not os.path.isabs(program):
 678             program = os.path.join(self._cwd, program)
 679         self.program = program
 680
 681     def read(self, file, mode = 'rb'):
 682         """Reads and returns the contents of the specified file name.
 683         The file name may be a list, in which case the elements are
 684         concatenated with the os.path.join() method.  The file is
 685         assumed to be under the temporary working directory unless it
 686         is an absolute path name.  The I/O mode for the file may
 687         be specified; it must begin with an 'r'.  The default is
 688         'rb' (binary read).
 689         """
 690         file = self.canonicalize(file)
 691         if mode[0] != 'r':
 692             raise ValueError, "mode must begin with 'r'"
 693         return open(file, mode).read()
 694
 695     def rmdir(self, dir):
 696         """Removes the specified dir name.
 697         The dir name may be a list, in which case the elements are
 698         concatenated with the os.path.join() method.  The dir is
 699         assumed to be under the temporary working directory unless it
 700         is an absolute path name.
 701         The dir must be empty.
 702         """
 703         dir = self.canonicalize(dir)
 704         os.rmdir(dir)
 705
 706     def run(self, program = None,
 707                   interpreter = None,
 708                   arguments = None,
 709                   chdir = None,
 710                   stdin = None,
 711                   universal_newlines = None):
 712         """Runs a test of the program or script for the test
 713         environment.  Standard output and error output are saved for
 714         future retrieval via the stdout() and stderr() methods.
 715
 716         The specified program will have the original directory
 717         prepending unless it is enclosed in a [list].
 718         """
 719         if chdir:
 720             oldcwd = os.getcwd()
 721             if not os.path.isabs(chdir):
 722                 chdir = os.path.join(self.workpath(chdir))
 723             if self.verbose:
 724                 sys.stderr.write("chdir(" + chdir + ")\n")
 725             os.chdir(chdir)
 726         if program:
 727             if type(program) == type('') and not os.path.isabs(program):
 728                 program = os.path.join(self._cwd, program)
 729         else:
 730             program = self.program
 731             if not interpreter:
 732                 interpreter = self.interpreter
 733         if not type(program) in [type([]), type(())]:
 734             program = [program]
 735         cmd = list(program)
 736         if interpreter:
 737             if not type(interpreter) in [type([]), type(())]:
 738                 interpreter = [interpreter]
 739             cmd = list(interpreter) + cmd
 740         if arguments:
 741             if type(arguments) == type(''):
 742                 arguments = string.split(arguments)
 743             cmd.extend(arguments)
 744         cmd_string = string.join(map(self.escape, cmd), ' ')
 745         if self.verbose:
 746             sys.stderr.write(cmd_string + "\n")
 747         if universal_newlines is None:
 748             universal_newlines = self.universal_newlines
 749
 750         try:
 751             import subprocess
 752         except ImportError:
 753             try:
 754                 Popen3 = popen2.Popen3
 755             except AttributeError:
 756                 class Popen3:
 757                     def __init__(self, command):
 758                         (stdin, stdout, stderr) = os.popen3(' ' + command)
 759                         self.stdin = stdin
 760                         self.stdout = stdout
 761                         self.stderr = stderr
 762                     def close_output(self):
 763                         self.stdout.close()
 764                         self.resultcode = self.stderr.close()
 765                     def wait(self):
 766                         return self.resultcode
 767                 if sys.platform == 'win32' and cmd_string[0] == '"':
 768                     cmd_string = '"' + cmd_string + '"'
 769                 p = Popen3(cmd_string)
 770             else:
 771                 p = Popen3(cmd, 1)
 772                 p.stdin = p.tochild
 773                 p.stdout = p.fromchild
 774                 p.stderr = p.childerr
 775         else:
 776             p = subprocess.Popen(cmd,
 777                                  stdin=subprocess.PIPE,
 778                                  stdout=subprocess.PIPE,
 779                                  stderr=subprocess.PIPE,
 780                                  universal_newlines=universal_newlines)
 781
 782         if stdin:
 783             if is_List(stdin):
 784                 for line in stdin:
 785                     p.stdin.write(line)
 786             else:
 787                 p.stdin.write(stdin)
 788         p.stdin.close()
 789
 790         out = p.stdout.read()
 791         err = p.stderr.read()
 792         try:
 793             p.close_output()
 794         except AttributeError:
 795             p.stdout.close()
 796             p.stderr.close()
 797
 798         self.status = p.wait()
 799         if not self.status:
 800             self.status = 0
 801
 802         if self.combine:
 803             self._stdout.append(out + err)
 804         else:
 805             self._stdout.append(out)
 806             self._stderr.append(err)
 807
 808         if chdir:
 809             os.chdir(oldcwd)
 810         if self.verbose >= 2:
 811             write = sys.stdout.write
 812             write('============ STATUS: %d\n' % self.status)
 813             out = self.stdout()
 814             if out or self.verbose >= 3:
 815                 write('============ BEGIN STDOUT (len=%d):\n' % len(out))
 816                 write(out)
 817                 write('============ END STDOUT\n')
 818             err = self.stderr()
 819             if err or self.verbose >= 3:
 820                 write('============ BEGIN STDERR (len=%d)\n' % len(err))
 821                 write(err)
 822                 write('============ END STDERR\n')
 823
 824     def sleep(self, seconds = default_sleep_seconds):
 825         """Sleeps at least the specified number of seconds.  If no
 826         number is specified, sleeps at least the minimum number of
 827         seconds necessary to advance file time stamps on the current
 828         system.  Sleeping more seconds is all right.
 829         """
 830         time.sleep(seconds)
 831
 832     def stderr(self, run = None):
 833         """Returns the error output from the specified run number.
 834         If there is no specified run number, then returns the error
 835         output of the last run.  If the run number is less than zero,
 836         then returns the error output from that many runs back from the
 837         current run.
 838         """
 839         if not run:
 840             run = len(self._stderr)
 841         elif run < 0:
 842             run = len(self._stderr) + run
 843         run = run - 1
 844         return self._stderr[run]
 845
 846     def stdout(self, run = None):
 847         """Returns the standard output from the specified run number.
 848         If there is no specified run number, then returns the standard
 849         output of the last run.  If the run number is less than zero,
 850         then returns the standard output from that many runs back from
 851         the current run.
 852         """
 853         if not run:
 854             run = len(self._stdout)
 855         elif run < 0:
 856             run = len(self._stdout) + run
 857         run = run - 1
 858         return self._stdout[run]
 859
 860     def subdir(self, *subdirs):
 861         """Create new subdirectories under the temporary working
 862         directory, one for each argument.  An argument may be a list,
 863         in which case the list elements are concatenated using the
 864         os.path.join() method.  Subdirectories multiple levels deep
 865         must be created using a separate argument for each level:
 866
 867                 test.subdir('sub', ['sub', 'dir'], ['sub', 'dir', 'ectory'])
 868
 869         Returns the number of subdirectories actually created.
 870         """
 871         count = 0
 872         for sub in subdirs:
 873             if sub is None:
 874                 continue
 875             if is_List(sub):
 876                 sub = apply(os.path.join, tuple(sub))
 877             new = os.path.join(self.workdir, sub)
 878             try:
 879                 os.mkdir(new)
 880             except OSError:
 881                 pass
 882             else:
 883                 count = count + 1
 884         return count
 885
 886     def symlink(self, target, link):
 887         """Creates a symlink to the specified target.
 888         The link name may be a list, in which case the elements are
 889         concatenated with the os.path.join() method.  The link is
 890         assumed to be under the temporary working directory unless it
 891         is an absolute path name. The target is *not* assumed to be
 892         under the temporary working directory.
 893         """
 894         link = self.canonicalize(link)
 895         os.symlink(target, link)
 896
 897     def tempdir(self, path=None):
 898         """Creates a temporary directory.
 899         A unique directory name is generated if no path name is specified.
 900         The directory is created, and will be removed when the TestCmd
 901         object is destroyed.
 902         """
 903         if path is None:
 904             try:
 905                 path = tempfile.mktemp(prefix=tempfile.template)
 906             except TypeError:
 907                 path = tempfile.mktemp()
 908         os.mkdir(path)
 909
 910         # Symlinks in the path will report things
 911         # differently from os.getcwd(), so chdir there
 912         # and back to fetch the canonical path.
 913         cwd = os.getcwd()
 914         try:
 915             os.chdir(path)
 916             path = os.getcwd()
 917         finally:
 918             os.chdir(cwd)
 919
 920         # Uppercase the drive letter since the case of drive
 921         # letters is pretty much random on win32:
 922         drive,rest = os.path.splitdrive(path)
 923         if drive:
 924             path = string.upper(drive) + rest
 925
 926         #
 927         self._dirlist.append(path)
 928         global _Cleanup
 929         try:
 930             _Cleanup.index(self)
 931         except ValueError:
 932             _Cleanup.append(self)
 933
 934         return path
 935
 936     def touch(self, path, mtime=None):
 937         """Updates the modification time on the specified file or
 938         directory path name.  The default is to update to the
 939         current time if no explicit modification time is specified.
 940         """
 941         path = self.canonicalize(path)
 942         atime = os.path.getatime(path)
 943         if mtime is None:
 944             mtime = time.time()
 945         os.utime(path, (atime, mtime))
 946
 947     def unlink(self, file):
 948         """Unlinks the specified file name.
 949         The file name may be a list, in which case the elements are
 950         concatenated with the os.path.join() method.  The file is
 951         assumed to be under the temporary working directory unless it
 952         is an absolute path name.
 953         """
 954         file = self.canonicalize(file)
 955         os.unlink(file)
 956
 957     def verbose_set(self, verbose):
 958         """Set the verbose level.
 959         """
 960         self.verbose = verbose
 961
 962     def where_is(self, file, path=None, pathext=None):
 963         """Find an executable file.
 964         """
 965         if is_List(file):
 966             file = apply(os.path.join, tuple(file))
 967         if not os.path.isabs(file):
 968             file = where_is(file, path, pathext)
 969         return file
 970
 971     def workdir_set(self, path):
 972         """Creates a temporary working directory with the specified
 973         path name.  If the path is a null string (''), a unique
 974         directory name is created.
 975         """
 976         if (path != None):
 977             if path == '':
 978                 path = None
 979             path = self.tempdir(path)
 980         self.workdir = path
 981
 982     def workpath(self, *args):
 983         """Returns the absolute path name to a subdirectory or file
 984         within the current temporary working directory.  Concatenates
 985         the temporary working directory name with the specified
 986         arguments using the os.path.join() method.
 987         """
 988         return apply(os.path.join, (self.workdir,) + tuple(args))
 989
 990     def readable(self, top, read=1):
 991         """Make the specified directory tree readable (read == 1)
 992         or not (read == None).
 993         """
 994
 995         if read:
 996             def do_chmod(fname):
 997                 try: st = os.stat(fname)
 998                 except OSError: pass
 999                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]|0400))
1000         else:
1001             def do_chmod(fname):
1002                 try: st = os.stat(fname)
1003                 except OSError: pass
1004                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]&~0400))
1005
1006         if os.path.isfile(top):
1007             # If it's a file, that's easy, just chmod it.
1008             do_chmod(top)
1009         elif read:
1010             # It's a directory and we're trying to turn on read
1011             # permission, so it's also pretty easy, just chmod the
1012             # directory and then chmod every entry on our walk down the
1013             # tree.  Because os.path.walk() is top-down, we'll enable
1014             # read permission on any directories that have it disabled
1015             # before os.path.walk() tries to list their contents.
1016             do_chmod(top)
1017
1018             def chmod_entries(arg, dirname, names, do_chmod=do_chmod):
1019                 pathnames = map(lambda n, d=dirname: os.path.join(d, n),
1020                                 names)
1021                 map(lambda p, do=do_chmod: do(p), pathnames)
1022
1023             os.path.walk(top, chmod_entries, None)
1024         else:
1025             # It's a directory and we're trying to turn off read
1026             # permission, which means we have to chmod the directoreis
1027             # in the tree bottom-up, lest disabling read permission from
1028             # the top down get in the way of being able to get at lower
1029             # parts of the tree.  But os.path.walk() visits things top
1030             # down, so we just use an object to collect a list of all
1031             # of the entries in the tree, reverse the list, and then
1032             # chmod the reversed (bottom-up) list.
1033             col = Collector(top)
1034             os.path.walk(top, col, None)
1035             col.entries.reverse()
1036             map(lambda d, do=do_chmod: do(d), col.entries)
1037
1038     def writable(self, top, write=1):
1039         """Make the specified directory tree writable (write == 1)
1040         or not (write == None).
1041         """
1042
1043         if write:
1044             def do_chmod(fname):
1045                 try: st = os.stat(fname)
1046                 except OSError: pass
1047                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]|0200))
1048         else:
1049             def do_chmod(fname):
1050                 try: st = os.stat(fname)
1051                 except OSError: pass
1052                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]&~0200))
1053
1054         if os.path.isfile(top):
1055             do_chmod(top)
1056         else:
1057             col = Collector(top)
1058             os.path.walk(top, col, None)
1059             map(lambda d, do=do_chmod: do(d), col.entries)
1060
1061     def executable(self, top, execute=1):
1062         """Make the specified directory tree executable (execute == 1)
1063         or not (execute == None).
1064         """
1065
1066         if execute:
1067             def do_chmod(fname):
1068                 try: st = os.stat(fname)
1069                 except OSError: pass
1070                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]|0100))
1071         else:
1072             def do_chmod(fname):
1073                 try: st = os.stat(fname)
1074                 except OSError: pass
1075                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]&~0100))
1076
1077         if os.path.isfile(top):
1078             # If it's a file, that's easy, just chmod it.
1079             do_chmod(top)
1080         elif execute:
1081             # It's a directory and we're trying to turn on execute
1082             # permission, so it's also pretty easy, just chmod the
1083             # directory and then chmod every entry on our walk down the
1084             # tree.  Because os.path.walk() is top-down, we'll enable
1085             # execute permission on any directories that have it disabled
1086             # before os.path.walk() tries to list their contents.
1087             do_chmod(top)
1088
1089             def chmod_entries(arg, dirname, names, do_chmod=do_chmod):
1090                 pathnames = map(lambda n, d=dirname: os.path.join(d, n),
1091                                 names)
1092                 map(lambda p, do=do_chmod: do(p), pathnames)
1093
1094             os.path.walk(top, chmod_entries, None)
1095         else:
1096             # It's a directory and we're trying to turn off execute
1097             # permission, which means we have to chmod the directories
1098             # in the tree bottom-up, lest disabling execute permission from
1099             # the top down get in the way of being able to get at lower
1100             # parts of the tree.  But os.path.walk() visits things top
1101             # down, so we just use an object to collect a list of all
1102             # of the entries in the tree, reverse the list, and then
1103             # chmod the reversed (bottom-up) list.
1104             col = Collector(top)
1105             os.path.walk(top, col, None)
1106             col.entries.reverse()
1107             map(lambda d, do=do_chmod: do(d), col.entries)
1108
1109     def write(self, file, content, mode = 'wb'):
1110         """Writes the specified content text (second argument) to the
1111         specified file name (first argument).  The file name may be
1112         a list, in which case the elements are concatenated with the
1113         os.path.join() method.  The file is created under the temporary
1114         working directory.  Any subdirectories in the path must already
1115         exist.  The I/O mode for the file may be specified; it must
1116         begin with a 'w'.  The default is 'wb' (binary write).
1117         """
1118         file = self.canonicalize(file)
1119         if mode[0] != 'w':
1120             raise ValueError, "mode must begin with 'w'"
1121         open(file, mode).write(content)