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