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