← Back to branch summary

~ubuntu-branches/debian/experimental/distcc/experimental

« back to all changes in this revision

Viewing changes to .pc/14_test-reliability.patch/test/comfychair.py

  • Committer: Package Import Robot
  • Author(s): Daniel Hartwig, Daniel Hartwig
  • Date: 2013-05-27 20:36:36 UTC
  • Revision ID: package-import@ubuntu.com-20130527203636-72ov7ez1pxmo0rdq
Tags: 3.2~rc1-2
http://bugs.debian.org/709169
[ Daniel Hartwig ]
* do not run tests in parallel
* debian/patches: 
  - 12_test-debian.patch: NoDetachDaemon_Case is broken and leaves an
    orphaned process; disabled to keep buildd happy (Closes: #709169)
  - 14_test-reliability.patch: improve reliability of tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
.pc/14_test-reliability.patch/test/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
 
8
# as published by the Free Software Foundation; either version 2
 
9
# of the License, or (at your option) any later version.
 
10
#
 
11
# This program is distributed in the hope that it will be useful,
 
12
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
14
# GNU 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., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301,
 
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, shutil
 
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
        shutil.rmtree(self.rundir, ignore_errors=1)
 
62
        os.makedirs(self.tmpdir)
 
63
        os.chdir(self.rundir)
 
64
        os.environ['LANG'] = 'C'
 
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
    def apply_cleanups(self, debugger):
 
101
        """Apply cleanup functions and return error code.
 
102
 
 
103
        Returns:
 
104
          0 on success; 2 if a KeyboardInterrupt occurs; 1 if any other exception
 
105
          occurs.
 
106
        """
 
107
        while self._cleanups:
 
108
            try:
 
109
                apply(self._cleanups.pop())
 
110
            except KeyboardInterrupt:
 
111
                print "interrupted during cleanups"
 
112
                _report_error(self, debugger)
 
113
                return 2
 
114
            except:
 
115
                print "error during cleanups"
 
116
                _report_error(self, debugger)
 
117
                return 1
 
118
        return 0
 
119
 
 
120
    def fail(self, reason = ""):
 
121
        """Say the test failed."""
 
122
        raise AssertionError(reason)
 
123
 
 
124
 
 
125
    #############################################################
 
126
    # Requisition methods
 
127
 
 
128
    def require(self, predicate, message):
 
129
        """Check a predicate for running this test.
 
130
 
 
131
If the predicate value is not true, the test is skipped with a message explaining
 
132
why."""
 
133
        if not predicate:
 
134
            raise NotRunError, message
 
135
 
 
136
    def require_root(self):
 
137
        """Skip this test unless run by root."""
 
138
        import os
 
139
        self.require(os.getuid() == 0,
 
140
                     "must be root to run this test")
 
141
 
 
142
    #############################################################
 
143
    # Assertion methods
 
144
 
 
145
    def assert_(self, expr, reason = ""):
 
146
        if not expr:
 
147
            raise AssertionError(reason)
 
148
 
 
149
    def assert_equal(self, a, b):
 
150
        if not a == b:
 
151
            raise AssertionError("assertEquals failed: %s" % `(a, b)`)
 
152
            
 
153
    def assert_notequal(self, a, b):
 
154
        if a == b:
 
155
            raise AssertionError("assertNotEqual failed: %s" % `(a, b)`)
 
156
 
 
157
    def assert_re_match(self, pattern, s):
 
158
        """Assert that a string matches a particular pattern
 
159
 
 
160
        Inputs:
 
161
          pattern      string: regular expression
 
162
          s            string: to be matched
 
163
 
 
164
        Raises:
 
165
          AssertionError if not matched
 
166
          """
 
167
        if not re.match(pattern, s):
 
168
            raise AssertionError("string does not match regexp\n"
 
169
                                 "    string: %s\n"
 
170
                                 "    re: %s" % (`s`, `pattern`))
 
171
 
 
172
    def assert_re_search(self, pattern, s):
 
173
        """Assert that a string *contains* a particular pattern
 
174
 
 
175
        Inputs:
 
176
          pattern      string: regular expression
 
177
          s            string: to be searched
 
178
 
 
179
        Raises:
 
180
          AssertionError if not matched
 
181
          """
 
182
        if not re.search(pattern, s):
 
183
            raise AssertionError("string does not contain regexp\n"
 
184
                                 "    string: %s\n"
 
185
                                 "    re: %s" % (`s`, `pattern`))
 
186
 
 
187
 
 
188
    def assert_no_file(self, filename):
 
189
        import os.path
 
190
        assert not os.path.exists(filename), ("file exists but should not: %s" % filename)
 
191
 
 
192
 
 
193
    #############################################################
 
194
    # Methods for running programs
 
195
 
 
196
    def runcmd_background(self, cmd):
 
197
        import os
 
198
        self.test_log = self.test_log + "Run in background:\n" + `cmd` + "\n"
 
199
        pid = os.fork()
 
200
        if pid == 0:
 
201
            # child
 
202
            try:
 
203
                os.execvp("/bin/sh", ["/bin/sh", "-c", cmd])
 
204
            finally:
 
205
                os._exit(127)
 
206
        self.test_log = self.test_log + "pid: %d\n" % pid
 
207
        return pid
 
208
 
 
209
 
 
210
    def runcmd(self, cmd, expectedResult = 0):
 
211
        """Run a command, fail if the command returns an unexpected exit
 
212
        code.  Return the output produced."""
 
213
        rc, output, stderr = self.runcmd_unchecked(cmd)
 
214
        if rc != expectedResult:
 
215
            raise AssertionError("""command returned %d; expected %s: \"%s\"
 
216
stdout:
 
217
%s
 
218
stderr:
 
219
%s""" % (rc, expectedResult, cmd, output, stderr))
 
220
 
 
221
        return output, stderr
 
222
 
 
223
 
 
224
    def run_captured(self, cmd):
 
225
        """Run a command, capturing stdout and stderr.
 
226
 
 
227
        Based in part on popen2.py
 
228
 
 
229
        Returns (waitstatus, stdout, stderr)."""
 
230
        import os, types
 
231
        pid = os.fork()
 
232
        if pid == 0:
 
233
            # child
 
234
            try: 
 
235
                pid = os.getpid()
 
236
                openmode = os.O_WRONLY|os.O_CREAT|os.O_TRUNC
 
237
 
 
238
                outfd = os.open('%d.out' % pid, openmode, 0666)
 
239
                os.dup2(outfd, 1)
 
240
                os.close(outfd)
 
241
 
 
242
                errfd = os.open('%d.err' % pid, openmode, 0666)
 
243
                os.dup2(errfd, 2)
 
244
                os.close(errfd)
 
245
 
 
246
                if isinstance(cmd, types.StringType):
 
247
                    cmd = ['/bin/sh', '-c', cmd]
 
248
 
 
249
                os.execvp(cmd[0], cmd)
 
250
            finally:
 
251
                os._exit(127)
 
252
        else:
 
253
            # parent
 
254
            exited_pid, waitstatus = os.waitpid(pid, 0)
 
255
            stdout = open('%d.out' % pid).read()
 
256
            stderr = open('%d.err' % pid).read()
 
257
            return waitstatus, stdout, stderr
 
258
 
 
259
 
 
260
    def runcmd_unchecked(self, cmd, skip_on_noexec = 0):
 
261
        """Invoke a command; return (exitcode, stdout, stderr)"""
 
262
        import os
 
263
        waitstatus, stdout, stderr = self.run_captured(cmd)
 
264
        assert not os.WIFSIGNALED(waitstatus), \
 
265
               ("%s terminated with signal %d" % (`cmd`, os.WTERMSIG(waitstatus)))
 
266
        rc = os.WEXITSTATUS(waitstatus)
 
267
        self.test_log = self.test_log + ("""Run command: %s
 
268
Wait status: %#x (exit code %d, signal %d)
 
269
stdout:
 
270
%s
 
271
stderr:
 
272
%s""" % (cmd, waitstatus, os.WEXITSTATUS(waitstatus), os.WTERMSIG(waitstatus),
 
273
         stdout, stderr))
 
274
        if skip_on_noexec and rc == 127:
 
275
            # Either we could not execute the command or the command
 
276
            # returned exit code 127.  According to system(3) we can't
 
277
            # tell the difference.
 
278
            raise NotRunError, "could not execute %s" % `cmd`
 
279
        return rc, stdout, stderr
 
280
    
 
281
 
 
282
    def explain_failure(self, exc_info = None):
 
283
        print "test_log:"
 
284
        print self.test_log
 
285
 
 
286
 
 
287
    def log(self, msg):
 
288
        """Log a message to the test log.  This message is displayed if
 
289
        the test fails, or when the runtests function is invoked with
 
290
        the verbose option."""
 
291
        self.test_log = self.test_log + msg + "\n"
 
292
 
 
293
 
 
294
class NotRunError(Exception):
 
295
    """Raised if a test must be skipped because of missing resources"""
 
296
    def __init__(self, value = None):
 
297
        self.value = value
 
298
 
 
299
 
 
300
def _report_error(case, debugger):
 
301
    """Ask the test case to explain failure, and optionally run a debugger
 
302
 
 
303
    Input:
 
304
      case         TestCase instance
 
305
      debugger     if true, a debugger function to be applied to the traceback
 
306
"""
 
307
    import sys
 
308
    ex = sys.exc_info()
 
309
    print "-----------------------------------------------------------------"
 
310
    if ex:
 
311
        import traceback
 
312
        traceback.print_exc(file=sys.stdout)
 
313
    case.explain_failure()
 
314
    print "-----------------------------------------------------------------"
 
315
 
 
316
    if debugger:
 
317
        tb = ex[2]
 
318
        debugger(tb)
 
319
 
 
320
 
 
321
def runtest(testcase_class, ret, verbose=0, debugger=None, subtest=0):
 
322
    """Instantiate test class, run it, and catch and report exceptions.
 
323
 
 
324
    Inputs:
 
325
       testcase_class  a class derived from TestCase
 
326
       ret             return status, an integer
 
327
       verbose         an integer (used as boolean)
 
328
       debugger        debugger object to be applied to errors
 
329
       subtest         an integer (used as boolean)
 
330
    Returns:
 
331
       a new return status
 
332
    Raises:
 
333
      KeyboardInterrupt
 
334
 
 
335
    If subtest is true, then the ordinary information about the
 
336
    test progress is not printed.  
 
337
    """
 
338
    if not subtest:
 
339
        print "%-30s" % _test_name(testcase_class),
 
340
        def failure_print(message):
 
341
            print message
 
342
    else:
 
343
        def failure_print(message):
 
344
            print '[%s %s]' % (_test_name(testcase_class), message)
 
345
            
 
346
    # flush now so that long running tests are easier to follow
 
347
    sys.stdout.flush()
 
348
 
 
349
    obj = None
 
350
    try:
 
351
        try:
 
352
            # run test and sometimes show result
 
353
            obj = testcase_class()
 
354
            obj.setup()
 
355
            obj.runtest()
 
356
            if not subtest:
 
357
                print "OK"
 
358
        except KeyboardInterrupt:
 
359
            failure_print("INTERRUPT")
 
360
            if obj:
 
361
                _report_error(obj, debugger)
 
362
            raise
 
363
        except NotRunError, msg:
 
364
            failure_print("NOTRUN, %s" % msg.value)
 
365
        except:
 
366
            failure_print("FAIL")
 
367
            if obj:
 
368
                _report_error(obj, debugger)
 
369
            return 1
 
370
    finally:
 
371
        if obj:
 
372
            ret = obj.apply_cleanups(debugger) or ret
 
373
    # Display log file if we're verbose
 
374
    if ret == 0 and verbose:
 
375
        obj.explain_failure()
 
376
    return ret
 
377
 
 
378
 
 
379
def runtests(test_list, verbose = 0, debugger = None):
 
380
    """Run a series of tests.
 
381
 
 
382
    Inputs:
 
383
      test_list    sequence of TestCase classes
 
384
      verbose      print more information as testing proceeds
 
385
      debugger     debugger object to be applied to errors
 
386
 
 
387
    Returns:
 
388
      unix return code: 0 for success, 1 for failures, 2 for test failure
 
389
    """
 
390
    import traceback
 
391
    ret = 0
 
392
    for testcase_class in test_list:
 
393
        try:
 
394
            ret = runtest(testcase_class, ret, verbose=verbose,
 
395
                          debugger=debugger)
 
396
        except KeyboardInterrupt:
 
397
            ret = 2
 
398
            break
 
399
    return ret
 
400
 
 
401
 
 
402
def _test_name(test_class):
 
403
    """Return a human-readable name for a test class.
 
404
    """
 
405
    try:
 
406
        return test_class.__name__
 
407
    except:
 
408
        return `test_class`
 
409
 
 
410
 
 
411
def print_help():
 
412
    """Help for people running tests"""
 
413
    import sys
 
414
    print """%s: software test suite based on ComfyChair
 
415
 
 
416
usage:
 
417
    To run all tests, just run this program.  To run particular tests,
 
418
    list them on the command line.
 
419
 
 
420
options:
 
421
    --help              show usage message
 
422
    --list              list available tests
 
423
    --verbose, -v       show more information while running tests
 
424
    --post-mortem, -p   enter Python debugger on error
 
425
""" % sys.argv[0]
 
426
 
 
427
 
 
428
def print_list(test_list):
 
429
    """Show list of available tests"""
 
430
    for test_class in test_list:
 
431
        print "    %s" % _test_name(test_class)
 
432
 
 
433
 
 
434
def main(tests, extra_tests=[]):
 
435
    """Main entry point for test suites based on ComfyChair.
 
436
 
 
437
    inputs:
 
438
      tests       Sequence of TestCase subclasses to be run by default.
 
439
      extra_tests Sequence of TestCase subclasses that are available but
 
440
                  not run by default.
 
441
 
 
442
Test suites should contain this boilerplate:
 
443
 
 
444
    if __name__ == '__main__':
 
445
        comfychair.main(tests)
 
446
 
 
447
This function handles standard options such as --help and --list, and
 
448
by default runs all tests in the suggested order.
 
449
 
 
450
Calls sys.exit() on completion.
 
451
"""
 
452
    from sys import argv
 
453
    import getopt, sys
 
454
 
 
455
    opt_verbose = 0
 
456
    debugger = None
 
457
 
 
458
    opts, args = getopt.getopt(argv[1:], 'pv',
 
459
                               ['help', 'list', 'verbose', 'post-mortem'])
 
460
    for opt, opt_arg in opts:
 
461
        if opt == '--help':
 
462
            print_help()
 
463
            return
 
464
        elif opt == '--list':
 
465
            print_list(tests + extra_tests)
 
466
            return
 
467
        elif opt == '--verbose' or opt == '-v':
 
468
            opt_verbose = 1
 
469
        elif opt == '--post-mortem' or opt == '-p':
 
470
            import pdb
 
471
            debugger = pdb.post_mortem
 
472
 
 
473
    if args:
 
474
        all_tests = tests + extra_tests
 
475
        by_name = {}
 
476
        for t in all_tests:
 
477
            by_name[_test_name(t)] = t
 
478
        which_tests = []
 
479
        for name in args:
 
480
            which_tests.append(by_name[name])
 
481
    else:
 
482
        which_tests = tests
 
483
 
 
484
    sys.exit(runtests(which_tests, verbose=opt_verbose,
 
485
                      debugger=debugger))
 
486
 
 
487
 
 
488
if __name__ == '__main__':
 
489
    print __doc__