ceph/src/boost/tools/build/test/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 of
   5 executable commands and scripts (in any language, not just Python), especially
   6 commands and scripts that require file system interaction.
   7
   8 In addition to running tests and evaluating conditions, the TestCmd module
   9 manages and cleans up one or more temporary workspace directories, and provides
  10 methods for creating files and directories in those workspace directories from
  11 in-line data, here-documents), allowing tests to be completely self-contained.
  12
  13 A TestCmd environment object is created via the usual invocation:
  14
  15     test = TestCmd()
  16
  17 The TestCmd module provides pass_test(), fail_test(), and no_result() unbound
  18 methods that report test results for use with the Aegis change management
  19 system. These methods terminate the test immediately, reporting PASSED, FAILED
  20 or NO RESULT respectively and exiting with status 0 (success), 1 or 2
  21 respectively. This allows for a distinction between an actual failed test and a
  22 test that could not be properly evaluated because of an external condition (such
  23 as a full file system or incorrect permissions).
  24
  25 """
  26
  27 # Copyright 2000 Steven Knight
  28 # This module is free software, and you may redistribute it and/or modify
  29 # it under the same terms as Python itself, so long as this copyright message
  30 # and disclaimer are retained in their original form.
  31 #
  32 # IN NO EVENT SHALL THE AUTHOR BE LIABLE TO ANY PARTY FOR DIRECT, INDIRECT,
  33 # SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE USE OF
  34 # THIS CODE, EVEN IF THE AUTHOR HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH
  35 # DAMAGE.
  36 #
  37 # THE AUTHOR SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT
  38 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
  39 # PARTICULAR PURPOSE.  THE CODE PROVIDED HEREUNDER IS ON AN "AS IS" BASIS,
  40 # AND THERE IS NO OBLIGATION WHATSOEVER TO PROVIDE MAINTENANCE,
  41 # SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
  42
  43 # Copyright 2002-2003 Vladimir Prus.
  44 # Copyright 2002-2003 Dave Abrahams.
  45 # Copyright 2006 Rene Rivera.
  46 # Distributed under the Boost Software License, Version 1.0.
  47 #    (See accompanying file LICENSE_1_0.txt or copy at
  48 #         http://www.boost.org/LICENSE_1_0.txt)
  49
  50
  51 from string import join, split
  52
  53 __author__ = "Steven Knight <knight@baldmt.com>"
  54 __revision__ = "TestCmd.py 0.D002 2001/08/31 14:56:12 software"
  55 __version__ = "0.02"
  56
  57 from types import *
  58
  59 import os
  60 import os.path
  61 import re
  62 import shutil
  63 import stat
  64 import subprocess
  65 import sys
  66 import tempfile
  67 import traceback
  68
  69
  70 tempfile.template = 'testcmd.'
  71
  72 _Cleanup = []
  73
  74 def _clean():
  75     global _Cleanup
  76     list = _Cleanup[:]
  77     _Cleanup = []
  78     list.reverse()
  79     for test in list:
  80         test.cleanup()
  81
  82 sys.exitfunc = _clean
  83
  84
  85 def caller(tblist, skip):
  86     string = ""
  87     arr = []
  88     for file, line, name, text in tblist:
  89         if file[-10:] == "TestCmd.py":
  90                 break
  91         arr = [(file, line, name, text)] + arr
  92     atfrom = "at"
  93     for file, line, name, text in arr[skip:]:
  94         if name == "?":
  95             name = ""
  96         else:
  97             name = " (" + name + ")"
  98         string = string + ("%s line %d of %s%s\n" % (atfrom, line, file, name))
  99         atfrom = "\tfrom"
 100     return string
 101
 102
 103 def fail_test(self=None, condition=True, function=None, skip=0):
 104     """Cause the test to fail.
 105
 106       By default, the fail_test() method reports that the test FAILED and exits
 107     with a status of 1. If a condition argument is supplied, the test fails
 108     only if the condition is true.
 109
 110     """
 111     if not condition:
 112         return
 113     if not function is None:
 114         function()
 115     of = ""
 116     desc = ""
 117     sep = " "
 118     if not self is None:
 119         if self.program:
 120             of = " of " + join(self.program, " ")
 121             sep = "\n\t"
 122         if self.description:
 123             desc = " [" + self.description + "]"
 124             sep = "\n\t"
 125
 126     at = caller(traceback.extract_stack(), skip)
 127
 128     sys.stderr.write("FAILED test" + of + desc + sep + at + """
 129 in directory: """ + os.getcwd() )
 130     sys.exit(1)
 131
 132
 133 def no_result(self=None, condition=True, function=None, skip=0):
 134     """Causes a test to exit with no valid result.
 135
 136       By default, the no_result() method reports NO RESULT for the test and
 137     exits with a status of 2. If a condition argument is supplied, the test
 138     fails only if the condition is true.
 139
 140     """
 141     if not condition:
 142         return
 143     if not function is None:
 144         function()
 145     of = ""
 146     desc = ""
 147     sep = " "
 148     if not self is None:
 149         if self.program:
 150             of = " of " + self.program
 151             sep = "\n\t"
 152         if self.description:
 153             desc = " [" + self.description + "]"
 154             sep = "\n\t"
 155
 156     at = caller(traceback.extract_stack(), skip)
 157     sys.stderr.write("NO RESULT for test" + of + desc + sep + at)
 158     sys.exit(2)
 159
 160
 161 def pass_test(self=None, condition=True, function=None):
 162     """Causes a test to pass.
 163
 164       By default, the pass_test() method reports PASSED for the test and exits
 165     with a status of 0. If a condition argument is supplied, the test passes
 166     only if the condition is true.
 167
 168     """
 169     if not condition:
 170         return
 171     if not function is None:
 172         function()
 173     sys.stderr.write("PASSED\n")
 174     sys.exit(0)
 175
 176
 177 def match_exact(lines=None, matches=None):
 178     """
 179       Returns whether the given lists or strings containing lines separated
 180     using newline characters contain exactly the same data.
 181
 182     """
 183     if not type(lines) is ListType:
 184         lines = split(lines, "\n")
 185     if not type(matches) is ListType:
 186         matches = split(matches, "\n")
 187     if len(lines) != len(matches):
 188         return
 189     for i in range(len(lines)):
 190         if lines[i] != matches[i]:
 191             return
 192     return 1
 193
 194
 195 def match_re(lines=None, res=None):
 196     """
 197       Given lists or strings contain lines separated using newline characters.
 198     This function matches those lines one by one, interpreting the lines in the
 199     res parameter as regular expressions.
 200
 201     """
 202     if not type(lines) is ListType:
 203         lines = split(lines, "\n")
 204     if not type(res) is ListType:
 205         res = split(res, "\n")
 206     if len(lines) != len(res):
 207         return
 208     for i in range(len(lines)):
 209         if not re.compile("^" + res[i] + "$").search(lines[i]):
 210             return
 211     return 1
 212
 213
 214 class TestCmd:
 215     def __init__(self, description=None, program=None, workdir=None,
 216         subdir=None, verbose=False, match=None, inpath=None):
 217
 218         self._cwd = os.getcwd()
 219         self.description_set(description)
 220         self.program_set(program, inpath)
 221         self.verbose_set(verbose)
 222         if match is None:
 223             self.match_func = match_re
 224         else:
 225             self.match_func = match
 226         self._dirlist = []
 227         self._preserve = {'pass_test': 0, 'fail_test': 0, 'no_result': 0}
 228         env = os.environ.get('PRESERVE')
 229         if env:
 230             self._preserve['pass_test'] = env
 231             self._preserve['fail_test'] = env
 232             self._preserve['no_result'] = env
 233         else:
 234             env = os.environ.get('PRESERVE_PASS')
 235             if env is not None:
 236                 self._preserve['pass_test'] = env
 237             env = os.environ.get('PRESERVE_FAIL')
 238             if env is not None:
 239                 self._preserve['fail_test'] = env
 240             env = os.environ.get('PRESERVE_PASS')
 241             if env is not None:
 242                 self._preserve['PRESERVE_NO_RESULT'] = env
 243         self._stdout = []
 244         self._stderr = []
 245         self.status = None
 246         self.condition = 'no_result'
 247         self.workdir_set(workdir)
 248         self.subdir(subdir)
 249
 250     def __del__(self):
 251         self.cleanup()
 252
 253     def __repr__(self):
 254         return "%x" % id(self)
 255
 256     def cleanup(self, condition=None):
 257         """
 258           Removes any temporary working directories for the specified TestCmd
 259         environment. If the environment variable PRESERVE was set when the
 260         TestCmd environment was created, temporary working directories are not
 261         removed. If any of the environment variables PRESERVE_PASS,
 262         PRESERVE_FAIL or PRESERVE_NO_RESULT were set when the TestCmd
 263         environment was created, then temporary working directories are not
 264         removed if the test passed, failed or had no result, respectively.
 265         Temporary working directories are also preserved for conditions
 266         specified via the preserve method.
 267
 268           Typically, this method is not called directly, but is used when the
 269         script exits to clean up temporary working directories as appropriate
 270         for the exit status.
 271
 272         """
 273         if not self._dirlist:
 274             return
 275         if condition is None:
 276             condition = self.condition
 277         if self._preserve[condition]:
 278             for dir in self._dirlist:
 279                 print("Preserved directory %s" % dir)
 280         else:
 281             list = self._dirlist[:]
 282             list.reverse()
 283             for dir in list:
 284                 self.writable(dir, 1)
 285                 shutil.rmtree(dir, ignore_errors=1)
 286
 287         self._dirlist = []
 288         self.workdir = None
 289         os.chdir(self._cwd)
 290         try:
 291             global _Cleanup
 292             _Cleanup.remove(self)
 293         except (AttributeError, ValueError):
 294             pass
 295
 296     def description_set(self, description):
 297         """Set the description of the functionality being tested."""
 298         self.description = description
 299
 300     def fail_test(self, condition=True, function=None, skip=0):
 301         """Cause the test to fail."""
 302         if not condition:
 303             return
 304         self.condition = 'fail_test'
 305         fail_test(self = self,
 306                   condition = condition,
 307                   function = function,
 308                   skip = skip)
 309
 310     def match(self, lines, matches):
 311         """Compare actual and expected file contents."""
 312         return self.match_func(lines, matches)
 313
 314     def match_exact(self, lines, matches):
 315         """Compare actual and expected file content exactly."""
 316         return match_exact(lines, matches)
 317
 318     def match_re(self, lines, res):
 319         """Compare file content with a regular expression."""
 320         return match_re(lines, res)
 321
 322     def no_result(self, condition=True, function=None, skip=0):
 323         """Report that the test could not be run."""
 324         if not condition:
 325             return
 326         self.condition = 'no_result'
 327         no_result(self = self,
 328                   condition = condition,
 329                   function = function,
 330                   skip = skip)
 331
 332     def pass_test(self, condition=True, function=None):
 333         """Cause the test to pass."""
 334         if not condition:
 335             return
 336         self.condition = 'pass_test'
 337         pass_test(self, condition, function)
 338
 339     def preserve(self, *conditions):
 340         """
 341           Arrange for the temporary working directories for the specified
 342         TestCmd environment to be preserved for one or more conditions. If no
 343         conditions are specified, arranges for the temporary working
 344         directories to be preserved for all conditions.
 345
 346         """
 347         if conditions is ():
 348             conditions = ('pass_test', 'fail_test', 'no_result')
 349         for cond in conditions:
 350             self._preserve[cond] = 1
 351
 352     def program_set(self, program, inpath):
 353         """Set the executable program or script to be tested."""
 354         if not inpath and program and not os.path.isabs(program[0]):
 355             program[0] = os.path.join(self._cwd, program[0])
 356         self.program = program
 357
 358     def read(self, file, mode='rb'):
 359         """
 360           Reads and returns the contents of the specified file name. The file
 361         name may be a list, in which case the elements are concatenated with
 362         the os.path.join() method. The file is assumed to be under the
 363         temporary working directory unless it is an absolute path name. The I/O
 364         mode for the file may be specified and must begin with an 'r'. The
 365         default is 'rb' (binary read).
 366
 367         """
 368         if type(file) is ListType:
 369             file = apply(os.path.join, tuple(file))
 370         if not os.path.isabs(file):
 371             file = os.path.join(self.workdir, file)
 372         if mode[0] != 'r':
 373             raise ValueError, "mode must begin with 'r'"
 374         return open(file, mode).read()
 375
 376     def run(self, program=None, arguments=None, chdir=None, stdin=None,
 377         universal_newlines=True):
 378         """
 379           Runs a test of the program or script for the test environment.
 380         Standard output and error output are saved for future retrieval via the
 381         stdout() and stderr() methods.
 382
 383           'universal_newlines' parameter controls how the child process
 384         input/output streams are opened as defined for the same named Python
 385         subprocess.POpen constructor parameter.
 386
 387         """
 388         if chdir:
 389             if not os.path.isabs(chdir):
 390                 chdir = os.path.join(self.workpath(chdir))
 391             if self.verbose:
 392                 sys.stderr.write("chdir(" + chdir + ")\n")
 393         else:
 394             chdir = self.workdir
 395
 396         cmd = []
 397         if program and program[0]:
 398             if program[0] != self.program[0] and not os.path.isabs(program[0]):
 399                 program[0] = os.path.join(self._cwd, program[0])
 400             cmd += program
 401         else:
 402             cmd += self.program
 403         if arguments:
 404             cmd += arguments.split(" ")
 405         if self.verbose:
 406             sys.stderr.write(join(cmd, " ") + "\n")
 407         p = subprocess.Popen(cmd, stdin=subprocess.PIPE,
 408             stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=chdir,
 409             universal_newlines=universal_newlines)
 410
 411         if stdin:
 412             if type(stdin) is ListType:
 413                 for line in stdin:
 414                     p.tochild.write(line)
 415             else:
 416                 p.tochild.write(stdin)
 417         out, err = p.communicate()
 418         self._stdout.append(out)
 419         self._stderr.append(err)
 420         self.status = p.returncode
 421
 422         if self.verbose:
 423             sys.stdout.write(self._stdout[-1])
 424             sys.stderr.write(self._stderr[-1])
 425
 426     def stderr(self, run=None):
 427         """
 428           Returns the error output from the specified run number. If there is
 429         no specified run number, then returns the error output of the last run.
 430         If the run number is less than zero, then returns the error output from
 431         that many runs back from the current run.
 432
 433         """
 434         if not run:
 435             run = len(self._stderr)
 436         elif run < 0:
 437             run = len(self._stderr) + run
 438         run -= 1
 439         if run < 0:
 440             return ''
 441         return self._stderr[run]
 442
 443     def stdout(self, run=None):
 444         """
 445           Returns the standard output from the specified run number. If there
 446         is no specified run number, then returns the standard output of the
 447         last run. If the run number is less than zero, then returns the
 448         standard output from that many runs back from the current run.
 449
 450         """
 451         if not run:
 452             run = len(self._stdout)
 453         elif run < 0:
 454             run = len(self._stdout) + run
 455         run -= 1
 456         if run < 0:
 457             return ''
 458         return self._stdout[run]
 459
 460     def subdir(self, *subdirs):
 461         """
 462           Create new subdirectories under the temporary working directory, one
 463         for each argument. An argument may be a list, in which case the list
 464         elements are concatenated using the os.path.join() method.
 465         Subdirectories multiple levels deep must be created using a separate
 466         argument for each level:
 467
 468             test.subdir('sub', ['sub', 'dir'], ['sub', 'dir', 'ectory'])
 469
 470         Returns the number of subdirectories actually created.
 471
 472         """
 473         count = 0
 474         for sub in subdirs:
 475             if sub is None:
 476                 continue
 477             if type(sub) is ListType:
 478                 sub = apply(os.path.join, tuple(sub))
 479             new = os.path.join(self.workdir, sub)
 480             try:
 481                 os.mkdir(new)
 482             except:
 483                 pass
 484             else:
 485                 count += 1
 486         return count
 487
 488     def unlink(self, file):
 489         """
 490           Unlinks the specified file name. The file name may be a list, in
 491         which case the elements are concatenated using the os.path.join()
 492         method. The file is assumed to be under the temporary working directory
 493         unless it is an absolute path name.
 494
 495         """
 496         if type(file) is ListType:
 497             file = apply(os.path.join, tuple(file))
 498         if not os.path.isabs(file):
 499             file = os.path.join(self.workdir, file)
 500         os.unlink(file)
 501
 502     def verbose_set(self, verbose):
 503         """Set the verbose level."""
 504         self.verbose = verbose
 505
 506     def workdir_set(self, path):
 507         """
 508           Creates a temporary working directory with the specified path name.
 509         If the path is a null string (''), a unique directory name is created.
 510
 511         """
 512         if os.path.isabs(path):
 513             self.workdir = path
 514         else:
 515             if path != None:
 516                 if path == '':
 517                     path = tempfile.mktemp()
 518                 if path != None:
 519                     os.mkdir(path)
 520                 self._dirlist.append(path)
 521                 global _Cleanup
 522                 try:
 523                     _Cleanup.index(self)
 524                 except ValueError:
 525                     _Cleanup.append(self)
 526                 # We would like to set self.workdir like this:
 527                 #     self.workdir = path
 528                 # But symlinks in the path will report things differently from
 529                 # os.getcwd(), so chdir there and back to fetch the canonical
 530                 # path.
 531                 cwd = os.getcwd()
 532                 os.chdir(path)
 533                 self.workdir = os.getcwd()
 534                 os.chdir(cwd)
 535             else:
 536                 self.workdir = None
 537
 538     def workpath(self, *args):
 539         """
 540           Returns the absolute path name to a subdirectory or file within the
 541         current temporary working directory. Concatenates the temporary working
 542         directory name with the specified arguments using os.path.join().
 543
 544         """
 545         return apply(os.path.join, (self.workdir,) + tuple(args))
 546
 547     def writable(self, top, write):
 548         """
 549           Make the specified directory tree writable (write == 1) or not
 550         (write == None).
 551
 552         """
 553         def _walk_chmod(arg, dirname, names):
 554             st = os.stat(dirname)
 555             os.chmod(dirname, arg(st[stat.ST_MODE]))
 556             for name in names:
 557                 fullname = os.path.join(dirname, name)
 558                 st = os.stat(fullname)
 559                 os.chmod(fullname, arg(st[stat.ST_MODE]))
 560
 561         _mode_writable = lambda mode: stat.S_IMODE(mode|0200)
 562         _mode_non_writable = lambda mode: stat.S_IMODE(mode&~0200)
 563
 564         if write:
 565             f = _mode_writable
 566         else:
 567             f = _mode_non_writable
 568         try:
 569             os.path.walk(top, _walk_chmod, f)
 570         except:
 571             pass  # Ignore any problems changing modes.
 572
 573     def write(self, file, content, mode='wb'):
 574         """
 575           Writes the specified content text (second argument) to the specified
 576         file name (first argument). The file name may be a list, in which case
 577         the elements are concatenated using the os.path.join() method. The file
 578         is created under the temporary working directory. Any subdirectories in
 579         the path must already exist. The I/O mode for the file may be specified
 580         and must begin with a 'w'. The default is 'wb' (binary write).
 581
 582         """
 583         if type(file) is ListType:
 584             file = apply(os.path.join, tuple(file))
 585         if not os.path.isabs(file):
 586             file = os.path.join(self.workdir, file)
 587         if mode[0] != 'w':
 588             raise ValueError, "mode must begin with 'w'"
 589         open(file, mode).write(content)