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 available at instantiation:
  20
  21     test = TestCmd.TestCmd(description = 'string',
  22                            program = 'program_or_script_to_test',
  23                            interpreter = 'script_interpreter',
  24                            workdir = 'prefix',
  25                            subdir = 'subdir',
  26                            verbose = Boolean,
  27                            match = default_match_function,
  28                            diff = default_diff_function,
  29                            combine = Boolean)
  30
  31 There are a bunch of methods that let you do different things:
  32
  33     test.verbose_set(1)
  34
  35     test.description_set('string')
  36
  37     test.program_set('program_or_script_to_test')
  38
  39     test.interpreter_set('script_interpreter')
  40     test.interpreter_set(['script_interpreter', 'arg'])
  41
  42     test.workdir_set('prefix')
  43     test.workdir_set('')
  44
  45     test.workpath('file')
  46     test.workpath('subdir', 'file')
  47
  48     test.subdir('subdir', ...)
  49
  50     test.rmdir('subdir', ...)
  51
  52     test.write('file', "contents\n")
  53     test.write(['subdir', 'file'], "contents\n")
  54
  55     test.read('file')
  56     test.read(['subdir', 'file'])
  57     test.read('file', mode)
  58     test.read(['subdir', 'file'], mode)
  59
  60     test.writable('dir', 1)
  61     test.writable('dir', None)
  62
  63     test.preserve(condition, ...)
  64
  65     test.cleanup(condition)
  66
  67     test.command_args(program = 'program_or_script_to_run',
  68                       interpreter = 'script_interpreter',
  69                       arguments = 'arguments to pass to program')
  70
  71     test.run(program = 'program_or_script_to_run',
  72              interpreter = 'script_interpreter',
  73              arguments = 'arguments to pass to program',
  74              chdir = 'directory_to_chdir_to',
  75              stdin = 'input to feed to the program\n')
  76              universal_newlines = True)
  77
  78     p = test.start(program = 'program_or_script_to_run',
  79                    interpreter = 'script_interpreter',
  80                    arguments = 'arguments to pass to program',
  81                    universal_newlines = None)
  82
  83     test.finish(self, p)
  84
  85     test.pass_test()
  86     test.pass_test(condition)
  87     test.pass_test(condition, function)
  88
  89     test.fail_test()
  90     test.fail_test(condition)
  91     test.fail_test(condition, function)
  92     test.fail_test(condition, function, skip)
  93
  94     test.no_result()
  95     test.no_result(condition)
  96     test.no_result(condition, function)
  97     test.no_result(condition, function, skip)
  98
  99     test.stdout()
 100     test.stdout(run)
 101
 102     test.stderr()
 103     test.stderr(run)
 104
 105     test.symlink(target, link)
 106
 107     test.banner(string)
 108     test.banner(string, width)
 109
 110     test.diff(actual, expected)
 111
 112     test.match(actual, expected)
 113
 114     test.match_exact("actual 1\nactual 2\n", "expected 1\nexpected 2\n")
 115     test.match_exact(["actual 1\n", "actual 2\n"],
 116                      ["expected 1\n", "expected 2\n"])
 117
 118     test.match_re("actual 1\nactual 2\n", regex_string)
 119     test.match_re(["actual 1\n", "actual 2\n"], list_of_regexes)
 120
 121     test.match_re_dotall("actual 1\nactual 2\n", regex_string)
 122     test.match_re_dotall(["actual 1\n", "actual 2\n"], list_of_regexes)
 123
 124     test.tempdir()
 125     test.tempdir('temporary-directory')
 126
 127     test.sleep()
 128     test.sleep(seconds)
 129
 130     test.where_is('foo')
 131     test.where_is('foo', 'PATH1:PATH2')
 132     test.where_is('foo', 'PATH1;PATH2', '.suffix3;.suffix4')
 133
 134     test.unlink('file')
 135     test.unlink('subdir', 'file')
 136
 137 The TestCmd module provides pass_test(), fail_test(), and no_result()
 138 unbound functions that report test results for use with the Aegis change
 139 management system.  These methods terminate the test immediately,
 140 reporting PASSED, FAILED, or NO RESULT respectively, and exiting with
 141 status 0 (success), 1 or 2 respectively.  This allows for a distinction
 142 between an actual failed test and a test that could not be properly
 143 evaluated because of an external condition (such as a full file system
 144 or incorrect permissions).
 145
 146     import TestCmd
 147
 148     TestCmd.pass_test()
 149     TestCmd.pass_test(condition)
 150     TestCmd.pass_test(condition, function)
 151
 152     TestCmd.fail_test()
 153     TestCmd.fail_test(condition)
 154     TestCmd.fail_test(condition, function)
 155     TestCmd.fail_test(condition, function, skip)
 156
 157     TestCmd.no_result()
 158     TestCmd.no_result(condition)
 159     TestCmd.no_result(condition, function)
 160     TestCmd.no_result(condition, function, skip)
 161
 162 The TestCmd module also provides unbound functions that handle matching
 163 in the same way as the match_*() methods described above.
 164
 165     import TestCmd
 166
 167     test = TestCmd.TestCmd(match = TestCmd.match_exact)
 168
 169     test = TestCmd.TestCmd(match = TestCmd.match_re)
 170
 171     test = TestCmd.TestCmd(match = TestCmd.match_re_dotall)
 172
 173 The TestCmd module provides unbound functions that can be used for the
 174 "diff" argument to TestCmd.TestCmd instantiation:
 175
 176     import TestCmd
 177
 178     test = TestCmd.TestCmd(match = TestCmd.match_re,
 179                            diff = TestCmd.diff_re)
 180
 181     test = TestCmd.TestCmd(diff = TestCmd.simple_diff)
 182
 183 The "diff" argument can also be used with standard difflib functions:
 184
 185     import difflib
 186
 187     test = TestCmd.TestCmd(diff = difflib.context_diff)
 188
 189     test = TestCmd.TestCmd(diff = difflib.unified_diff)
 190
 191 Lastly, the where_is() method also exists in an unbound function
 192 version.
 193
 194     import TestCmd
 195
 196     TestCmd.where_is('foo')
 197     TestCmd.where_is('foo', 'PATH1:PATH2')
 198     TestCmd.where_is('foo', 'PATH1;PATH2', '.suffix3;.suffix4')
 199 """
 200
 201 # Copyright 2000-2010 Steven Knight
 202 # This module is free software, and you may redistribute it and/or modify
 203 # it under the same terms as Python itself, so long as this copyright message
 204 # and disclaimer are retained in their original form.
 205 #
 206 # IN NO EVENT SHALL THE AUTHOR BE LIABLE TO ANY PARTY FOR DIRECT, INDIRECT,
 207 # SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE USE OF
 208 # THIS CODE, EVEN IF THE AUTHOR HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH
 209 # DAMAGE.
 210 #
 211 # THE AUTHOR SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT
 212 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
 213 # PARTICULAR PURPOSE.  THE CODE PROVIDED HEREUNDER IS ON AN "AS IS" BASIS,
 214 # AND THERE IS NO OBLIGATION WHATSOEVER TO PROVIDE MAINTENANCE,
 215 # SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 216 from __future__ import generators  ### KEEP FOR COMPATIBILITY FIXERS
 217
 218 __author__ = "Steven Knight <knight at baldmt dot com>"
 219 __revision__ = "TestCmd.py 0.37.D001 2010/01/11 16:55:50 knight"
 220 __version__ = "0.37"
 221
 222 import errno
 223 import os
 224 import os.path
 225 import re
 226 import shutil
 227 import stat
 228 import sys
 229 import tempfile
 230 import time
 231 import traceback
 232 import UserList
 233
 234 __all__ = [
 235     'diff_re',
 236     'fail_test',
 237     'no_result',
 238     'pass_test',
 239     'match_exact',
 240     'match_re',
 241     'match_re_dotall',
 242     'python_executable',
 243     'TestCmd'
 244 ]
 245
 246 try:
 247     import difflib
 248 except ImportError:
 249     __all__.append('simple_diff')
 250
 251 def is_List(e):
 252     return isinstance(e, list) \
 253         or isinstance(e, UserList.UserList)
 254
 255 try:
 256     from UserString import UserString
 257 except ImportError:
 258     class UserString:
 259         pass
 260
 261 try: unicode
 262 except NameError:
 263     def is_String(e):
 264         return isinstance(e, str) or isinstance(e, UserString)
 265 else:
 266     def is_String(e):
 267         return isinstance(e, str) \
 268             or isinstance(e, unicode) \
 269             or isinstance(e, UserString)
 270
 271 tempfile.template = 'testcmd.'
 272 if os.name in ('posix', 'nt'):
 273     tempfile.template = 'testcmd.' + str(os.getpid()) + '.'
 274 else:
 275     tempfile.template = 'testcmd.'
 276
 277 re_space = re.compile('\s')
 278
 279 _Cleanup = []
 280
 281 _chain_to_exitfunc = None
 282
 283 def _clean():
 284     global _Cleanup
 285     cleanlist = [_f for _f in _Cleanup if _f]
 286     del _Cleanup[:]
 287     cleanlist.reverse()
 288     for test in cleanlist:
 289         test.cleanup()
 290     if _chain_to_exitfunc:
 291         _chain_to_exitfunc()
 292
 293 try:
 294     import atexit
 295 except ImportError:
 296     # TODO(1.5): atexit requires python 2.0, so chain sys.exitfunc
 297     try:
 298         _chain_to_exitfunc = sys.exitfunc
 299     except AttributeError:
 300         pass
 301     sys.exitfunc = _clean
 302 else:
 303     atexit.register(_clean)
 304
 305 class Collector:
 306     def __init__(self, top):
 307         self.entries = [top]
 308     def __call__(self, arg, dirname, names):
 309         pathjoin = lambda n: os.path.join(dirname, n)
 310         self.entries.extend(list(map(pathjoin, names)))
 311
 312 def _caller(tblist, skip):
 313     string = ""
 314     arr = []
 315     for file, line, name, text in tblist:
 316         if file[-10:] == "TestCmd.py":
 317                 break
 318         arr = [(file, line, name, text)] + arr
 319     atfrom = "at"
 320     for file, line, name, text in arr[skip:]:
 321         if name in ("?", "<module>"):
 322             name = ""
 323         else:
 324             name = " (" + name + ")"
 325         string = string + ("%s line %d of %s%s\n" % (atfrom, line, file, name))
 326         atfrom = "\tfrom"
 327     return string
 328
 329 def fail_test(self = None, condition = 1, function = None, skip = 0):
 330     """Cause the test to fail.
 331
 332     By default, the fail_test() method reports that the test FAILED
 333     and exits with a status of 1.  If a condition argument is supplied,
 334     the test fails only if the condition is true.
 335     """
 336     if not condition:
 337         return
 338     if not function is None:
 339         function()
 340     of = ""
 341     desc = ""
 342     sep = " "
 343     if not self is None:
 344         if self.program:
 345             of = " of " + self.program
 346             sep = "\n\t"
 347         if self.description:
 348             desc = " [" + self.description + "]"
 349             sep = "\n\t"
 350
 351     at = _caller(traceback.extract_stack(), skip)
 352     sys.stderr.write("FAILED test" + of + desc + sep + at)
 353
 354     sys.exit(1)
 355
 356 def no_result(self = None, condition = 1, function = None, skip = 0):
 357     """Causes a test to exit with no valid result.
 358
 359     By default, the no_result() method reports NO RESULT for the test
 360     and exits with a status of 2.  If a condition argument is supplied,
 361     the test fails only if the condition is true.
 362     """
 363     if not condition:
 364         return
 365     if not function is None:
 366         function()
 367     of = ""
 368     desc = ""
 369     sep = " "
 370     if not self is None:
 371         if self.program:
 372             of = " of " + self.program
 373             sep = "\n\t"
 374         if self.description:
 375             desc = " [" + self.description + "]"
 376             sep = "\n\t"
 377
 378     at = _caller(traceback.extract_stack(), skip)
 379     sys.stderr.write("NO RESULT for test" + of + desc + sep + at)
 380
 381     sys.exit(2)
 382
 383 def pass_test(self = None, condition = 1, function = None):
 384     """Causes a test to pass.
 385
 386     By default, the pass_test() method reports PASSED for the test
 387     and exits with a status of 0.  If a condition argument is supplied,
 388     the test passes only if the condition is true.
 389     """
 390     if not condition:
 391         return
 392     if not function is None:
 393         function()
 394     sys.stderr.write("PASSED\n")
 395     sys.exit(0)
 396
 397 def match_exact(lines = None, matches = None):
 398     """
 399     """
 400     if not is_List(lines):
 401         lines = lines.split("\n")
 402     if not is_List(matches):
 403         matches = matches.split("\n")
 404     if len(lines) != len(matches):
 405         return
 406     for i in range(len(lines)):
 407         if lines[i] != matches[i]:
 408             return
 409     return 1
 410
 411 def match_re(lines = None, res = None):
 412     """
 413     """
 414     if not is_List(lines):
 415         lines = lines.split("\n")
 416     if not is_List(res):
 417         res = res.split("\n")
 418     if len(lines) != len(res):
 419         return
 420     for i in range(len(lines)):
 421         s = "^" + res[i] + "$"
 422         try:
 423             expr = re.compile(s)
 424         except re.error, e:
 425             msg = "Regular expression error in %s: %s"
 426             raise re.error, msg % (repr(s), e[0])
 427         if not expr.search(lines[i]):
 428             return
 429     return 1
 430
 431 def match_re_dotall(lines = None, res = None):
 432     """
 433     """
 434     if not isinstance(lines, str):
 435         lines = "\n".join(lines)
 436     if not isinstance(res, str):
 437         res = "\n".join(res)
 438     s = "^" + res + "$"
 439     try:
 440         expr = re.compile(s, re.DOTALL)
 441     except re.error, e:
 442         msg = "Regular expression error in %s: %s"
 443         raise re.error, msg % (repr(s), e[0])
 444     if expr.match(lines):
 445         return 1
 446
 447 try:
 448     import difflib
 449 except ImportError:
 450     pass
 451 else:
 452     def simple_diff(a, b, fromfile='', tofile='',
 453                     fromfiledate='', tofiledate='', n=3, lineterm='\n'):
 454         """
 455         A function with the same calling signature as difflib.context_diff
 456         (diff -c) and difflib.unified_diff (diff -u) but which prints
 457         output like the simple, unadorned 'diff" command.
 458         """
 459         sm = difflib.SequenceMatcher(None, a, b)
 460         def comma(x1, x2):
 461             return x1+1 == x2 and str(x2) or '%s,%s' % (x1+1, x2)
 462         result = []
 463         for op, a1, a2, b1, b2 in sm.get_opcodes():
 464             if op == 'delete':
 465                 result.append("%sd%d" % (comma(a1, a2), b1))
 466                 result.extend(['< ' + l for l in a[a1:a2]])
 467             elif op == 'insert':
 468                 result.append("%da%s" % (a1, comma(b1, b2)))
 469                 result.extend(['> ' + l for l in b[b1:b2]])
 470             elif op == 'replace':
 471                 result.append("%sc%s" % (comma(a1, a2), comma(b1, b2)))
 472                 result.extend(['< ' + l for l in a[a1:a2]])
 473                 result.append('---')
 474                 result.extend(['> ' + l for l in b[b1:b2]])
 475         return result
 476
 477 def diff_re(a, b, fromfile='', tofile='',
 478                 fromfiledate='', tofiledate='', n=3, lineterm='\n'):
 479     """
 480     A simple "diff" of two sets of lines when the expected lines
 481     are regular expressions.  This is a really dumb thing that
 482     just compares each line in turn, so it doesn't look for
 483     chunks of matching lines and the like--but at least it lets
 484     you know exactly which line first didn't compare correctl...
 485     """
 486     result = []
 487     diff = len(a) - len(b)
 488     if diff < 0:
 489         a = a + ['']*(-diff)
 490     elif diff > 0:
 491         b = b + ['']*diff
 492     i = 0
 493     for aline, bline in zip(a, b):
 494         s = "^" + aline + "$"
 495         try:
 496             expr = re.compile(s)
 497         except re.error, e:
 498             msg = "Regular expression error in %s: %s"
 499             raise re.error, msg % (repr(s), e[0])
 500         if not expr.search(bline):
 501             result.append("%sc%s" % (i+1, i+1))
 502             result.append('< ' + repr(a[i]))
 503             result.append('---')
 504             result.append('> ' + repr(b[i]))
 505         i = i+1
 506     return result
 507
 508 if os.name == 'java':
 509
 510     python_executable = os.path.join(sys.prefix, 'jython')
 511
 512 else:
 513
 514     python_executable = sys.executable
 515
 516 if sys.platform == 'win32':
 517
 518     default_sleep_seconds = 2
 519
 520     def where_is(file, path=None, pathext=None):
 521         if path is None:
 522             path = os.environ['PATH']
 523         if is_String(path):
 524             path = path.split(os.pathsep)
 525         if pathext is None:
 526             pathext = os.environ['PATHEXT']
 527         if is_String(pathext):
 528             pathext = pathext.split(os.pathsep)
 529         for ext in pathext:
 530             if ext.lower() == file[-len(ext):].lower():
 531                 pathext = ['']
 532                 break
 533         for dir in path:
 534             f = os.path.join(dir, file)
 535             for ext in pathext:
 536                 fext = f + ext
 537                 if os.path.isfile(fext):
 538                     return fext
 539         return None
 540
 541 else:
 542
 543     def where_is(file, path=None, pathext=None):
 544         if path is None:
 545             path = os.environ['PATH']
 546         if is_String(path):
 547             path = path.split(os.pathsep)
 548         for dir in path:
 549             f = os.path.join(dir, file)
 550             if os.path.isfile(f):
 551                 try:
 552                     st = os.stat(f)
 553                 except OSError:
 554                     continue
 555                 if stat.S_IMODE(st[stat.ST_MODE]) & 0111:
 556                     return f
 557         return None
 558
 559     default_sleep_seconds = 1
 560
 561
 562
 563 try:
 564     import subprocess
 565 except ImportError:
 566     # The subprocess module doesn't exist in this version of Python,
 567     # so we're going to cobble up something that looks just enough
 568     # like its API for our purposes below.
 569     import new
 570
 571     subprocess = new.module('subprocess')
 572
 573     subprocess.PIPE = 'PIPE'
 574     subprocess.STDOUT = 'STDOUT'
 575     subprocess.mswindows = (sys.platform == 'win32')
 576
 577     try:
 578         import popen2
 579         popen2.Popen3
 580     except AttributeError:
 581         class Popen3:
 582             universal_newlines = 1
 583             def __init__(self, command, **kw):
 584                 if sys.platform == 'win32' and command[0] == '"':
 585                     command = '"' + command + '"'
 586                 (stdin, stdout, stderr) = os.popen3(' ' + command)
 587                 self.stdin = stdin
 588                 self.stdout = stdout
 589                 self.stderr = stderr
 590             def close_output(self):
 591                 self.stdout.close()
 592                 self.resultcode = self.stderr.close()
 593             def wait(self):
 594                 resultcode = self.resultcode
 595                 if os.WIFEXITED(resultcode):
 596                     return os.WEXITSTATUS(resultcode)
 597                 elif os.WIFSIGNALED(resultcode):
 598                     return os.WTERMSIG(resultcode)
 599                 else:
 600                     return None
 601
 602     else:
 603         try:
 604             popen2.Popen4
 605         except AttributeError:
 606             # A cribbed Popen4 class, with some retrofitted code from
 607             # the Python 1.5 Popen3 class methods to do certain things
 608             # by hand.
 609             class Popen4(popen2.Popen3):
 610                 childerr = None
 611
 612                 def __init__(self, cmd, bufsize=-1):
 613                     p2cread, p2cwrite = os.pipe()
 614                     c2pread, c2pwrite = os.pipe()
 615                     self.pid = os.fork()
 616                     if self.pid == 0:
 617                         # Child
 618                         os.dup2(p2cread, 0)
 619                         os.dup2(c2pwrite, 1)
 620                         os.dup2(c2pwrite, 2)
 621                         for i in range(3, popen2.MAXFD):
 622                             try:
 623                                 os.close(i)
 624                             except: pass
 625                         try:
 626                             os.execvp(cmd[0], cmd)
 627                         finally:
 628                             os._exit(1)
 629                         # Shouldn't come here, I guess
 630                         os._exit(1)
 631                     os.close(p2cread)
 632                     self.tochild = os.fdopen(p2cwrite, 'w', bufsize)
 633                     os.close(c2pwrite)
 634                     self.fromchild = os.fdopen(c2pread, 'r', bufsize)
 635                     popen2._active.append(self)
 636
 637             popen2.Popen4 = Popen4
 638
 639         class Popen3(popen2.Popen3, popen2.Popen4):
 640             universal_newlines = 1
 641             def __init__(self, command, **kw):
 642                 if kw.get('stderr') == 'STDOUT':
 643                     popen2.Popen4.__init__(self, command, 1)
 644                 else:
 645                     popen2.Popen3.__init__(self, command, 1)
 646                 self.stdin = self.tochild
 647                 self.stdout = self.fromchild
 648                 self.stderr = self.childerr
 649             def wait(self, *args, **kw):
 650                 resultcode = popen2.Popen3.wait(self, *args, **kw)
 651                 if os.WIFEXITED(resultcode):
 652                     return os.WEXITSTATUS(resultcode)
 653                 elif os.WIFSIGNALED(resultcode):
 654                     return os.WTERMSIG(resultcode)
 655                 else:
 656                     return None
 657
 658     subprocess.Popen = Popen3
 659
 660
 661
 662 # From Josiah Carlson,
 663 # ASPN : Python Cookbook : Module to allow Asynchronous subprocess use on Windows and Posix platforms
 664 # http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/440554
 665
 666 PIPE = subprocess.PIPE
 667
 668 if subprocess.mswindows:
 669     from win32file import ReadFile, WriteFile
 670     from win32pipe import PeekNamedPipe
 671     import msvcrt
 672 else:
 673     import select
 674     import fcntl
 675
 676     try:                    fcntl.F_GETFL
 677     except AttributeError:  fcntl.F_GETFL = 3
 678
 679     try:                    fcntl.F_SETFL
 680     except AttributeError:  fcntl.F_SETFL = 4
 681
 682 class Popen(subprocess.Popen):
 683     def recv(self, maxsize=None):
 684         return self._recv('stdout', maxsize)
 685
 686     def recv_err(self, maxsize=None):
 687         return self._recv('stderr', maxsize)
 688
 689     def send_recv(self, input='', maxsize=None):
 690         return self.send(input), self.recv(maxsize), self.recv_err(maxsize)
 691
 692     def get_conn_maxsize(self, which, maxsize):
 693         if maxsize is None:
 694             maxsize = 1024
 695         elif maxsize < 1:
 696             maxsize = 1
 697         return getattr(self, which), maxsize
 698
 699     def _close(self, which):
 700         getattr(self, which).close()
 701         setattr(self, which, None)
 702
 703     if subprocess.mswindows:
 704         def send(self, input):
 705             if not self.stdin:
 706                 return None
 707
 708             try:
 709                 x = msvcrt.get_osfhandle(self.stdin.fileno())
 710                 (errCode, written) = WriteFile(x, input)
 711             except ValueError:
 712                 return self._close('stdin')
 713             except (subprocess.pywintypes.error, Exception), why:
 714                 if why[0] in (109, errno.ESHUTDOWN):
 715                     return self._close('stdin')
 716                 raise
 717
 718             return written
 719
 720         def _recv(self, which, maxsize):
 721             conn, maxsize = self.get_conn_maxsize(which, maxsize)
 722             if conn is None:
 723                 return None
 724
 725             try:
 726                 x = msvcrt.get_osfhandle(conn.fileno())
 727                 (read, nAvail, nMessage) = PeekNamedPipe(x, 0)
 728                 if maxsize < nAvail:
 729                     nAvail = maxsize
 730                 if nAvail > 0:
 731                     (errCode, read) = ReadFile(x, nAvail, None)
 732             except ValueError:
 733                 return self._close(which)
 734             except (subprocess.pywintypes.error, Exception), why:
 735                 if why[0] in (109, errno.ESHUTDOWN):
 736                     return self._close(which)
 737                 raise
 738
 739             #if self.universal_newlines:
 740             #    read = self._translate_newlines(read)
 741             return read
 742
 743     else:
 744         def send(self, input):
 745             if not self.stdin:
 746                 return None
 747
 748             if not select.select([], [self.stdin], [], 0)[1]:
 749                 return 0
 750
 751             try:
 752                 written = os.write(self.stdin.fileno(), input)
 753             except OSError, why:
 754                 if why[0] == errno.EPIPE: #broken pipe
 755                     return self._close('stdin')
 756                 raise
 757
 758             return written
 759
 760         def _recv(self, which, maxsize):
 761             conn, maxsize = self.get_conn_maxsize(which, maxsize)
 762             if conn is None:
 763                 return None
 764
 765             try:
 766                 flags = fcntl.fcntl(conn, fcntl.F_GETFL)
 767             except TypeError:
 768                 flags = None
 769             else:
 770                 if not conn.closed:
 771                     fcntl.fcntl(conn, fcntl.F_SETFL, flags| os.O_NONBLOCK)
 772
 773             try:
 774                 if not select.select([conn], [], [], 0)[0]:
 775                     return ''
 776
 777                 r = conn.read(maxsize)
 778                 if not r:
 779                     return self._close(which)
 780
 781                 #if self.universal_newlines:
 782                 #    r = self._translate_newlines(r)
 783                 return r
 784             finally:
 785                 if not conn.closed and not flags is None:
 786                     fcntl.fcntl(conn, fcntl.F_SETFL, flags)
 787
 788 disconnect_message = "Other end disconnected!"
 789
 790 def recv_some(p, t=.1, e=1, tr=5, stderr=0):
 791     if tr < 1:
 792         tr = 1
 793     x = time.time()+t
 794     y = []
 795     r = ''
 796     pr = p.recv
 797     if stderr:
 798         pr = p.recv_err
 799     while time.time() < x or r:
 800         r = pr()
 801         if r is None:
 802             if e:
 803                 raise Exception(disconnect_message)
 804             else:
 805                 break
 806         elif r:
 807             y.append(r)
 808         else:
 809             time.sleep(max((x-time.time())/tr, 0))
 810     return ''.join(y)
 811
 812 # TODO(3.0:  rewrite to use memoryview()
 813 def send_all(p, data):
 814     while len(data):
 815         sent = p.send(data)
 816         if sent is None:
 817             raise Exception(disconnect_message)
 818         data = buffer(data, sent)
 819
 820
 821
 822 try:
 823     object
 824 except NameError:
 825     class object:
 826         pass
 827
 828
 829
 830 class TestCmd(object):
 831     """Class TestCmd
 832     """
 833
 834     def __init__(self, description = None,
 835                        program = None,
 836                        interpreter = None,
 837                        workdir = None,
 838                        subdir = None,
 839                        verbose = None,
 840                        match = None,
 841                        diff = None,
 842                        combine = 0,
 843                        universal_newlines = 1):
 844         self._cwd = os.getcwd()
 845         self.description_set(description)
 846         self.program_set(program)
 847         self.interpreter_set(interpreter)
 848         if verbose is None:
 849             try:
 850                 verbose = max( 0, int(os.environ.get('TESTCMD_VERBOSE', 0)) )
 851             except ValueError:
 852                 verbose = 0
 853         self.verbose_set(verbose)
 854         self.combine = combine
 855         self.universal_newlines = universal_newlines
 856         if not match is None:
 857             self.match_function = match
 858         else:
 859             self.match_function = match_re
 860         if not diff is None:
 861             self.diff_function = diff
 862         else:
 863             try:
 864                 difflib
 865             except NameError:
 866                 pass
 867             else:
 868                 self.diff_function = simple_diff
 869                 #self.diff_function = difflib.context_diff
 870                 #self.diff_function = difflib.unified_diff
 871         self._dirlist = []
 872         self._preserve = {'pass_test': 0, 'fail_test': 0, 'no_result': 0}
 873         if 'PRESERVE' in os.environ and not os.environ['PRESERVE'] is '':
 874             self._preserve['pass_test'] = os.environ['PRESERVE']
 875             self._preserve['fail_test'] = os.environ['PRESERVE']
 876             self._preserve['no_result'] = os.environ['PRESERVE']
 877         else:
 878             try:
 879                 self._preserve['pass_test'] = os.environ['PRESERVE_PASS']
 880             except KeyError:
 881                 pass
 882             try:
 883                 self._preserve['fail_test'] = os.environ['PRESERVE_FAIL']
 884             except KeyError:
 885                 pass
 886             try:
 887                 self._preserve['no_result'] = os.environ['PRESERVE_NO_RESULT']
 888             except KeyError:
 889                 pass
 890         self._stdout = []
 891         self._stderr = []
 892         self.status = None
 893         self.condition = 'no_result'
 894         self.workdir_set(workdir)
 895         self.subdir(subdir)
 896
 897     def __del__(self):
 898         self.cleanup()
 899
 900     def __repr__(self):
 901         return "%x" % id(self)
 902
 903     banner_char = '='
 904     banner_width = 80
 905
 906     def banner(self, s, width=None):
 907         if width is None:
 908             width = self.banner_width
 909         return s + self.banner_char * (width - len(s))
 910
 911     if os.name == 'posix':
 912
 913         def escape(self, arg):
 914             "escape shell special characters"
 915             slash = '\\'
 916             special = '"$'
 917
 918             arg = arg.replace(slash, slash+slash)
 919             for c in special:
 920                 arg = arg.replace(c, slash+c)
 921
 922             if re_space.search(arg):
 923                 arg = '"' + arg + '"'
 924             return arg
 925
 926     else:
 927
 928         # Windows does not allow special characters in file names
 929         # anyway, so no need for an escape function, we will just quote
 930         # the arg.
 931         def escape(self, arg):
 932             if re_space.search(arg):
 933                 arg = '"' + arg + '"'
 934             return arg
 935
 936     def canonicalize(self, path):
 937         if is_List(path):
 938             path = os.path.join(*tuple(path))
 939         if not os.path.isabs(path):
 940             path = os.path.join(self.workdir, path)
 941         return path
 942
 943     def chmod(self, path, mode):
 944         """Changes permissions on the specified file or directory
 945         path name."""
 946         path = self.canonicalize(path)
 947         os.chmod(path, mode)
 948
 949     def cleanup(self, condition = None):
 950         """Removes any temporary working directories for the specified
 951         TestCmd environment.  If the environment variable PRESERVE was
 952         set when the TestCmd environment was created, temporary working
 953         directories are not removed.  If any of the environment variables
 954         PRESERVE_PASS, PRESERVE_FAIL, or PRESERVE_NO_RESULT were set
 955         when the TestCmd environment was created, then temporary working
 956         directories are not removed if the test passed, failed, or had
 957         no result, respectively.  Temporary working directories are also
 958         preserved for conditions specified via the preserve method.
 959
 960         Typically, this method is not called directly, but is used when
 961         the script exits to clean up temporary working directories as
 962         appropriate for the exit status.
 963         """
 964         if not self._dirlist:
 965             return
 966         os.chdir(self._cwd)
 967         self.workdir = None
 968         if condition is None:
 969             condition = self.condition
 970         if self._preserve[condition]:
 971             for dir in self._dirlist:
 972                 print "Preserved directory", dir
 973         else:
 974             list = self._dirlist[:]
 975             list.reverse()
 976             for dir in list:
 977                 self.writable(dir, 1)
 978                 shutil.rmtree(dir, ignore_errors = 1)
 979             self._dirlist = []
 980
 981         try:
 982             global _Cleanup
 983             _Cleanup.remove(self)
 984         except (AttributeError, ValueError):
 985             pass
 986
 987     def command_args(self, program = None,
 988                            interpreter = None,
 989                            arguments = None):
 990         if program:
 991             if isinstance(program, str) and not os.path.isabs(program):
 992                 program = os.path.join(self._cwd, program)
 993         else:
 994             program = self.program
 995             if not interpreter:
 996                 interpreter = self.interpreter
 997         if not type(program) in [list, tuple]:
 998             program = [program]
 999         cmd = list(program)
1000         if interpreter:
1001             if not type(interpreter) in [list, tuple]:
1002                 interpreter = [interpreter]
1003             cmd = list(interpreter) + cmd
1004         if arguments:
1005             if isinstance(arguments, str):
1006                 arguments = arguments.split()
1007             cmd.extend(arguments)
1008         return cmd
1009
1010     def description_set(self, description):
1011         """Set the description of the functionality being tested.
1012         """
1013         self.description = description
1014
1015     try:
1016         difflib
1017     except NameError:
1018         def diff(self, a, b, name, *args, **kw):
1019             print self.banner('Expected %s' % name)
1020             print a
1021             print self.banner('Actual %s' % name)
1022             print b
1023     else:
1024         def diff(self, a, b, name, *args, **kw):
1025             print self.banner(name)
1026             args = (a.splitlines(), b.splitlines()) + args
1027             lines = self.diff_function(*args, **kw)
1028             for l in lines:
1029                 print l
1030
1031     def fail_test(self, condition = 1, function = None, skip = 0):
1032         """Cause the test to fail.
1033         """
1034         if not condition:
1035             return
1036         self.condition = 'fail_test'
1037         fail_test(self = self,
1038                   condition = condition,
1039                   function = function,
1040                   skip = skip)
1041
1042     def interpreter_set(self, interpreter):
1043         """Set the program to be used to interpret the program
1044         under test as a script.
1045         """
1046         self.interpreter = interpreter
1047
1048     def match(self, lines, matches):
1049         """Compare actual and expected file contents.
1050         """
1051         return self.match_function(lines, matches)
1052
1053     def match_exact(self, lines, matches):
1054         """Compare actual and expected file contents.
1055         """
1056         return match_exact(lines, matches)
1057
1058     def match_re(self, lines, res):
1059         """Compare actual and expected file contents.
1060         """
1061         return match_re(lines, res)
1062
1063     def match_re_dotall(self, lines, res):
1064         """Compare actual and expected file contents.
1065         """
1066         return match_re_dotall(lines, res)
1067
1068     def no_result(self, condition = 1, function = None, skip = 0):
1069         """Report that the test could not be run.
1070         """
1071         if not condition:
1072             return
1073         self.condition = 'no_result'
1074         no_result(self = self,
1075                   condition = condition,
1076                   function = function,
1077                   skip = skip)
1078
1079     def pass_test(self, condition = 1, function = None):
1080         """Cause the test to pass.
1081         """
1082         if not condition:
1083             return
1084         self.condition = 'pass_test'
1085         pass_test(self = self, condition = condition, function = function)
1086
1087     def preserve(self, *conditions):
1088         """Arrange for the temporary working directories for the
1089         specified TestCmd environment to be preserved for one or more
1090         conditions.  If no conditions are specified, arranges for
1091         the temporary working directories to be preserved for all
1092         conditions.
1093         """
1094         if conditions is ():
1095             conditions = ('pass_test', 'fail_test', 'no_result')
1096         for cond in conditions:
1097             self._preserve[cond] = 1
1098
1099     def program_set(self, program):
1100         """Set the executable program or script to be tested.
1101         """
1102         if program and not os.path.isabs(program):
1103             program = os.path.join(self._cwd, program)
1104         self.program = program
1105
1106     def read(self, file, mode = 'rb'):
1107         """Reads and returns the contents of the specified file name.
1108         The file name may be a list, in which case the elements are
1109         concatenated with the os.path.join() method.  The file is
1110         assumed to be under the temporary working directory unless it
1111         is an absolute path name.  The I/O mode for the file may
1112         be specified; it must begin with an 'r'.  The default is
1113         'rb' (binary read).
1114         """
1115         file = self.canonicalize(file)
1116         if mode[0] != 'r':
1117             raise ValueError, "mode must begin with 'r'"
1118         return open(file, mode).read()
1119
1120     def rmdir(self, dir):
1121         """Removes the specified dir name.
1122         The dir name may be a list, in which case the elements are
1123         concatenated with the os.path.join() method.  The dir is
1124         assumed to be under the temporary working directory unless it
1125         is an absolute path name.
1126         The dir must be empty.
1127         """
1128         dir = self.canonicalize(dir)
1129         os.rmdir(dir)
1130
1131     def start(self, program = None,
1132                     interpreter = None,
1133                     arguments = None,
1134                     universal_newlines = None,
1135                     **kw):
1136         """
1137         Starts a program or script for the test environment.
1138
1139         The specified program will have the original directory
1140         prepended unless it is enclosed in a [list].
1141         """
1142         cmd = self.command_args(program, interpreter, arguments)
1143         cmd_string = ' '.join(map(self.escape, cmd))
1144         if self.verbose:
1145             sys.stderr.write(cmd_string + "\n")
1146         if universal_newlines is None:
1147             universal_newlines = self.universal_newlines
1148
1149         # On Windows, if we make stdin a pipe when we plan to send
1150         # no input, and the test program exits before
1151         # Popen calls msvcrt.open_osfhandle, that call will fail.
1152         # So don't use a pipe for stdin if we don't need one.
1153         stdin = kw.get('stdin', None)
1154         if stdin is not None:
1155             stdin = subprocess.PIPE
1156
1157         combine = kw.get('combine', self.combine)
1158         if combine:
1159             stderr_value = subprocess.STDOUT
1160         else:
1161             stderr_value = subprocess.PIPE
1162
1163         return Popen(cmd,
1164                      stdin=stdin,
1165                      stdout=subprocess.PIPE,
1166                      stderr=stderr_value,
1167                      universal_newlines=universal_newlines)
1168
1169     def finish(self, popen, **kw):
1170         """
1171         Finishes and waits for the process being run under control of
1172         the specified popen argument, recording the exit status,
1173         standard output and error output.
1174         """
1175         popen.stdin.close()
1176         self.status = popen.wait()
1177         if not self.status:
1178             self.status = 0
1179         self._stdout.append(popen.stdout.read())
1180         if popen.stderr:
1181             stderr = popen.stderr.read()
1182         else:
1183             stderr = ''
1184         self._stderr.append(stderr)
1185
1186     def run(self, program = None,
1187                   interpreter = None,
1188                   arguments = None,
1189                   chdir = None,
1190                   stdin = None,
1191                   universal_newlines = None):
1192         """Runs a test of the program or script for the test
1193         environment.  Standard output and error output are saved for
1194         future retrieval via the stdout() and stderr() methods.
1195
1196         The specified program will have the original directory
1197         prepended unless it is enclosed in a [list].
1198         """
1199         if chdir:
1200             oldcwd = os.getcwd()
1201             if not os.path.isabs(chdir):
1202                 chdir = os.path.join(self.workpath(chdir))
1203             if self.verbose:
1204                 sys.stderr.write("chdir(" + chdir + ")\n")
1205             os.chdir(chdir)
1206         p = self.start(program,
1207                        interpreter,
1208                        arguments,
1209                        universal_newlines,
1210                        stdin=stdin)
1211         if stdin:
1212             if is_List(stdin):
1213                 for line in stdin:
1214                     p.stdin.write(line)
1215             else:
1216                 p.stdin.write(stdin)
1217             p.stdin.close()
1218
1219         out = p.stdout.read()
1220         if p.stderr is None:
1221             err = ''
1222         else:
1223             err = p.stderr.read()
1224         try:
1225             close_output = p.close_output
1226         except AttributeError:
1227             p.stdout.close()
1228             if not p.stderr is None:
1229                 p.stderr.close()
1230         else:
1231             close_output()
1232
1233         self._stdout.append(out)
1234         self._stderr.append(err)
1235
1236         self.status = p.wait()
1237         if not self.status:
1238             self.status = 0
1239
1240         if chdir:
1241             os.chdir(oldcwd)
1242         if self.verbose >= 2:
1243             write = sys.stdout.write
1244             write('============ STATUS: %d\n' % self.status)
1245             out = self.stdout()
1246             if out or self.verbose >= 3:
1247                 write('============ BEGIN STDOUT (len=%d):\n' % len(out))
1248                 write(out)
1249                 write('============ END STDOUT\n')
1250             err = self.stderr()
1251             if err or self.verbose >= 3:
1252                 write('============ BEGIN STDERR (len=%d)\n' % len(err))
1253                 write(err)
1254                 write('============ END STDERR\n')
1255
1256     def sleep(self, seconds = default_sleep_seconds):
1257         """Sleeps at least the specified number of seconds.  If no
1258         number is specified, sleeps at least the minimum number of
1259         seconds necessary to advance file time stamps on the current
1260         system.  Sleeping more seconds is all right.
1261         """
1262         time.sleep(seconds)
1263
1264     def stderr(self, run = None):
1265         """Returns the error output from the specified run number.
1266         If there is no specified run number, then returns the error
1267         output of the last run.  If the run number is less than zero,
1268         then returns the error output from that many runs back from the
1269         current run.
1270         """
1271         if not run:
1272             run = len(self._stderr)
1273         elif run < 0:
1274             run = len(self._stderr) + run
1275         run = run - 1
1276         return self._stderr[run]
1277
1278     def stdout(self, run = None):
1279         """Returns the standard output from the specified run number.
1280         If there is no specified run number, then returns the standard
1281         output of the last run.  If the run number is less than zero,
1282         then returns the standard output from that many runs back from
1283         the current run.
1284         """
1285         if not run:
1286             run = len(self._stdout)
1287         elif run < 0:
1288             run = len(self._stdout) + run
1289         run = run - 1
1290         return self._stdout[run]
1291
1292     def subdir(self, *subdirs):
1293         """Create new subdirectories under the temporary working
1294         directory, one for each argument.  An argument may be a list,
1295         in which case the list elements are concatenated using the
1296         os.path.join() method.  Subdirectories multiple levels deep
1297         must be created using a separate argument for each level:
1298
1299                 test.subdir('sub', ['sub', 'dir'], ['sub', 'dir', 'ectory'])
1300
1301         Returns the number of subdirectories actually created.
1302         """
1303         count = 0
1304         for sub in subdirs:
1305             if sub is None:
1306                 continue
1307             if is_List(sub):
1308                 sub = os.path.join(*tuple(sub))
1309             new = os.path.join(self.workdir, sub)
1310             try:
1311                 os.mkdir(new)
1312             except OSError:
1313                 pass
1314             else:
1315                 count = count + 1
1316         return count
1317
1318     def symlink(self, target, link):
1319         """Creates a symlink to the specified target.
1320         The link name may be a list, in which case the elements are
1321         concatenated with the os.path.join() method.  The link is
1322         assumed to be under the temporary working directory unless it
1323         is an absolute path name. The target is *not* assumed to be
1324         under the temporary working directory.
1325         """
1326         link = self.canonicalize(link)
1327         os.symlink(target, link)
1328
1329     def tempdir(self, path=None):
1330         """Creates a temporary directory.
1331         A unique directory name is generated if no path name is specified.
1332         The directory is created, and will be removed when the TestCmd
1333         object is destroyed.
1334         """
1335         if path is None:
1336             try:
1337                 path = tempfile.mktemp(prefix=tempfile.template)
1338             except TypeError:
1339                 path = tempfile.mktemp()
1340         os.mkdir(path)
1341
1342         # Symlinks in the path will report things
1343         # differently from os.getcwd(), so chdir there
1344         # and back to fetch the canonical path.
1345         cwd = os.getcwd()
1346         try:
1347             os.chdir(path)
1348             path = os.getcwd()
1349         finally:
1350             os.chdir(cwd)
1351
1352         # Uppercase the drive letter since the case of drive
1353         # letters is pretty much random on win32:
1354         drive,rest = os.path.splitdrive(path)
1355         if drive:
1356             path = drive.upper() + rest
1357
1358         #
1359         self._dirlist.append(path)
1360         global _Cleanup
1361         try:
1362             _Cleanup.index(self)
1363         except ValueError:
1364             _Cleanup.append(self)
1365
1366         return path
1367
1368     def touch(self, path, mtime=None):
1369         """Updates the modification time on the specified file or
1370         directory path name.  The default is to update to the
1371         current time if no explicit modification time is specified.
1372         """
1373         path = self.canonicalize(path)
1374         atime = os.path.getatime(path)
1375         if mtime is None:
1376             mtime = time.time()
1377         os.utime(path, (atime, mtime))
1378
1379     def unlink(self, file):
1380         """Unlinks the specified file name.
1381         The file name may be a list, in which case the elements are
1382         concatenated with the os.path.join() method.  The file is
1383         assumed to be under the temporary working directory unless it
1384         is an absolute path name.
1385         """
1386         file = self.canonicalize(file)
1387         os.unlink(file)
1388
1389     def verbose_set(self, verbose):
1390         """Set the verbose level.
1391         """
1392         self.verbose = verbose
1393
1394     def where_is(self, file, path=None, pathext=None):
1395         """Find an executable file.
1396         """
1397         if is_List(file):
1398             file = os.path.join(*tuple(file))
1399         if not os.path.isabs(file):
1400             file = where_is(file, path, pathext)
1401         return file
1402
1403     def workdir_set(self, path):
1404         """Creates a temporary working directory with the specified
1405         path name.  If the path is a null string (''), a unique
1406         directory name is created.
1407         """
1408         if (path != None):
1409             if path == '':
1410                 path = None
1411             path = self.tempdir(path)
1412         self.workdir = path
1413
1414     def workpath(self, *args):
1415         """Returns the absolute path name to a subdirectory or file
1416         within the current temporary working directory.  Concatenates
1417         the temporary working directory name with the specified
1418         arguments using the os.path.join() method.
1419         """
1420         return os.path.join(self.workdir, *tuple(args))
1421
1422     def readable(self, top, read=1):
1423         """Make the specified directory tree readable (read == 1)
1424         or not (read == None).
1425
1426         This method has no effect on Windows systems, which use a
1427         completely different mechanism to control file readability.
1428         """
1429
1430         if sys.platform == 'win32':
1431             return
1432
1433         if read:
1434             def do_chmod(fname):
1435                 try: st = os.stat(fname)
1436                 except OSError: pass
1437                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]|stat.S_IREAD))
1438         else:
1439             def do_chmod(fname):
1440                 try: st = os.stat(fname)
1441                 except OSError: pass
1442                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]&~stat.S_IREAD))
1443
1444         if os.path.isfile(top):
1445             # If it's a file, that's easy, just chmod it.
1446             do_chmod(top)
1447         elif read:
1448             # It's a directory and we're trying to turn on read
1449             # permission, so it's also pretty easy, just chmod the
1450             # directory and then chmod every entry on our walk down the
1451             # tree.  Because os.path.walk() is top-down, we'll enable
1452             # read permission on any directories that have it disabled
1453             # before os.path.walk() tries to list their contents.
1454             do_chmod(top)
1455
1456             def chmod_entries(arg, dirname, names, do_chmod=do_chmod):
1457                 for n in names:
1458                     do_chmod(os.path.join(dirname, n))
1459
1460             os.path.walk(top, chmod_entries, None)
1461         else:
1462             # It's a directory and we're trying to turn off read
1463             # permission, which means we have to chmod the directoreis
1464             # in the tree bottom-up, lest disabling read permission from
1465             # the top down get in the way of being able to get at lower
1466             # parts of the tree.  But os.path.walk() visits things top
1467             # down, so we just use an object to collect a list of all
1468             # of the entries in the tree, reverse the list, and then
1469             # chmod the reversed (bottom-up) list.
1470             col = Collector(top)
1471             os.path.walk(top, col, None)
1472             col.entries.reverse()
1473             for d in col.entries: do_chmod(d)
1474
1475     def writable(self, top, write=1):
1476         """Make the specified directory tree writable (write == 1)
1477         or not (write == None).
1478         """
1479
1480         if sys.platform == 'win32':
1481
1482             if write:
1483                 def do_chmod(fname):
1484                     try: os.chmod(fname, stat.S_IWRITE)
1485                     except OSError: pass
1486             else:
1487                 def do_chmod(fname):
1488                     try: os.chmod(fname, stat.S_IREAD)
1489                     except OSError: pass
1490
1491         else:
1492
1493             if write:
1494                 def do_chmod(fname):
1495                     try: st = os.stat(fname)
1496                     except OSError: pass
1497                     else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]|0200))
1498             else:
1499                 def do_chmod(fname):
1500                     try: st = os.stat(fname)
1501                     except OSError: pass
1502                     else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]&~0200))
1503
1504         if os.path.isfile(top):
1505             do_chmod(top)
1506         else:
1507             col = Collector(top)
1508             os.path.walk(top, col, None)
1509             for d in col.entries: do_chmod(d)
1510
1511     def executable(self, top, execute=1):
1512         """Make the specified directory tree executable (execute == 1)
1513         or not (execute == None).
1514
1515         This method has no effect on Windows systems, which use a
1516         completely different mechanism to control file executability.
1517         """
1518
1519         if sys.platform == 'win32':
1520             return
1521
1522         if execute:
1523             def do_chmod(fname):
1524                 try: st = os.stat(fname)
1525                 except OSError: pass
1526                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]|stat.S_IEXEC))
1527         else:
1528             def do_chmod(fname):
1529                 try: st = os.stat(fname)
1530                 except OSError: pass
1531                 else: os.chmod(fname, stat.S_IMODE(st[stat.ST_MODE]&~stat.S_IEXEC))
1532
1533         if os.path.isfile(top):
1534             # If it's a file, that's easy, just chmod it.
1535             do_chmod(top)
1536         elif execute:
1537             # It's a directory and we're trying to turn on execute
1538             # permission, so it's also pretty easy, just chmod the
1539             # directory and then chmod every entry on our walk down the
1540             # tree.  Because os.path.walk() is top-down, we'll enable
1541             # execute permission on any directories that have it disabled
1542             # before os.path.walk() tries to list their contents.
1543             do_chmod(top)
1544
1545             def chmod_entries(arg, dirname, names, do_chmod=do_chmod):
1546                 for n in names:
1547                     do_chmod(os.path.join(dirname, n))
1548
1549             os.path.walk(top, chmod_entries, None)
1550         else:
1551             # It's a directory and we're trying to turn off execute
1552             # permission, which means we have to chmod the directories
1553             # in the tree bottom-up, lest disabling execute permission from
1554             # the top down get in the way of being able to get at lower
1555             # parts of the tree.  But os.path.walk() visits things top
1556             # down, so we just use an object to collect a list of all
1557             # of the entries in the tree, reverse the list, and then
1558             # chmod the reversed (bottom-up) list.
1559             col = Collector(top)
1560             os.path.walk(top, col, None)
1561             col.entries.reverse()
1562             for d in col.entries: do_chmod(d)
1563
1564     def write(self, file, content, mode = 'wb'):
1565         """Writes the specified content text (second argument) to the
1566         specified file name (first argument).  The file name may be
1567         a list, in which case the elements are concatenated with the
1568         os.path.join() method.  The file is created under the temporary
1569         working directory.  Any subdirectories in the path must already
1570         exist.  The I/O mode for the file may be specified; it must
1571         begin with a 'w'.  The default is 'wb' (binary write).
1572         """
1573         file = self.canonicalize(file)
1574         if mode[0] != 'w':
1575             raise ValueError, "mode must begin with 'w'"
1576         open(file, mode).write(content)
1577
1578 # Local Variables:
1579 # tab-width:4
1580 # indent-tabs-mode:nil
1581 # End:
1582 # vim: set expandtab tabstop=4 shiftwidth=4: