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     for hart in target.harts:
 451         if parsed.misaval:
 452             hart.misa = int(parsed.misaval, 16)
 453             print "Using $misa from command line: 0x%x" % hart.misa
 454         elif hart.misa:
 455             print "Using $misa from hart definition: 0x%x" % hart.misa
 456         else:
 457             todo.append(("ExamineTarget", ExamineTarget, hart))
 458
 459     for name in dir(module):
 460         definition = getattr(module, name)
 461         if type(definition) == type and hasattr(definition, 'test') and \
 462                 (not parsed.test or any(test in name for test in parsed.test)):
 463             todo.append((name, definition, None))
 464
 465     results, count = run_tests(parsed, target, todo)
 466
 467     header("ran %d tests in %.0fs" % (count, time.time() - overall_start),
 468             dash=':')
 469
 470     return print_results(results)
 471
 472 good_results = set(('pass', 'not_applicable'))
 473 def run_tests(parsed, target, todo):
 474     results = {}
 475     count = 0
 476
 477     for name, definition, hart in todo:
 478         log_name = os.path.join(parsed.logs, "%s-%s-%s.log" %
 479                 (time.strftime("%Y%m%d-%H%M%S"), type(target).__name__, name))
 480         log_fd = open(log_name, 'w')
 481         print "Running %s > %s ..." % (name, log_name),
 482         instance = definition(target, hart)
 483         sys.stdout.flush()
 484         log_fd.write("Test: %s\n" % name)
 485         log_fd.write("Target: %s\n" % type(target).__name__)
 486         start = time.time()
 487         real_stdout = sys.stdout
 488         sys.stdout = log_fd
 489         try:
 490             result = instance.run()
 491             log_fd.write("Result: %s\n" % result)
 492         finally:
 493             sys.stdout = real_stdout
 494             log_fd.write("Time elapsed: %.2fs\n" % (time.time() - start))
 495         print "%s in %.2fs" % (result, time.time() - start)
 496         if result not in good_results and parsed.print_failures:
 497             sys.stdout.write(open(log_name).read())
 498         sys.stdout.flush()
 499         results.setdefault(result, []).append((name, log_name))
 500         count += 1
 501         if result not in good_results and parsed.fail_fast:
 502             break
 503
 504     return results, count
 505
 506 def print_results(results):
 507     result = 0
 508     for key, value in results.iteritems():
 509         print "%d tests returned %s" % (len(value), key)
 510         if key not in good_results:
 511             result = 1
 512             for name, log_name in value:
 513                 print "   %s > %s" % (name, log_name)
 514
 515     return result
 516
 517 def add_test_run_options(parser):
 518     parser.add_argument("--logs", default="logs",
 519             help="Store logs in the specified directory.")
 520     parser.add_argument("--fail-fast", "-f", action="store_true",
 521             help="Exit as soon as any test fails.")
 522     parser.add_argument("--print-failures", action="store_true",
 523             help="When a test fails, print the log file to stdout.")
 524     parser.add_argument("test", nargs='*',
 525             help="Run only tests that are named here.")
 526     parser.add_argument("--gdb",
 527             help="The command to use to start gdb.")
 528     parser.add_argument("--misaval",
 529             help="Don't run ExamineTarget, just assume the misa value which is "
 530             "specified.")
 531
 532 def header(title, dash='-', length=78):
 533     if title:
 534         dashes = dash * (length - 4 - len(title))
 535         before = dashes[:len(dashes)/2]
 536         after = dashes[len(dashes)/2:]
 537         print "%s[ %s ]%s" % (before, title, after)
 538     else:
 539         print dash * length
 540
 541 def print_log(path):
 542     header(path)
 543     lines = open(path, "r").readlines()
 544     for l in lines:
 545         sys.stdout.write(l)
 546     print
 547
 548 class BaseTest(object):
 549     compiled = {}
 550
 551     def __init__(self, target, hart=None):
 552         self.target = target
 553         if hart:
 554             self.hart = hart
 555         else:
 556             self.hart = random.choice(target.harts)
 557             self.hart = target.harts[-1]    #<<<
 558         self.server = None
 559         self.target_process = None
 560         self.binary = None
 561         self.start = 0
 562         self.logs = []
 563
 564     def early_applicable(self):
 565         """Return a false value if the test has determined it cannot run
 566         without ever needing to talk to the target or server."""
 567         # pylint: disable=no-self-use
 568         return True
 569
 570     def setup(self):
 571         pass
 572
 573     def compile(self):
 574         compile_args = getattr(self, 'compile_args', None)
 575         if compile_args:
 576             if compile_args not in BaseTest.compiled:
 577                 # pylint: disable=star-args
 578                 BaseTest.compiled[compile_args] = \
 579                         self.target.compile(self.hart, *compile_args)
 580         self.binary = BaseTest.compiled.get(compile_args)
 581
 582     def classSetup(self):
 583         self.compile()
 584         self.target_process = self.target.create()
 585         if self.target_process:
 586             self.logs.append(self.target_process.logname)
 587         try:
 588             self.server = self.target.server()
 589             self.logs.append(self.server.logname)
 590         except Exception:
 591             for log in self.logs:
 592                 print_log(log)
 593             raise
 594
 595     def classTeardown(self):
 596         del self.server
 597         del self.target_process
 598
 599     def postMortem(self):
 600         pass
 601
 602     def run(self):
 603         """
 604         If compile_args is set, compile a program and set self.binary.
 605
 606         Call setup().
 607
 608         Then call test() and return the result, displaying relevant information
 609         if an exception is raised.
 610         """
 611
 612         sys.stdout.flush()
 613
 614         if not self.early_applicable():
 615             return "not_applicable"
 616
 617         self.start = time.time()
 618
 619         try:
 620             self.classSetup()
 621             self.setup()
 622             result = self.test()    # pylint: disable=no-member
 623         except TestNotApplicable:
 624             result = "not_applicable"
 625         except Exception as e: # pylint: disable=broad-except
 626             if isinstance(e, TestFailed):
 627                 result = "fail"
 628             else:
 629                 result = "exception"
 630             if isinstance(e, TestFailed):
 631                 header("Message")
 632                 print e.message
 633             header("Traceback")
 634             traceback.print_exc(file=sys.stdout)
 635             self.postMortem()
 636             return result
 637
 638         finally:
 639             for log in self.logs:
 640                 print_log(log)
 641             header("End of logs")
 642             self.classTeardown()
 643
 644         if not result:
 645             result = 'pass'
 646         return result
 647
 648 gdb_cmd = None
 649 class GdbTest(BaseTest):
 650     def __init__(self, target, hart=None):
 651         BaseTest.__init__(self, target, hart=hart)
 652         self.gdb = None
 653
 654     def classSetup(self):
 655         BaseTest.classSetup(self)
 656
 657         if gdb_cmd:
 658             self.gdb = Gdb(gdb_cmd)
 659         else:
 660             self.gdb = Gdb()
 661
 662         self.logs.append(self.gdb.logname)
 663
 664         if self.binary:
 665             self.gdb.command("file %s" % self.binary)
 666         if self.target:
 667             self.gdb.command("set arch riscv:rv%d" % self.hart.xlen)
 668             self.gdb.command("set remotetimeout %d" % self.target.timeout_sec)
 669         if self.server.port:
 670             self.gdb.command(
 671                     "target extended-remote localhost:%d" % self.server.port)
 672             self.gdb.select_hart(self.hart)
 673
 674         for cmd in self.target.gdb_setup:
 675             self.gdb.command(cmd)
 676
 677         # FIXME: OpenOCD doesn't handle PRIV now
 678         #self.gdb.p("$priv=3")
 679
 680     def postMortem(self):
 681         if not self.gdb:
 682             return
 683         self.gdb.interrupt()
 684         self.gdb.command("info registers all", timeout=10)
 685
 686     def classTeardown(self):
 687         del self.gdb
 688         BaseTest.classTeardown(self)
 689
 690 class GdbSingleHartTest(GdbTest):
 691     def classSetup(self):
 692         GdbTest.classSetup(self)
 693
 694         for hart in self.target.harts:
 695             # Park all harts that we're not using in a safe place.
 696             if hart != self.hart:
 697                 self.gdb.select_hart(hart)
 698                 self.gdb.p("$pc=loop_forever")
 699         self.gdb.select_hart(self.hart)
 700
 701 class ExamineTarget(GdbTest):
 702     def test(self):
 703         self.hart.misa = self.gdb.p("$misa")
 704
 705         txt = "RV"
 706         if (self.hart.misa >> 30) == 1:
 707             txt += "32"
 708         elif (self.hart.misa >> 62) == 2:
 709             txt += "64"
 710         elif (self.hart.misa >> 126) == 3:
 711             txt += "128"
 712         else:
 713             raise TestFailed("Couldn't determine XLEN from $misa (0x%x)" %
 714                     self.hart.misa)
 715
 716         for i in range(26):
 717             if self.hart.misa & (1<<i):
 718                 txt += chr(i + ord('A'))
 719         print txt,
 720
 721 class TestFailed(Exception):
 722     def __init__(self, message):
 723         Exception.__init__(self)
 724         self.message = message
 725
 726 class TestNotApplicable(Exception):
 727     def __init__(self, message):
 728         Exception.__init__(self)
 729         self.message = message
 730
 731 def assertEqual(a, b):
 732     if a != b:
 733         raise TestFailed("%r != %r" % (a, b))
 734
 735 def assertNotEqual(a, b):
 736     if a == b:
 737         raise TestFailed("%r == %r" % (a, b))
 738
 739 def assertIn(a, b):
 740     if a not in b:
 741         raise TestFailed("%r not in %r" % (a, b))
 742
 743 def assertNotIn(a, b):
 744     if a in b:
 745         raise TestFailed("%r in %r" % (a, b))
 746
 747 def assertGreater(a, b):
 748     if not a > b:
 749         raise TestFailed("%r not greater than %r" % (a, b))
 750
 751 def assertLess(a, b):
 752     if not a < b:
 753         raise TestFailed("%r not less than %r" % (a, b))
 754
 755 def assertTrue(a):
 756     if not a:
 757         raise TestFailed("%r is not True" % a)
 758
 759 def assertRegexpMatches(text, regexp):
 760     if not re.search(regexp, text):
 761         raise TestFailed("can't find %r in %r" % (regexp, text))