1""" 2TestCmd.py: a testing framework for commands and scripts. 3 4The TestCmd module provides a framework for portable automated testing of 5executable commands and scripts (in any language, not just Python), especially 6commands and scripts that require file system interaction. 7 8In addition to running tests and evaluating conditions, the TestCmd module 9manages and cleans up one or more temporary workspace directories, and provides 10methods for creating files and directories in those workspace directories from 11in-line data, here-documents), allowing tests to be completely self-contained. 12 13A TestCmd environment object is created via the usual invocation: 14 15 test = TestCmd() 16 17The TestCmd module provides pass_test(), fail_test(), and no_result() unbound 18methods that report test results for use with the Aegis change management 19system. These methods terminate the test immediately, reporting PASSED, FAILED 20or NO RESULT respectively and exiting with status 0 (success), 1 or 2 21respectively. This allows for a distinction between an actual failed test and a 22test that could not be properly evaluated because of an external condition (such 23as 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 51from 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 57from types import * 58 59import os 60import os.path 61import re 62import shutil 63import stat 64import subprocess 65import sys 66import tempfile 67import traceback 68 69 70tempfile.template = 'testcmd.' 71 72_Cleanup = [] 73 74def _clean(): 75 global _Cleanup 76 list = _Cleanup[:] 77 _Cleanup = [] 78 list.reverse() 79 for test in list: 80 test.cleanup() 81 82sys.exitfunc = _clean 83 84 85def 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 103def 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 + """ 129in directory: """ + os.getcwd() ) 130 sys.exit(1) 131 132 133def 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 161def 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 177def 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 195def 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 214class 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) 590