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