xapian-bindings/python/testsuite.py

   1 # Utility functions for running tests and reporting the results.
   2 #
   3 # Copyright (C) 2007 Lemur Consulting Ltd
   4 # Copyright (C) 2008,2011 Olly Betts
   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,
  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 St, Fifth Floor, Boston, MA  02110-1301
  19 # USA
  20
  21 import gc
  22 import os as _os
  23 import os.path as _path
  24 import sys as _sys
  25 import traceback as _traceback
  26 import xapian as _xapian
  27
  28 class TestFail(Exception):
  29     pass
  30
  31 class TestRunner(object):
  32     def __init__(self):
  33         """Initialise the TestRunner.
  34
  35         """
  36
  37         self._out = OutProxy(_sys.stdout)
  38
  39         # _verbose is an integer, higher meaning more verbose
  40         self._verbose = _os.environ.get('VERBOSE', '').lower()
  41         if self._verbose in ('', '0', 'no', 'off', 'false'):
  42             self._verbose = 0
  43         else:
  44             try:
  45                 self._verbose = int(self._verbose)
  46             except:
  47                 self._verbose = 1
  48
  49         # context is a description of what the test is currently checking
  50         self._context = None
  51
  52     def context(self, context):
  53         """Set the context.
  54
  55         This should be a string describing what a test is checking, and will be
  56         displayed if the test fails.
  57
  58         A test may change the context several times - each call will override
  59         subsequent calls.
  60
  61         Set the context to None to remove display of a specific context message.
  62         This is performed automatically at the start of each test.
  63
  64         """
  65         self._context = context
  66         if context is not None and self._verbose > 1:
  67             self._out.start_line()
  68             self._out.write("Context: %s\n" % context)
  69             self._out.flush()
  70
  71     def expect(self, got, expected, message="Expected equality"):
  72         """Function used to check for a particular expected value.
  73
  74         """
  75         if self._verbose > 2:
  76             self._out.start_line()
  77             self._out.write("Checking for %r: expecting %r ... " % (message, expected))
  78             self._out.flush()
  79         if got != expected:
  80             if self._verbose > 2:
  81                 self._out.write_colour(" #red#failed##")
  82                 self._out.write(": got %r\n" % got)
  83                 self._out.flush()
  84             raise TestFail("%s: got %r, expected %r" % (message, got, expected))
  85         if self._verbose > 2:
  86             self._out.write_colour(" #green#ok##\n")
  87             self._out.flush()
  88
  89     def expect_query(self, query, expected):
  90         """Check that the description of a query is as expected.
  91
  92         """
  93         expected = 'Query(' + expected + ')'
  94         desc = str(query)
  95         if self._verbose > 2:
  96             self._out.start_line()
  97             self._out.write("Checking str(query): expecting %r ... " % expected)
  98             self._out.flush()
  99         if desc != expected:
 100             if self._verbose > 2:
 101                 self._out.write_colour(" #red#failed##")
 102                 self._out.write(": got %r\n" % desc)
 103                 self._out.flush()
 104             raise TestFail("Unexpected str(query): got %r, expected %r" % (desc, expected))
 105         if self._verbose > 2:
 106             self._out.write_colour(" #green#ok##\n")
 107             self._out.flush()
 108
 109     def expect_exception(self, expectedclass, expectedmsg, callable, *args):
 110         """Check that an exception is raised.
 111
 112          - expectedclass is the class of the exception to check for.
 113          - expectedmsg is the message to check for, or None to skip checking
 114            the message.
 115          - callable is the thing to call.
 116          - args are the arguments to pass to it.
 117
 118         """
 119         if self._verbose > 2:
 120             self._out.start_line()
 121             self._out.write("Checking for exception: %s(%r) ... " % (str(expectedclass), expectedmsg))
 122             self._out.flush()
 123         try:
 124             callable(*args)
 125             if self._verbose > 2:
 126                 self._out.write_colour(" #red#failed##: no exception occurred\n")
 127                 self._out.flush()
 128             raise TestFail("Expected %s(%r) exception" % (str(expectedclass), expectedmsg))
 129         except expectedclass, e:
 130             if expectedmsg is not None and str(e) != expectedmsg:
 131                 if self._verbose > 2:
 132                     self._out.write_colour(" #red#failed##")
 133                     self._out.write(": exception string not as expected: got '%s'\n" % str(e))
 134                     self._out.flush()
 135                 raise TestFail("Exception string not as expected: got '%s', expected '%s'" % (str(e), expectedmsg))
 136             if e.__class__ != expectedclass:
 137                 if self._verbose > 2:
 138                     self._out.write_colour(" #red#failed##")
 139                     self._out.write(": didn't get right exception class: got '%s'\n" % str(e.__class__))
 140                     self._out.flush()
 141                 raise TestFail("Didn't get right exception class: got '%s', expected '%s'" % (str(e.__class__), str(expectedclass)))
 142         if self._verbose > 2:
 143             self._out.write_colour(" #green#ok##\n")
 144             self._out.flush()
 145
 146     def report_failure(self, name, msg, show_traceback=True):
 147         "Report a test failure, with some useful context."
 148
 149         tb = _traceback.extract_tb(_sys.exc_info()[2])
 150
 151         # Move up the traceback until we get to the line in the test
 152         # function which caused the failure.
 153         for line in xrange(1, len(tb) + 1):
 154             if tb[-line][2] == 'test_' + name:
 155                 break
 156
 157         # Display the context in the text function.
 158         filepath, linenum, functionname, text = tb[-line]
 159         filename = _os.path.basename(filepath)
 160
 161         self._out.ensure_space()
 162         self._out.write_colour("#red#FAILED##\n")
 163         if self._verbose > 0:
 164             if self._context is None:
 165                 context = ''
 166             else:
 167                 context = ", when %s" % self._context
 168             firstline = "%s:%d" % (filename, linenum)
 169             self._out.write("\n%s:%s%s\n" % (firstline, msg, context))
 170
 171             # Display sourcecode lines
 172             lines = open(filepath).readlines()
 173             startline = max(linenum - 3, 0)
 174             endline = min(linenum + 2, len(lines))
 175             for num in range(startline, endline):
 176                 if num + 1 == linenum:
 177                     self._out.write('->')
 178                 else:
 179                     self._out.write('  ')
 180                 self._out.write("%4d %s\n" % (num + 1, lines[num].rstrip()))
 181
 182             # Display the traceback
 183             if show_traceback:
 184                 self._out.write("Traceback (most recent call last):\n")
 185                 for line in _traceback.format_list(tb):
 186                     self._out.write(line.rstrip() + '\n')
 187                 self._out.write('\n')
 188
 189             # Display some information about the xapian version and platform
 190             self._out.write("Xapian version: %s\n" % _xapian.version_string())
 191             try:
 192                 import platform
 193                 platdesc = "%s %s (%s)" % platform.system_alias(platform.system(),
 194                                                                 platform.release(),
 195                                                                 platform.version())
 196                 self._out.write("Platform: %s\n" % platdesc)
 197             except:
 198                 pass
 199             self._out.write('\nWhen reporting this problem, please quote all the preceding lines from\n"%s" onwards.\n\n' % firstline)
 200
 201         self._out.flush()
 202
 203     def gc_object_count(self):
 204         # Python 2.7 doesn't seem to free all objects even for a full
 205         # collection, so collect repeatedly until no further objects get freed.
 206         old_count, count = len(gc.get_objects()), 0
 207         while True:
 208             gc.collect()
 209             count = len(gc.get_objects())
 210             if count == old_count:
 211                 return count
 212             old_count = count
 213
 214     def runtest(self, name, test_fn):
 215         """Run a single test.
 216
 217         """
 218         startline = "Running test: %s..." % name
 219         self._out.write(startline)
 220         self._out.flush()
 221         try:
 222             object_count = self.gc_object_count()
 223             test_fn()
 224             object_count = self.gc_object_count() - object_count
 225             if object_count != 0:
 226                 # Maybe some lazily initialised object got initialised for the
 227                 # first time, so rerun the test.
 228                 self._out.ensure_space()
 229                 msg = "#yellow#possible leak (%d), rerunning## " % object_count
 230                 self._out.write_colour(msg)
 231                 object_count = self.gc_object_count()
 232                 test_fn()
 233                 expect(self.gc_object_count(), object_count)
 234                 self._out.write_colour("#green#ok##\n")
 235
 236             if self._verbose > 0 or self._out.plain:
 237                 self._out.ensure_space()
 238                 self._out.write_colour("#green#ok##\n")
 239             else:
 240                 self._out.clear_line()
 241             self._out.flush()
 242             return True
 243         except TestFail, e:
 244             self.report_failure(name, str(e), show_traceback=False)
 245         except _xapian.Error, e:
 246             self.report_failure(name, "%s: %s" % (str(e.__class__), str(e)))
 247         except Exception, e:
 248             self.report_failure(name, "%s: %s" % (str(e.__class__), str(e)))
 249         return False
 250
 251     def runtests(self, namedict, runonly=None):
 252         """Run a set of tests.
 253
 254         Takes a dictionary of name-value pairs and runs all the values which are
 255         callables, for which the name begins with "test_".
 256
 257         Typical usage is to pass "locals()" as the parameter, to run all callables
 258         with names starting "test_" in local scope.
 259
 260         If runonly is supplied, and non-empty, only those tests which appear in
 261         runonly will be run.
 262
 263         """
 264         tests = []
 265         if isinstance(namedict, dict):
 266             for name in namedict:
 267                 if name.startswith('test_'):
 268                     fn = namedict[name]
 269                     name = name[5:]
 270                     if hasattr(fn, '__call__'):
 271                         tests.append((name, fn))
 272             tests.sort()
 273         else:
 274             tests = namedict
 275
 276         if runonly is not None and len(runonly) != 0:
 277             oldtests = tests
 278             tests = []
 279             for name, fn in oldtests:
 280                 if name in runonly:
 281                     tests.append((name, fn))
 282
 283         passed, failed = 0, 0
 284         for name, fn in tests:
 285             self.context(None)
 286             if self.runtest(name, fn):
 287                 passed += 1
 288             else:
 289                 failed += 1
 290         if failed:
 291             if self._verbose == 0:
 292                 self._out.write('Re-run with the environment variable VERBOSE=1 to see details.\n')
 293                 self._out.write('E.g. make check VERBOSE=1\n')
 294             self._out.write_colour("#green#%d## tests passed, #red#%d## tests failed\n" % (passed, failed))
 295             return False
 296         else:
 297             self._out.write_colour("#green#%d## tests passed, no failures\n" % passed)
 298             return True
 299
 300 class OutProxy(object):
 301     """Proxy output class to make formatting easier.
 302
 303     Allows colourisation, and keeps track of whether we're mid-line or not.
 304
 305     """
 306
 307     def __init__(self, out):
 308         self._out = out
 309         self._line_pos = 0 # Position on current line
 310         self._had_space = True # True iff we're preceded by whitespace (including newline)
 311         self.plain = not self._allow_control_sequences()
 312         self._colours = self.get_colour_strings()
 313
 314     def _allow_control_sequences(self):
 315         "Return True if output device allows control sequences."
 316         mode = _os.environ.get("XAPIAN_TESTSUITE_OUTPUT", '').lower()
 317         if mode in ('', 'auto'):
 318             if _sys.platform == 'win32':
 319                 return False
 320             elif not hasattr(self._out, "isatty"):
 321                 return False
 322             else:
 323                 return self._out.isatty()
 324         elif mode == 'plain':
 325             return False
 326         return True
 327
 328     def get_colour_strings(self):
 329         """Return a mapping of colour names to colour output sequences.
 330
 331         """
 332         colours = {
 333             'red': "\x1b[1m\x1b[31m",
 334             'green': "\x1b[1m\x1b[32m",
 335             'yellow': "\x1b[1m\x1b[33m",
 336             '': "\x1b[0m",
 337         }
 338         if self.plain:
 339             for key in colours:
 340                 colours[key] = ''
 341         return colours
 342
 343     def _colourise(self, msg):
 344         """Apply colours to a message.
 345
 346         #colourname# will change the text colour, ## will change the colour back.
 347
 348         """
 349         for colour, val in self._colours.iteritems():
 350             msg = msg.replace('#%s#' % colour, val)
 351         return msg
 352
 353     def clear_line(self):
 354         """Clear the current line of output, if possible.
 355
 356         Otherwise, just move to the start of the next line.
 357
 358         """
 359         if self._line_pos == 0:
 360             return
 361         if self.plain:
 362             self.write('\n')
 363         else:
 364             self.write("\r" + " " * self._line_pos + "\r")
 365
 366     def start_line(self):
 367         """Ensure that we're at the start of a line.
 368
 369         """
 370         if self._line_pos != 0:
 371             self.write('\n')
 372
 373     def ensure_space(self):
 374         """Ensure that we're preceded by whitespace.
 375
 376         """
 377         if not self._had_space:
 378             self.write(' ')
 379
 380     def write(self, msg):
 381         """Write the message to the output stream.
 382
 383         """
 384         if len(msg) == 0:
 385             return
 386
 387         # Adjust the line position counted
 388         nlpos = max(msg.rfind('\n'), msg.rfind('\r'))
 389         if nlpos >= 0:
 390             subline = msg[nlpos + 1:]
 391             self._line_pos = len(subline) # Note - doesn't cope with tabs.
 392         else:
 393             self._line_pos += len(msg) # Note - doesn't cope with tabs.
 394
 395         # Record whether we ended with whitespace
 396         self._had_space = msg[-1].isspace()
 397
 398         self._out.write(msg)
 399
 400     def write_colour(self, msg):
 401         """Write a message, first substituting markup for colours.
 402
 403         """
 404         self.write(self._colourise(msg))
 405
 406     def flush(self):
 407         self._out.flush()
 408
 409
 410 _runner = TestRunner()
 411 context = _runner.context
 412 expect = _runner.expect
 413 expect_query = _runner.expect_query
 414 expect_exception = _runner.expect_exception
 415 runtests = _runner.runtests
 416
 417 __all__ = ('TestFail', 'context', 'expect', 'expect_query', 'expect_exception', 'runtests')
 418
 419 def next(iterator):
 420     return iterator.next()
 421 __all__ = __all__ + ('next',)