import HEAD into svn+ssh://svn.samba.org/home/svn/samba/trunk
[metze/old/v3-2-winbind-ndr.git] / source / stf / comfychair.py
1 #! /usr/bin/env python
2
3 # Copyright (C) 2002, 2003 by Martin Pool <mbp@samba.org>
4 # Copyright (C) 2003 by Tim Potter <tpot@samba.org>
5
6 # This program is free software; you can redistribute it and/or
7 # modify it under the terms of the GNU General Public License as
8 # published by the Free Software Foundation; either version 2 of the
9 # License, or (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14 # General Public License for more details.
15
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
19 # USA
20
21 """comfychair: a Python-based instrument of software torture.
22
23 Copyright (C) 2002, 2003 by Martin Pool <mbp@samba.org>
24 Copyright (C) 2003 by Tim Potter <tpot@samba.org>
25
26 This is a test framework designed for testing programs written in
27 Python, or (through a fork/exec interface) any other language.
28
29 For more information, see the file README.comfychair.
30
31 To run a test suite based on ComfyChair, just run it as a program.
32 """
33
34 import sys, re
35
36
37 class TestCase:
38     """A base class for tests.  This class defines required functions which
39     can optionally be overridden by subclasses.  It also provides some
40     utility functions for"""
41
42     def __init__(self):
43         self.test_log = ""
44         self.background_pids = []
45         self._cleanups = []
46         self._enter_rundir()
47         self._save_environment()
48         self.add_cleanup(self.teardown)
49
50
51     # --------------------------------------------------
52     # Save and restore directory
53     def _enter_rundir(self):
54         import os
55         self.basedir = os.getcwd()
56         self.add_cleanup(self._restore_directory)
57         self.rundir = os.path.join(self.basedir,
58                                    'testtmp', 
59                                    self.__class__.__name__)
60         self.tmpdir = os.path.join(self.rundir, 'tmp')
61         os.system("rm -fr %s" % self.rundir)
62         os.makedirs(self.tmpdir)
63         os.system("mkdir -p %s" % self.rundir)
64         os.chdir(self.rundir)
65
66     def _restore_directory(self):
67         import os
68         os.chdir(self.basedir)
69
70     # --------------------------------------------------
71     # Save and restore environment
72     def _save_environment(self):
73         import os
74         self._saved_environ = os.environ.copy()
75         self.add_cleanup(self._restore_environment)
76
77     def _restore_environment(self):
78         import os
79         os.environ.clear()
80         os.environ.update(self._saved_environ)
81
82     
83     def setup(self):
84         """Set up test fixture."""
85         pass
86
87     def teardown(self):
88         """Tear down test fixture."""
89         pass
90
91     def runtest(self):
92         """Run the test."""
93         pass
94
95
96     def add_cleanup(self, c):
97         """Queue a cleanup to be run when the test is complete."""
98         self._cleanups.append(c)
99         
100
101     def fail(self, reason = ""):
102         """Say the test failed."""
103         raise AssertionError(reason)
104
105
106     #############################################################
107     # Requisition methods
108
109     def require(self, predicate, message):
110         """Check a predicate for running this test.
111
112 If the predicate value is not true, the test is skipped with a message explaining
113 why."""
114         if not predicate:
115             raise NotRunError, message
116
117     def require_root(self):
118         """Skip this test unless run by root."""
119         import os
120         self.require(os.getuid() == 0,
121                      "must be root to run this test")
122
123     #############################################################
124     # Assertion methods
125
126     def assert_(self, expr, reason = ""):
127         if not expr:
128             raise AssertionError(reason)
129
130     def assert_equal(self, a, b):
131         if not a == b:
132             raise AssertionError("assertEquals failed: %s" % `(a, b)`)
133             
134     def assert_notequal(self, a, b):
135         if a == b:
136             raise AssertionError("assertNotEqual failed: %s" % `(a, b)`)
137
138     def assert_re_match(self, pattern, s):
139         """Assert that a string matches a particular pattern
140
141         Inputs:
142           pattern      string: regular expression
143           s            string: to be matched
144
145         Raises:
146           AssertionError if not matched
147           """
148         if not re.match(pattern, s):
149             raise AssertionError("string does not match regexp\n"
150                                  "    string: %s\n"
151                                  "    re: %s" % (`s`, `pattern`))
152
153     def assert_re_search(self, pattern, s):
154         """Assert that a string *contains* a particular pattern
155
156         Inputs:
157           pattern      string: regular expression
158           s            string: to be searched
159
160         Raises:
161           AssertionError if not matched
162           """
163         if not re.search(pattern, s):
164             raise AssertionError("string does not contain regexp\n"
165                                  "    string: %s\n"
166                                  "    re: %s" % (`s`, `pattern`))
167
168
169     def assert_no_file(self, filename):
170         import os.path
171         assert not os.path.exists(filename), ("file exists but should not: %s" % filename)
172
173
174     #############################################################
175     # Methods for running programs
176
177     def runcmd_background(self, cmd):
178         import os
179         self.test_log = self.test_log + "Run in background:\n" + `cmd` + "\n"
180         pid = os.fork()
181         if pid == 0:
182             # child
183             try:
184                 os.execvp("/bin/sh", ["/bin/sh", "-c", cmd])
185             finally:
186                 os._exit(127)
187         self.test_log = self.test_log + "pid: %d\n" % pid
188         return pid
189
190
191     def runcmd(self, cmd, expectedResult = 0):
192         """Run a command, fail if the command returns an unexpected exit
193         code.  Return the output produced."""
194         rc, output, stderr = self.runcmd_unchecked(cmd)
195         if rc != expectedResult:
196             raise AssertionError("""command returned %d; expected %s: \"%s\"
197 stdout:
198 %s
199 stderr:
200 %s""" % (rc, expectedResult, cmd, output, stderr))
201
202         return output, stderr
203
204
205     def run_captured(self, cmd):
206         """Run a command, capturing stdout and stderr.
207
208         Based in part on popen2.py
209
210         Returns (waitstatus, stdout, stderr)."""
211         import os, types
212         pid = os.fork()
213         if pid == 0:
214             # child
215             try: 
216                 pid = os.getpid()
217                 openmode = os.O_WRONLY|os.O_CREAT|os.O_TRUNC
218
219                 outfd = os.open('%d.out' % pid, openmode, 0666)
220                 os.dup2(outfd, 1)
221                 os.close(outfd)
222
223                 errfd = os.open('%d.err' % pid, openmode, 0666)
224                 os.dup2(errfd, 2)
225                 os.close(errfd)
226
227                 if isinstance(cmd, types.StringType):
228                     cmd = ['/bin/sh', '-c', cmd]
229
230                 os.execvp(cmd[0], cmd)
231             finally:
232                 os._exit(127)
233         else:
234             # parent
235             exited_pid, waitstatus = os.waitpid(pid, 0)
236             stdout = open('%d.out' % pid).read()
237             stderr = open('%d.err' % pid).read()
238             return waitstatus, stdout, stderr
239
240
241     def runcmd_unchecked(self, cmd, skip_on_noexec = 0):
242         """Invoke a command; return (exitcode, stdout, stderr)"""
243         import os
244         waitstatus, stdout, stderr = self.run_captured(cmd)
245         assert not os.WIFSIGNALED(waitstatus), \
246                ("%s terminated with signal %d" % (`cmd`, os.WTERMSIG(waitstatus)))
247         rc = os.WEXITSTATUS(waitstatus)
248         self.test_log = self.test_log + ("""Run command: %s
249 Wait status: %#x (exit code %d, signal %d)
250 stdout:
251 %s
252 stderr:
253 %s""" % (cmd, waitstatus, os.WEXITSTATUS(waitstatus), os.WTERMSIG(waitstatus),
254          stdout, stderr))
255         if skip_on_noexec and rc == 127:
256             # Either we could not execute the command or the command
257             # returned exit code 127.  According to system(3) we can't
258             # tell the difference.
259             raise NotRunError, "could not execute %s" % `cmd`
260         return rc, stdout, stderr
261     
262
263     def explain_failure(self, exc_info = None):
264         print "test_log:"
265         print self.test_log
266
267
268     def log(self, msg):
269         """Log a message to the test log.  This message is displayed if
270         the test fails, or when the runtests function is invoked with
271         the verbose option."""
272         self.test_log = self.test_log + msg + "\n"
273
274
275 class NotRunError(Exception):
276     """Raised if a test must be skipped because of missing resources"""
277     def __init__(self, value = None):
278         self.value = value
279
280
281 def _report_error(case, debugger):
282     """Ask the test case to explain failure, and optionally run a debugger
283
284     Input:
285       case         TestCase instance
286       debugger     if true, a debugger function to be applied to the traceback
287 """
288     import sys
289     ex = sys.exc_info()
290     print "-----------------------------------------------------------------"
291     if ex:
292         import traceback
293         traceback.print_exc(file=sys.stdout)
294     case.explain_failure()
295     print "-----------------------------------------------------------------"
296
297     if debugger:
298         tb = ex[2]
299         debugger(tb)
300
301
302 def runtests(test_list, verbose = 0, debugger = None):
303     """Run a series of tests.
304
305     Inputs:
306       test_list    sequence of TestCase classes
307       verbose      print more information as testing proceeds
308       debugger     debugger object to be applied to errors
309
310     Returns:
311       unix return code: 0 for success, 1 for failures, 2 for test failure
312     """
313     import traceback
314     ret = 0
315     for test_class in test_list:
316         print "%-30s" % _test_name(test_class),
317         # flush now so that long running tests are easier to follow
318         sys.stdout.flush()
319
320         obj = None
321         try:
322             try: # run test and show result
323                 obj = test_class()
324                 obj.setup()
325                 obj.runtest()
326                 print "OK"
327             except KeyboardInterrupt:
328                 print "INTERRUPT"
329                 _report_error(obj, debugger)
330                 ret = 2
331                 break
332             except NotRunError, msg:
333                 print "NOTRUN, %s" % msg.value
334             except:
335                 print "FAIL"
336                 _report_error(obj, debugger)
337                 ret = 1
338         finally:
339             while obj and obj._cleanups:
340                 try:
341                     apply(obj._cleanups.pop())
342                 except KeyboardInterrupt:
343                     print "interrupted during teardown"
344                     _report_error(obj, debugger)
345                     ret = 2
346                     break
347                 except:
348                     print "error during teardown"
349                     _report_error(obj, debugger)
350                     ret = 1
351         # Display log file if we're verbose
352         if ret == 0 and verbose:
353             obj.explain_failure()
354             
355     return ret
356
357
358 def _test_name(test_class):
359     """Return a human-readable name for a test class.
360     """
361     try:
362         return test_class.__name__
363     except:
364         return `test_class`
365
366
367 def print_help():
368     """Help for people running tests"""
369     import sys
370     print """%s: software test suite based on ComfyChair
371
372 usage:
373     To run all tests, just run this program.  To run particular tests,
374     list them on the command line.
375
376 options:
377     --help              show usage message
378     --list              list available tests
379     --verbose, -v       show more information while running tests
380     --post-mortem, -p   enter Python debugger on error
381 """ % sys.argv[0]
382
383
384 def print_list(test_list):
385     """Show list of available tests"""
386     for test_class in test_list:
387         print "    %s" % _test_name(test_class)
388
389
390 def main(tests, extra_tests=[]):
391     """Main entry point for test suites based on ComfyChair.
392
393     inputs:
394       tests       Sequence of TestCase subclasses to be run by default.
395       extra_tests Sequence of TestCase subclasses that are available but
396                   not run by default.
397
398 Test suites should contain this boilerplate:
399
400     if __name__ == '__main__':
401         comfychair.main(tests)
402
403 This function handles standard options such as --help and --list, and
404 by default runs all tests in the suggested order.
405
406 Calls sys.exit() on completion.
407 """
408     from sys import argv
409     import getopt, sys
410
411     opt_verbose = 0
412     debugger = None
413
414     opts, args = getopt.getopt(argv[1:], 'pv',
415                                ['help', 'list', 'verbose', 'post-mortem'])
416     for opt, opt_arg in opts:
417         if opt == '--help':
418             print_help()
419             return
420         elif opt == '--list':
421             print_list(tests + extra_tests)
422             return
423         elif opt == '--verbose' or opt == '-v':
424             opt_verbose = 1
425         elif opt == '--post-mortem' or opt == '-p':
426             import pdb
427             debugger = pdb.post_mortem
428
429     if args:
430         all_tests = tests + extra_tests
431         by_name = {}
432         for t in all_tests:
433             by_name[_test_name(t)] = t
434         which_tests = []
435         for name in args:
436             which_tests.append(by_name[name])
437     else:
438         which_tests = tests
439
440     sys.exit(runtests(which_tests, verbose=opt_verbose,
441                       debugger=debugger))
442
443
444 if __name__ == '__main__':
445     print __doc__