debug/testlib.py

   1 import collections
   2 import os.path
   3 import random
   4 import re
   5 import shlex
   6 import subprocess
   7 import sys
   8 import tempfile
   9 import time
  10 import traceback
  11
  12 import pexpect
  13
  14 # Note that gdb comes with its own testsuite. I was unable to figure out how to
  15 # run that testsuite against the spike simulator.
  16
  17 def find_file(path):
  18     for directory in (os.getcwd(), os.path.dirname(__file__)):
  19         fullpath = os.path.join(directory, path)
  20         relpath = os.path.relpath(fullpath)
  21         if len(relpath) >= len(fullpath):
  22             relpath = fullpath
  23         if os.path.exists(relpath):
  24             return relpath
  25     return None
  26
  27 def compile(args, xlen=32): # pylint: disable=redefined-builtin
  28     cc = os.path.expandvars("$RISCV/bin/riscv64-unknown-elf-gcc")
  29     cmd = [cc, "-g"]
  30     if xlen == 32:
  31         cmd.append("-march=rv32imac")
  32         cmd.append("-mabi=ilp32")
  33     else:
  34         cmd.append("-march=rv64imac")
  35         cmd.append("-mabi=lp64")
  36     for arg in args:
  37         found = find_file(arg)
  38         if found:
  39             cmd.append(found)
  40         else:
  41             cmd.append(arg)
  42     header("Compile")
  43     print "+", " ".join(cmd)
  44     process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
  45                                stderr=subprocess.PIPE)
  46     stdout, stderr = process.communicate()
  47     if process.returncode:
  48         print stdout,
  49         print stderr,
  50         header("")
  51         raise Exception("Compile failed!")
  52
  53 def unused_port():
  54     # http://stackoverflow.com/questions/2838244/get-open-tcp-port-in-python/2838309#2838309
  55     import socket
  56     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  57     s.bind(("", 0))
  58     port = s.getsockname()[1]
  59     s.close()
  60     return port
  61
  62 class Spike(object):
  63     logname = "spike-%d.log" % os.getpid()
  64
  65     def __init__(self, target, halted=False, timeout=None, with_jtag_gdb=True):
  66         """Launch spike. Return tuple of its process and the port it's running
  67         on."""
  68         self.process = None
  69
  70         if target.harts:
  71             harts = target.harts
  72         else:
  73             harts = [target]
  74
  75         cmd = self.command(target, harts, halted, timeout, with_jtag_gdb)
  76         self.infinite_loop = target.compile(harts[0],
  77                 "programs/checksum.c", "programs/tiny-malloc.c",
  78                 "programs/infinite_loop.S", "-DDEFINE_MALLOC", "-DDEFINE_FREE")
  79         cmd.append(self.infinite_loop)
  80         logfile = open(self.logname, "w")
  81         logfile.write("+ %s\n" % " ".join(cmd))
  82         logfile.flush()
  83         self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
  84                 stdout=logfile, stderr=logfile)
  85
  86         if with_jtag_gdb:
  87             self.port = None
  88             for _ in range(30):
  89                 m = re.search(r"Listening for remote bitbang connection on "
  90                         r"port (\d+).", open(self.logname).read())
  91                 if m:
  92                     self.port = int(m.group(1))
  93                     os.environ['REMOTE_BITBANG_PORT'] = m.group(1)
  94                     break
  95                 time.sleep(0.11)
  96             if not self.port:
  97                 print_log(self.logname)
  98                 raise Exception("Didn't get spike message about bitbang "
  99                         "connection")
 100
 101     def command(self, target, harts, halted, timeout, with_jtag_gdb):
 102         # pylint: disable=no-self-use
 103         if target.sim_cmd:
 104             cmd = shlex.split(target.sim_cmd)
 105         else:
 106             spike = os.path.expandvars("$RISCV/bin/spike")
 107             cmd = [spike]
 108
 109         cmd += ["-p%d" % len(harts)]
 110
 111         assert len(set(t.xlen for t in harts)) == 1, \
 112                 "All spike harts must have the same XLEN"
 113
 114         if harts[0].xlen == 32:
 115             cmd += ["--isa", "RV32G"]
 116         else:
 117             cmd += ["--isa", "RV64G"]
 118
 119         assert len(set(t.ram for t in harts)) == 1, \
 120                 "All spike harts must have the same RAM layout"
 121         assert len(set(t.ram_size for t in harts)) == 1, \
 122                 "All spike harts must have the same RAM layout"
 123         cmd += ["-m0x%x:0x%x" % (harts[0].ram, harts[0].ram_size)]
 124
 125         if timeout:
 126             cmd = ["timeout", str(timeout)] + cmd
 127
 128         if halted:
 129             cmd.append('-H')
 130         if with_jtag_gdb:
 131             cmd += ['--rbb-port', '0']
 132             os.environ['REMOTE_BITBANG_HOST'] = 'localhost'
 133
 134         return cmd
 135
 136     def __del__(self):
 137         if self.process:
 138             try:
 139                 self.process.kill()
 140                 self.process.wait()
 141             except OSError:
 142                 pass
 143
 144     def wait(self, *args, **kwargs):
 145         return self.process.wait(*args, **kwargs)
 146
 147 class VcsSim(object):
 148     logname = "simv.log"
 149
 150     def __init__(self, sim_cmd=None, debug=False):
 151         if sim_cmd:
 152             cmd = shlex.split(sim_cmd)
 153         else:
 154             cmd = ["simv"]
 155         cmd += ["+jtag_vpi_enable"]
 156         if debug:
 157             cmd[0] = cmd[0] + "-debug"
 158             cmd += ["+vcdplusfile=output/gdbserver.vpd"]
 159         logfile = open(self.logname, "w")
 160         logfile.write("+ %s\n" % " ".join(cmd))
 161         logfile.flush()
 162         listenfile = open(self.logname, "r")
 163         listenfile.seek(0, 2)
 164         self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
 165                 stdout=logfile, stderr=logfile)
 166         done = False
 167         while not done:
 168             # Fail if VCS exits early
 169             exit_code = self.process.poll()
 170             if exit_code is not None:
 171                 raise RuntimeError('VCS simulator exited early with status %d'
 172                                    % exit_code)
 173
 174             line = listenfile.readline()
 175             if not line:
 176                 time.sleep(1)
 177             match = re.match(r"^Listening on port (\d+)$", line)
 178             if match:
 179                 done = True
 180                 self.port = int(match.group(1))
 181                 os.environ['JTAG_VPI_PORT'] = str(self.port)
 182
 183     def __del__(self):
 184         try:
 185             self.process.kill()
 186             self.process.wait()
 187         except OSError:
 188             pass
 189
 190 class Openocd(object):
 191     logfile = tempfile.NamedTemporaryFile(prefix='openocd', suffix='.log')
 192     logname = logfile.name
 193     print "OpenOCD Temporary Log File: %s" % logname
 194
 195     def __init__(self, server_cmd=None, config=None, debug=False, timeout=60):
 196         self.timeout = timeout
 197
 198         if server_cmd:
 199             cmd = shlex.split(server_cmd)
 200         else:
 201             openocd = os.path.expandvars("$RISCV/bin/openocd")
 202             cmd = [openocd]
 203             if debug:
 204                 cmd.append("-d")
 205
 206         # This command needs to come before any config scripts on the command
 207         # line, since they are executed in order.
 208         cmd += [
 209             # Tell OpenOCD to bind gdb to an unused, ephemeral port.
 210             "--command",
 211             "gdb_port 0",
 212             # Disable tcl and telnet servers, since they are unused and because
 213             # the port numbers will conflict if multiple OpenOCD processes are
 214             # running on the same server.
 215             "--command",
 216             "tcl_port disabled",
 217             "--command",
 218             "telnet_port disabled",
 219         ]
 220
 221         if config:
 222             f = find_file(config)
 223             if f is None:
 224                 print "Unable to read file " + config
 225                 exit(1)
 226
 227             cmd += ["-f", f]
 228         if debug:
 229             cmd.append("-d")
 230
 231         logfile = open(Openocd.logname, "w")
 232         logfile.write("+ %s\n" % " ".join(cmd))
 233         logfile.flush()
 234
 235         self.ports = []
 236         self.port = None
 237         self.process = self.start(cmd, logfile)
 238
 239     def start(self, cmd, logfile):
 240         process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
 241                 stdout=logfile, stderr=logfile)
 242
 243         try:
 244             # Wait for OpenOCD to have made it through riscv_examine(). When
 245             # using OpenOCD to communicate with a simulator this may take a
 246             # long time, and gdb will time out when trying to connect if we
 247             # attempt too early.
 248             start = time.time()
 249             messaged = False
 250             while True:
 251                 log = open(Openocd.logname).read()
 252                 m = re.search(r"Listening on port (\d+) for gdb connections",
 253                         log)
 254                 if m:
 255                     if not self.ports:
 256                         self.port = int(m.group(1))
 257                     self.ports.append(int(m.group(1)))
 258
 259                 if "telnet server disabled" in log:
 260                     break
 261
 262                 if not process.poll() is None:
 263                     raise Exception(
 264                             "OpenOCD exited before completing riscv_examine()")
 265                 if not messaged and time.time() - start > 1:
 266                     messaged = True
 267                     print "Waiting for OpenOCD to start..."
 268                 if (time.time() - start) > self.timeout:
 269                     raise Exception("ERROR: Timed out waiting for OpenOCD to "
 270                             "listen for gdb")
 271             return process
 272         except Exception:
 273             header("OpenOCD log")
 274             sys.stdout.write(log)
 275             raise
 276
 277     def __del__(self):
 278         try:
 279             self.process.kill()
 280             self.process.wait()
 281         except (OSError, AttributeError):
 282             pass
 283
 284 class OpenocdCli(object):
 285     def __init__(self, port=4444):
 286         self.child = pexpect.spawn(
 287                 "sh -c 'telnet localhost %d | tee openocd-cli.log'" % port)
 288         self.child.expect("> ")
 289
 290     def command(self, cmd):
 291         self.child.sendline(cmd)
 292         self.child.expect(cmd)
 293         self.child.expect("\n")
 294         self.child.expect("> ")
 295         return self.child.before.strip("\t\r\n \0")
 296
 297     def reg(self, reg=''):
 298         output = self.command("reg %s" % reg)
 299         matches = re.findall(r"(\w+) \(/\d+\): (0x[0-9A-F]+)", output)
 300         values = {r: int(v, 0) for r, v in matches}
 301         if reg:
 302             return values[reg]
 303         return values
 304
 305     def load_image(self, image):
 306         output = self.command("load_image %s" % image)
 307         if 'invalid ELF file, only 32bits files are supported' in output:
 308             raise TestNotApplicable(output)
 309
 310 class CannotAccess(Exception):
 311     def __init__(self, address):
 312         Exception.__init__(self)
 313         self.address = address
 314
 315 Thread = collections.namedtuple('Thread', ('id', 'target_id', 'name',
 316     'frame'))
 317
 318 class Gdb(object):
 319     logfile = tempfile.NamedTemporaryFile(prefix="gdb", suffix=".log")
 320     logname = logfile.name
 321     print "GDB Temporary Log File: %s" % logname
 322
 323     def __init__(self,
 324             cmd=os.path.expandvars("$RISCV/bin/riscv64-unknown-elf-gdb")):
 325         self.child = pexpect.spawn(cmd)
 326         self.child.logfile = open(self.logname, "w")
 327         self.child.logfile.write("+ %s\n" % cmd)
 328         self.wait()
 329         self.command("set confirm off")
 330         self.command("set width 0")
 331         self.command("set height 0")
 332         # Force consistency.
 333         self.command("set print entry-values no")
 334
 335     def select_hart(self, hart):
 336         output = self.command("thread %d" % (hart.index + 1))
 337         assert "Unknown" not in output
 338
 339     def wait(self):
 340         """Wait for prompt."""
 341         self.child.expect(r"\(gdb\)")
 342
 343     def command(self, command, timeout=6000):
 344         """timeout is in seconds"""
 345         self.child.sendline(command)
 346         self.child.expect("\n", timeout=timeout)
 347         self.child.expect(r"\(gdb\)", timeout=timeout)
 348         return self.child.before.strip()
 349
 350     def c(self, wait=True, timeout=-1, async=False):
 351         if async:
 352             async = "&"
 353         else:
 354             async = ""
 355         if wait:
 356             output = self.command("c%s" % async, timeout=timeout)
 357             assert "Continuing" in output
 358             return output
 359         else:
 360             self.child.sendline("c%s" % async)
 361             self.child.expect("Continuing")
 362
 363     def interrupt(self):
 364         self.child.send("\003")
 365         self.child.expect(r"\(gdb\)", timeout=6000)
 366         return self.child.before.strip()
 367
 368     def x(self, address, size='w'):
 369         output = self.command("x/%s %s" % (size, address))
 370         value = int(output.split(':')[1].strip(), 0)
 371         return value
 372
 373     def p_raw(self, obj):
 374         output = self.command("p %s" % obj)
 375         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 376         if m:
 377             raise CannotAccess(int(m.group(1), 0))
 378         return output.split('=')[-1].strip()
 379
 380     def parse_string(self, text):
 381         text = text.strip()
 382         if text.startswith("{") and text.endswith("}"):
 383             inner = text[1:-1]
 384             return [self.parse_string(t) for t in inner.split(", ")]
 385         elif text.startswith('"') and text.endswith('"'):
 386             return text[1:-1]
 387         else:
 388             return int(text, 0)
 389
 390     def p(self, obj, fmt="/x"):
 391         output = self.command("p%s %s" % (fmt, obj))
 392         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 393         if m:
 394             raise CannotAccess(int(m.group(1), 0))
 395         rhs = output.split('=')[-1]
 396         return self.parse_string(rhs)
 397
 398     def p_string(self, obj):
 399         output = self.command("p %s" % obj)
 400         value = shlex.split(output.split('=')[-1].strip())[1]
 401         return value
 402
 403     def stepi(self):
 404         output = self.command("stepi", timeout=60)
 405         return output
 406
 407     def load(self):
 408         output = self.command("load", timeout=6000)
 409         assert "failed" not in  output
 410         assert "Transfer rate" in output
 411
 412     def b(self, location):
 413         output = self.command("b %s" % location)
 414         assert "not defined" not in output
 415         assert "Breakpoint" in output
 416         return output
 417
 418     def hbreak(self, location):
 419         output = self.command("hbreak %s" % location)
 420         assert "not defined" not in output
 421         assert "Hardware assisted breakpoint" in output
 422         return output
 423
 424     def threads(self):
 425         output = self.command("info threads")
 426         threads = []
 427         for line in output.splitlines():
 428             m = re.match(
 429                     r"[\s\*]*(\d+)\s*Thread (\d+)\s*\(Name: ([^\)]+)\s*(.*)",
 430                     line)
 431             if m:
 432                 threads.append(Thread(*m.groups()))
 433         if not threads:
 434             threads.append(Thread('1', '1', 'Default', '???'))
 435         return threads
 436
 437     def thread(self, thread):
 438         return self.command("thread %s" % thread.id)
 439
 440 def run_all_tests(module, target, parsed):
 441     if not os.path.exists(parsed.logs):
 442         os.makedirs(parsed.logs)
 443
 444     overall_start = time.time()
 445
 446     global gdb_cmd  # pylint: disable=global-statement
 447     gdb_cmd = parsed.gdb
 448
 449     todo = []
 450     examine_added = False
 451     for hart in target.harts:
 452         if parsed.misaval:
 453             hart.misa = int(parsed.misaval, 16)
 454             print "Using $misa from command line: 0x%x" % hart.misa
 455         elif hart.misa:
 456             print "Using $misa from hart definition: 0x%x" % hart.misa
 457         elif not examine_added:
 458             todo.append(("ExamineTarget", ExamineTarget, None))
 459             examine_added = True
 460
 461     for name in dir(module):
 462         definition = getattr(module, name)
 463         if type(definition) == type and hasattr(definition, 'test') and \
 464                 (not parsed.test or any(test in name for test in parsed.test)):
 465             todo.append((name, definition, None))
 466
 467     results, count = run_tests(parsed, target, todo)
 468
 469     header("ran %d tests in %.0fs" % (count, time.time() - overall_start),
 470             dash=':')
 471
 472     return print_results(results)
 473
 474 good_results = set(('pass', 'not_applicable'))
 475 def run_tests(parsed, target, todo):
 476     results = {}
 477     count = 0
 478
 479     for name, definition, hart in todo:
 480         log_name = os.path.join(parsed.logs, "%s-%s-%s.log" %
 481                 (time.strftime("%Y%m%d-%H%M%S"), type(target).__name__, name))
 482         log_fd = open(log_name, 'w')
 483         print "Running %s > %s ..." % (name, log_name),
 484         instance = definition(target, hart)
 485         sys.stdout.flush()
 486         log_fd.write("Test: %s\n" % name)
 487         log_fd.write("Target: %s\n" % type(target).__name__)
 488         start = time.time()
 489         real_stdout = sys.stdout
 490         sys.stdout = log_fd
 491         try:
 492             result = instance.run()
 493             log_fd.write("Result: %s\n" % result)
 494         finally:
 495             sys.stdout = real_stdout
 496             log_fd.write("Time elapsed: %.2fs\n" % (time.time() - start))
 497         print "%s in %.2fs" % (result, time.time() - start)
 498         if result not in good_results and parsed.print_failures:
 499             sys.stdout.write(open(log_name).read())
 500         sys.stdout.flush()
 501         results.setdefault(result, []).append((name, log_name))
 502         count += 1
 503         if result not in good_results and parsed.fail_fast:
 504             break
 505
 506     return results, count
 507
 508 def print_results(results):
 509     result = 0
 510     for key, value in results.iteritems():
 511         print "%d tests returned %s" % (len(value), key)
 512         if key not in good_results:
 513             result = 1
 514             for name, log_name in value:
 515                 print "   %s > %s" % (name, log_name)
 516
 517     return result
 518
 519 def add_test_run_options(parser):
 520     parser.add_argument("--logs", default="logs",
 521             help="Store logs in the specified directory.")
 522     parser.add_argument("--fail-fast", "-f", action="store_true",
 523             help="Exit as soon as any test fails.")
 524     parser.add_argument("--print-failures", action="store_true",
 525             help="When a test fails, print the log file to stdout.")
 526     parser.add_argument("test", nargs='*',
 527             help="Run only tests that are named here.")
 528     parser.add_argument("--gdb",
 529             help="The command to use to start gdb.")
 530     parser.add_argument("--misaval",
 531             help="Don't run ExamineTarget, just assume the misa value which is "
 532             "specified.")
 533
 534 def header(title, dash='-', length=78):
 535     if title:
 536         dashes = dash * (length - 4 - len(title))
 537         before = dashes[:len(dashes)/2]
 538         after = dashes[len(dashes)/2:]
 539         print "%s[ %s ]%s" % (before, title, after)
 540     else:
 541         print dash * length
 542
 543 def print_log(path):
 544     header(path)
 545     for l in open(path, "r"):
 546         sys.stdout.write(l)
 547     print
 548
 549 class BaseTest(object):
 550     compiled = {}
 551
 552     def __init__(self, target, hart=None):
 553         self.target = target
 554         if hart:
 555             self.hart = hart
 556         else:
 557             self.hart = random.choice(target.harts)
 558             self.hart = target.harts[-1]    #<<<
 559         self.server = None
 560         self.target_process = None
 561         self.binary = None
 562         self.start = 0
 563         self.logs = []
 564
 565     def early_applicable(self):
 566         """Return a false value if the test has determined it cannot run
 567         without ever needing to talk to the target or server."""
 568         # pylint: disable=no-self-use
 569         return True
 570
 571     def setup(self):
 572         pass
 573
 574     def compile(self):
 575         compile_args = getattr(self, 'compile_args', None)
 576         if compile_args:
 577             if compile_args not in BaseTest.compiled:
 578                 # pylint: disable=star-args
 579                 BaseTest.compiled[compile_args] = \
 580                         self.target.compile(self.hart, *compile_args)
 581         self.binary = BaseTest.compiled.get(compile_args)
 582
 583     def classSetup(self):
 584         self.compile()
 585         self.target_process = self.target.create()
 586         if self.target_process:
 587             self.logs.append(self.target_process.logname)
 588         try:
 589             self.server = self.target.server()
 590             self.logs.append(self.server.logname)
 591         except Exception:
 592             for log in self.logs:
 593                 print_log(log)
 594             raise
 595
 596     def classTeardown(self):
 597         del self.server
 598         del self.target_process
 599
 600     def postMortem(self):
 601         pass
 602
 603     def run(self):
 604         """
 605         If compile_args is set, compile a program and set self.binary.
 606
 607         Call setup().
 608
 609         Then call test() and return the result, displaying relevant information
 610         if an exception is raised.
 611         """
 612
 613         sys.stdout.flush()
 614
 615         if not self.early_applicable():
 616             return "not_applicable"
 617
 618         self.start = time.time()
 619
 620         try:
 621             self.classSetup()
 622             self.setup()
 623             result = self.test()    # pylint: disable=no-member
 624         except TestNotApplicable:
 625             result = "not_applicable"
 626         except Exception as e: # pylint: disable=broad-except
 627             if isinstance(e, TestFailed):
 628                 result = "fail"
 629             else:
 630                 result = "exception"
 631             if isinstance(e, TestFailed):
 632                 header("Message")
 633                 print e.message
 634             header("Traceback")
 635             traceback.print_exc(file=sys.stdout)
 636             self.postMortem()
 637             return result
 638
 639         finally:
 640             for log in self.logs:
 641                 print_log(log)
 642             header("End of logs")
 643             self.classTeardown()
 644
 645         if not result:
 646             result = 'pass'
 647         return result
 648
 649 gdb_cmd = None
 650 class GdbTest(BaseTest):
 651     def __init__(self, target, hart=None):
 652         BaseTest.__init__(self, target, hart=hart)
 653         self.gdb = None
 654
 655     def classSetup(self):
 656         BaseTest.classSetup(self)
 657
 658         if gdb_cmd:
 659             self.gdb = Gdb(gdb_cmd)
 660         else:
 661             self.gdb = Gdb()
 662
 663         self.logs.append(self.gdb.logname)
 664
 665         if self.binary:
 666             self.gdb.command("file %s" % self.binary)
 667         if self.target:
 668             self.gdb.command("set arch riscv:rv%d" % self.hart.xlen)
 669             self.gdb.command("set remotetimeout %d" % self.target.timeout_sec)
 670         if self.server.port:
 671             self.gdb.command(
 672                     "target extended-remote localhost:%d" % self.server.port)
 673             self.gdb.select_hart(self.hart)
 674
 675         for cmd in self.target.gdb_setup:
 676             self.gdb.command(cmd)
 677
 678         # FIXME: OpenOCD doesn't handle PRIV now
 679         #self.gdb.p("$priv=3")
 680
 681     def postMortem(self):
 682         if not self.gdb:
 683             return
 684         self.gdb.interrupt()
 685         self.gdb.command("info registers all", timeout=10)
 686
 687     def classTeardown(self):
 688         del self.gdb
 689         BaseTest.classTeardown(self)
 690
 691 class GdbSingleHartTest(GdbTest):
 692     def classSetup(self):
 693         GdbTest.classSetup(self)
 694
 695         for hart in self.target.harts:
 696             # Park all harts that we're not using in a safe place.
 697             if hart != self.hart:
 698                 self.gdb.select_hart(hart)
 699                 self.gdb.p("$pc=loop_forever")
 700         self.gdb.select_hart(self.hart)
 701
 702 class ExamineTarget(GdbTest):
 703     def test(self):
 704         for hart in self.target.harts:
 705             self.gdb.select_hart(hart)
 706
 707             hart.misa = self.gdb.p("$misa")
 708
 709             txt = "RV"
 710             if (hart.misa >> 30) == 1:
 711                 txt += "32"
 712             elif (hart.misa >> 62) == 2:
 713                 txt += "64"
 714             elif (hart.misa >> 126) == 3:
 715                 txt += "128"
 716             else:
 717                 raise TestFailed("Couldn't determine XLEN from $misa (0x%x)" %
 718                         self.hart.misa)
 719
 720             for i in range(26):
 721                 if hart.misa & (1<<i):
 722                     txt += chr(i + ord('A'))
 723             print txt,
 724
 725 class TestFailed(Exception):
 726     def __init__(self, message):
 727         Exception.__init__(self)
 728         self.message = message
 729
 730 class TestNotApplicable(Exception):
 731     def __init__(self, message):
 732         Exception.__init__(self)
 733         self.message = message
 734
 735 def assertEqual(a, b):
 736     if a != b:
 737         raise TestFailed("%r != %r" % (a, b))
 738
 739 def assertNotEqual(a, b):
 740     if a == b:
 741         raise TestFailed("%r == %r" % (a, b))
 742
 743 def assertIn(a, b):
 744     if a not in b:
 745         raise TestFailed("%r not in %r" % (a, b))
 746
 747 def assertNotIn(a, b):
 748     if a in b:
 749         raise TestFailed("%r in %r" % (a, b))
 750
 751 def assertGreater(a, b):
 752     if not a > b:
 753         raise TestFailed("%r not greater than %r" % (a, b))
 754
 755 def assertLess(a, b):
 756     if not a < b:
 757         raise TestFailed("%r not less than %r" % (a, b))
 758
 759 def assertTrue(a):
 760     if not a:
 761         raise TestFailed("%r is not True" % a)
 762
 763 def assertRegexpMatches(text, regexp):
 764     if not re.search(regexp, text):
 765         raise TestFailed("can't find %r in %r" % (regexp, text))