debug/testlib.py

   1 import collections
   2 import os
   3 import os.path
   4 import random
   5 import re
   6 import shlex
   7 import subprocess
   8 import sys
   9 import tempfile
  10 import time
  11 import traceback
  12
  13 import pexpect
  14
  15 # Note that gdb comes with its own testsuite. I was unable to figure out how to
  16 # run that testsuite against the spike simulator.
  17
  18 def find_file(path):
  19     for directory in (os.getcwd(), os.path.dirname(__file__)):
  20         fullpath = os.path.join(directory, path)
  21         relpath = os.path.relpath(fullpath)
  22         if len(relpath) >= len(fullpath):
  23             relpath = fullpath
  24         if os.path.exists(relpath):
  25             return relpath
  26     return None
  27
  28 def compile(args, xlen=32): # pylint: disable=redefined-builtin
  29     cc = os.path.expandvars("$RISCV/bin/riscv64-unknown-elf-gcc")
  30     cmd = [cc, "-g"]
  31     if xlen == 32:
  32         cmd.append("-march=rv32imac")
  33         cmd.append("-mabi=ilp32")
  34     else:
  35         cmd.append("-march=rv64imac")
  36         cmd.append("-mabi=lp64")
  37     for arg in args:
  38         found = find_file(arg)
  39         if found:
  40             cmd.append(found)
  41         else:
  42             cmd.append(arg)
  43     header("Compile")
  44     print "+", " ".join(cmd)
  45     process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
  46                                stderr=subprocess.PIPE)
  47     stdout, stderr = process.communicate()
  48     if process.returncode:
  49         print stdout,
  50         print stderr,
  51         header("")
  52         raise Exception("Compile failed!")
  53
  54 class Spike(object):
  55     def __init__(self, target, halted=False, timeout=None, with_jtag_gdb=True):
  56         """Launch spike. Return tuple of its process and the port it's running
  57         on."""
  58         self.process = None
  59
  60         if target.harts:
  61             harts = target.harts
  62         else:
  63             harts = [target]
  64
  65         cmd = self.command(target, harts, halted, timeout, with_jtag_gdb)
  66         self.infinite_loop = target.compile(harts[0],
  67                 "programs/checksum.c", "programs/tiny-malloc.c",
  68                 "programs/infinite_loop.S", "-DDEFINE_MALLOC", "-DDEFINE_FREE")
  69         cmd.append(self.infinite_loop)
  70         self.logfile = tempfile.NamedTemporaryFile(prefix="spike-",
  71                 suffix=".log")
  72         self.logname = self.logfile.name
  73         self.logfile.write("+ %s\n" % " ".join(cmd))
  74         self.logfile.flush()
  75         self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
  76                 stdout=self.logfile, stderr=self.logfile)
  77
  78         if with_jtag_gdb:
  79             self.port = None
  80             for _ in range(30):
  81                 m = re.search(r"Listening for remote bitbang connection on "
  82                         r"port (\d+).", open(self.logname).read())
  83                 if m:
  84                     self.port = int(m.group(1))
  85                     os.environ['REMOTE_BITBANG_PORT'] = m.group(1)
  86                     break
  87                 time.sleep(0.11)
  88             if not self.port:
  89                 print_log(self.logname)
  90                 raise Exception("Didn't get spike message about bitbang "
  91                         "connection")
  92
  93     def command(self, target, harts, halted, timeout, with_jtag_gdb):
  94         # pylint: disable=no-self-use
  95         if target.sim_cmd:
  96             cmd = shlex.split(target.sim_cmd)
  97         else:
  98             spike = os.path.expandvars("$RISCV/bin/spike")
  99             cmd = [spike]
 100
 101         cmd += ["-p%d" % len(harts)]
 102
 103         assert len(set(t.xlen for t in harts)) == 1, \
 104                 "All spike harts must have the same XLEN"
 105
 106         if harts[0].xlen == 32:
 107             cmd += ["--isa", "RV32G"]
 108         else:
 109             cmd += ["--isa", "RV64G"]
 110
 111         assert len(set(t.ram for t in harts)) == 1, \
 112                 "All spike harts must have the same RAM layout"
 113         assert len(set(t.ram_size for t in harts)) == 1, \
 114                 "All spike harts must have the same RAM layout"
 115         cmd += ["-m0x%x:0x%x" % (harts[0].ram, harts[0].ram_size)]
 116
 117         if timeout:
 118             cmd = ["timeout", str(timeout)] + cmd
 119
 120         if halted:
 121             cmd.append('-H')
 122         if with_jtag_gdb:
 123             cmd += ['--rbb-port', '0']
 124             os.environ['REMOTE_BITBANG_HOST'] = 'localhost'
 125
 126         return cmd
 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         self.timeout = timeout
 189
 190         if server_cmd:
 191             cmd = shlex.split(server_cmd)
 192         else:
 193             openocd = os.path.expandvars("$RISCV/bin/openocd")
 194             cmd = [openocd]
 195             if debug:
 196                 cmd.append("-d")
 197
 198         # This command needs to come before any config scripts on the command
 199         # line, since they are executed in order.
 200         cmd += [
 201             # Tell OpenOCD to bind gdb to an unused, ephemeral port.
 202             "--command",
 203             "gdb_port 0",
 204             # Disable tcl and telnet servers, since they are unused and because
 205             # the port numbers will conflict if multiple OpenOCD processes are
 206             # running on the same server.
 207             "--command",
 208             "tcl_port disabled",
 209             "--command",
 210             "telnet_port disabled",
 211         ]
 212
 213         if config:
 214             f = find_file(config)
 215             if f is None:
 216                 print "Unable to read file " + config
 217                 exit(1)
 218
 219             cmd += ["-f", f]
 220         if debug:
 221             cmd.append("-d")
 222
 223         logfile = open(Openocd.logname, "w")
 224         logfile.write("+ %s\n" % " ".join(cmd))
 225         logfile.flush()
 226
 227         self.gdb_ports = []
 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             fd = open(Openocd.logname, "r")
 242             while True:
 243                 line = fd.readline()
 244                 if not line:
 245                     if not process.poll() is None:
 246                         raise Exception("OpenOCD exited early.")
 247                     time.sleep(0.1)
 248                     continue
 249
 250                 m = re.search(r"Listening on port (\d+) for gdb connections",
 251                         line)
 252                 if m:
 253                     self.gdb_ports.append(int(m.group(1)))
 254
 255                 if "telnet server disabled" in line:
 256                     return process
 257
 258                 if not messaged and time.time() - start > 1:
 259                     messaged = True
 260                     print "Waiting for OpenOCD to start..."
 261                 if (time.time() - start) > self.timeout:
 262                     raise Exception("Timed out waiting for OpenOCD to "
 263                             "listen for gdb")
 264
 265         except Exception:
 266             print_log(Openocd.logname)
 267             raise
 268
 269     def __del__(self):
 270         try:
 271             self.process.kill()
 272             self.process.wait()
 273         except (OSError, AttributeError):
 274             pass
 275
 276 class OpenocdCli(object):
 277     def __init__(self, port=4444):
 278         self.child = pexpect.spawn(
 279                 "sh -c 'telnet localhost %d | tee openocd-cli.log'" % port)
 280         self.child.expect("> ")
 281
 282     def command(self, cmd):
 283         self.child.sendline(cmd)
 284         self.child.expect(cmd)
 285         self.child.expect("\n")
 286         self.child.expect("> ")
 287         return self.child.before.strip("\t\r\n \0")
 288
 289     def reg(self, reg=''):
 290         output = self.command("reg %s" % reg)
 291         matches = re.findall(r"(\w+) \(/\d+\): (0x[0-9A-F]+)", output)
 292         values = {r: int(v, 0) for r, v in matches}
 293         if reg:
 294             return values[reg]
 295         return values
 296
 297     def load_image(self, image):
 298         output = self.command("load_image %s" % image)
 299         if 'invalid ELF file, only 32bits files are supported' in output:
 300             raise TestNotApplicable(output)
 301
 302 class CannotAccess(Exception):
 303     def __init__(self, address):
 304         Exception.__init__(self)
 305         self.address = address
 306
 307 Thread = collections.namedtuple('Thread', ('id', 'description', 'target_id',
 308     'name', 'frame'))
 309
 310 class Gdb(object):
 311     """A single gdb class which can interact with one or more gdb instances."""
 312
 313     # pylint: disable=too-many-public-methods
 314     # pylint: disable=too-many-instance-attributes
 315
 316     def __init__(self, ports,
 317             cmd=os.path.expandvars("$RISCV/bin/riscv64-unknown-elf-gdb"),
 318             timeout=60, binary=None):
 319         assert ports
 320
 321         self.ports = ports
 322         self.cmd = cmd
 323         self.timeout = timeout
 324         self.binary = binary
 325
 326         self.stack = []
 327         self.harts = {}
 328
 329         self.logfiles = []
 330         self.children = []
 331         for port in ports:
 332             logfile = tempfile.NamedTemporaryFile(prefix="gdb@%d-" % port,
 333                     suffix=".log")
 334             self.logfiles.append(logfile)
 335             child = pexpect.spawn(cmd)
 336             child.logfile = logfile
 337             child.logfile.write("+ %s\n" % cmd)
 338             self.children.append(child)
 339         self.active_child = self.children[0]
 340
 341     def connect(self):
 342         for port, child in zip(self.ports, self.children):
 343             self.select_child(child)
 344             self.wait()
 345             self.command("set confirm off")
 346             self.command("set width 0")
 347             self.command("set height 0")
 348             # Force consistency.
 349             self.command("set print entry-values no")
 350             self.command("set remotetimeout %d" % self.timeout)
 351             self.command("target extended-remote localhost:%d" % port)
 352             if self.binary:
 353                 self.command("file %s" % self.binary)
 354             threads = self.threads()
 355             for t in threads:
 356                 hartid = None
 357                 if t.name:
 358                     m = re.search(r"Hart (\d+)", t.name)
 359                     if m:
 360                         hartid = int(m.group(1))
 361                 if hartid is None:
 362                     if self.harts:
 363                         hartid = max(self.harts) + 1
 364                     else:
 365                         hartid = 0
 366                 self.harts[hartid] = (child, t)
 367
 368     def __del__(self):
 369         for child in self.children:
 370             del child
 371
 372     def lognames(self):
 373         return [logfile.name for logfile in self.logfiles]
 374
 375     def select_child(self, child):
 376         self.active_child = child
 377
 378     def select_hart(self, hart):
 379         child, thread = self.harts[hart.id]
 380         self.select_child(child)
 381         output = self.command("thread %s" % thread.id)
 382         assert "Unknown" not in output
 383
 384     def push_state(self):
 385         self.stack.append({
 386             'active_child': self.active_child
 387             })
 388
 389     def pop_state(self):
 390         state = self.stack.pop()
 391         self.active_child = state['active_child']
 392
 393     def wait(self):
 394         """Wait for prompt."""
 395         self.active_child.expect(r"\(gdb\)")
 396
 397     def command(self, command, timeout=6000):
 398         """timeout is in seconds"""
 399         self.active_child.sendline(command)
 400         self.active_child.expect("\n", timeout=timeout)
 401         self.active_child.expect(r"\(gdb\)", timeout=timeout)
 402         return self.active_child.before.strip()
 403
 404     def global_command(self, command):
 405         """Execute this command on every gdb that we control."""
 406         with PrivateState(self):
 407             for child in self.children:
 408                 self.select_child(child)
 409                 self.command(command)
 410
 411     def c(self, wait=True, timeout=-1, async=False):
 412         """
 413         Dumb c command.
 414         In RTOS mode, gdb will resume all harts.
 415         In multi-gdb mode, this command will just go to the current gdb, so
 416         will only resume one hart.
 417         """
 418         if async:
 419             async = "&"
 420         else:
 421             async = ""
 422         if wait:
 423             output = self.command("c%s" % async, timeout=timeout)
 424             assert "Continuing" in output
 425             return output
 426         else:
 427             self.active_child.sendline("c%s" % async)
 428             self.active_child.expect("Continuing")
 429
 430     def c_all(self):
 431         """
 432         Resume every hart.
 433
 434         This function works fine when using multiple gdb sessions, but the
 435         caller must be careful when using it nonetheless. gdb's behavior is to
 436         not set breakpoints until just before the hart is resumed, and then
 437         clears them as soon as the hart halts. That means that you can't set
 438         one software breakpoint, and expect multiple harts to hit it. It's
 439         possible that the first hart completes set/run/halt/clear before the
 440         second hart even gets to resume, so it will never hit the breakpoint.
 441         """
 442         with PrivateState(self):
 443             for child in self.children:
 444                 child.sendline("c")
 445                 child.expect("Continuing")
 446
 447             # Now wait for them all to halt
 448             for child in self.children:
 449                 child.expect(r"\(gdb\)")
 450
 451     def interrupt(self):
 452         self.active_child.send("\003")
 453         self.active_child.expect(r"\(gdb\)", timeout=6000)
 454         return self.active_child.before.strip()
 455
 456     def x(self, address, size='w'):
 457         output = self.command("x/%s %s" % (size, address))
 458         value = int(output.split(':')[1].strip(), 0)
 459         return value
 460
 461     def p_raw(self, obj):
 462         output = self.command("p %s" % obj)
 463         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 464         if m:
 465             raise CannotAccess(int(m.group(1), 0))
 466         return output.split('=')[-1].strip()
 467
 468     def parse_string(self, text):
 469         text = text.strip()
 470         if text.startswith("{") and text.endswith("}"):
 471             inner = text[1:-1]
 472             return [self.parse_string(t) for t in inner.split(", ")]
 473         elif text.startswith('"') and text.endswith('"'):
 474             return text[1:-1]
 475         else:
 476             return int(text, 0)
 477
 478     def p(self, obj, fmt="/x"):
 479         output = self.command("p%s %s" % (fmt, obj))
 480         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 481         if m:
 482             raise CannotAccess(int(m.group(1), 0))
 483         rhs = output.split('=')[-1]
 484         return self.parse_string(rhs)
 485
 486     def p_string(self, obj):
 487         output = self.command("p %s" % obj)
 488         value = shlex.split(output.split('=')[-1].strip())[1]
 489         return value
 490
 491     def stepi(self):
 492         output = self.command("stepi", timeout=60)
 493         return output
 494
 495     def load(self):
 496         output = self.command("load", timeout=6000)
 497         assert "failed" not in  output
 498         assert "Transfer rate" in output
 499
 500     def b(self, location):
 501         output = self.command("b %s" % location)
 502         assert "not defined" not in output
 503         assert "Breakpoint" in output
 504         return output
 505
 506     def hbreak(self, location):
 507         output = self.command("hbreak %s" % location)
 508         assert "not defined" not in output
 509         assert "Hardware assisted breakpoint" in output
 510         return output
 511
 512     def threads(self):
 513         output = self.command("info threads")
 514         threads = []
 515         for line in output.splitlines():
 516             m = re.match(
 517                     r"[\s\*]*(\d+)\s*"
 518                     r"(Remote target|Thread (\d+)\s*\(Name: ([^\)]+))"
 519                     r"\s*(.*)", line)
 520             if m:
 521                 threads.append(Thread(*m.groups()))
 522         assert threads
 523         #>>>if not threads:
 524         #>>>    threads.append(Thread('1', '1', 'Default', '???'))
 525         return threads
 526
 527     def thread(self, thread):
 528         return self.command("thread %s" % thread.id)
 529
 530     def where(self):
 531         return self.command("where 1")
 532
 533 class PrivateState(object):
 534     def __init__(self, gdb):
 535         self.gdb = gdb
 536
 537     def __enter__(self):
 538         self.gdb.push_state()
 539
 540     def __exit__(self, _type, _value, _traceback):
 541         self.gdb.pop_state()
 542
 543 def run_all_tests(module, target, parsed):
 544     if not os.path.exists(parsed.logs):
 545         os.makedirs(parsed.logs)
 546
 547     overall_start = time.time()
 548
 549     global gdb_cmd  # pylint: disable=global-statement
 550     gdb_cmd = parsed.gdb
 551
 552     todo = []
 553     examine_added = False
 554     for hart in target.harts:
 555         if parsed.misaval:
 556             hart.misa = int(parsed.misaval, 16)
 557             print "Using $misa from command line: 0x%x" % hart.misa
 558         elif hart.misa:
 559             print "Using $misa from hart definition: 0x%x" % hart.misa
 560         elif not examine_added:
 561             todo.append(("ExamineTarget", ExamineTarget, None))
 562             examine_added = True
 563
 564     for name in dir(module):
 565         definition = getattr(module, name)
 566         if isinstance(definition, type) and hasattr(definition, 'test') and \
 567                 (not parsed.test or any(test in name for test in parsed.test)):
 568             todo.append((name, definition, None))
 569
 570     results, count = run_tests(parsed, target, todo)
 571
 572     header("ran %d tests in %.0fs" % (count, time.time() - overall_start),
 573             dash=':')
 574
 575     return print_results(results)
 576
 577 good_results = set(('pass', 'not_applicable'))
 578 def run_tests(parsed, target, todo):
 579     results = {}
 580     count = 0
 581
 582     for name, definition, hart in todo:
 583         log_name = os.path.join(parsed.logs, "%s-%s-%s.log" %
 584                 (time.strftime("%Y%m%d-%H%M%S"), type(target).__name__, name))
 585         log_fd = open(log_name, 'w')
 586         print "[%s] Starting > %s" % (name, log_name)
 587         instance = definition(target, hart)
 588         sys.stdout.flush()
 589         log_fd.write("Test: %s\n" % name)
 590         log_fd.write("Target: %s\n" % type(target).__name__)
 591         start = time.time()
 592         real_stdout = sys.stdout
 593         sys.stdout = log_fd
 594         try:
 595             result = instance.run(real_stdout)
 596             log_fd.write("Result: %s\n" % result)
 597         finally:
 598             sys.stdout = real_stdout
 599             log_fd.write("Time elapsed: %.2fs\n" % (time.time() - start))
 600         print "[%s] %s in %.2fs" % (name, result, time.time() - start)
 601         if result not in good_results and parsed.print_failures:
 602             sys.stdout.write(open(log_name).read())
 603         sys.stdout.flush()
 604         results.setdefault(result, []).append((name, log_name))
 605         count += 1
 606         if result not in good_results and parsed.fail_fast:
 607             break
 608
 609     return results, count
 610
 611 def print_results(results):
 612     result = 0
 613     for key, value in results.iteritems():
 614         print "%d tests returned %s" % (len(value), key)
 615         if key not in good_results:
 616             result = 1
 617             for name, log_name in value:
 618                 print "   %s > %s" % (name, log_name)
 619
 620     return result
 621
 622 def add_test_run_options(parser):
 623     parser.add_argument("--logs", default="logs",
 624             help="Store logs in the specified directory.")
 625     parser.add_argument("--fail-fast", "-f", action="store_true",
 626             help="Exit as soon as any test fails.")
 627     parser.add_argument("--print-failures", action="store_true",
 628             help="When a test fails, print the log file to stdout.")
 629     parser.add_argument("test", nargs='*',
 630             help="Run only tests that are named here.")
 631     parser.add_argument("--gdb",
 632             help="The command to use to start gdb.")
 633     parser.add_argument("--misaval",
 634             help="Don't run ExamineTarget, just assume the misa value which is "
 635             "specified.")
 636
 637 def header(title, dash='-', length=78):
 638     if title:
 639         dashes = dash * (length - 4 - len(title))
 640         before = dashes[:len(dashes)/2]
 641         after = dashes[len(dashes)/2:]
 642         print "%s[ %s ]%s" % (before, title, after)
 643     else:
 644         print dash * length
 645
 646 def print_log(path):
 647     header(path)
 648     for l in open(path, "r"):
 649         sys.stdout.write(l)
 650     print
 651
 652 class BaseTest(object):
 653     compiled = {}
 654
 655     def __init__(self, target, hart=None):
 656         self.target = target
 657         if hart:
 658             self.hart = hart
 659         else:
 660             self.hart = random.choice(target.harts)
 661             self.hart = target.harts[-1]    #<<<
 662         self.server = None
 663         self.target_process = None
 664         self.binary = None
 665         self.start = 0
 666         self.logs = []
 667
 668     def early_applicable(self):
 669         """Return a false value if the test has determined it cannot run
 670         without ever needing to talk to the target or server."""
 671         # pylint: disable=no-self-use
 672         return True
 673
 674     def setup(self):
 675         pass
 676
 677     def compile(self):
 678         compile_args = getattr(self, 'compile_args', None)
 679         if compile_args:
 680             if compile_args not in BaseTest.compiled:
 681                 BaseTest.compiled[compile_args] = \
 682                         self.target.compile(self.hart, *compile_args)
 683         self.binary = BaseTest.compiled.get(compile_args)
 684
 685     def classSetup(self):
 686         self.compile()
 687         self.target_process = self.target.create()
 688         if self.target_process:
 689             self.logs.append(self.target_process.logname)
 690         try:
 691             self.server = self.target.server()
 692             self.logs.append(self.server.logname)
 693         except Exception:
 694             for log in self.logs:
 695                 print_log(log)
 696             raise
 697
 698     def classTeardown(self):
 699         del self.server
 700         del self.target_process
 701
 702     def postMortem(self):
 703         pass
 704
 705     def run(self, real_stdout):
 706         """
 707         If compile_args is set, compile a program and set self.binary.
 708
 709         Call setup().
 710
 711         Then call test() and return the result, displaying relevant information
 712         if an exception is raised.
 713         """
 714
 715         sys.stdout.flush()
 716
 717         if not self.early_applicable():
 718             return "not_applicable"
 719
 720         self.start = time.time()
 721
 722         try:
 723             self.classSetup()
 724             real_stdout.write("[%s] Temporary logs: %s\n" % (
 725                 type(self).__name__, ", ".join(self.logs)))
 726             self.setup()
 727             result = self.test()    # pylint: disable=no-member
 728         except TestNotApplicable:
 729             result = "not_applicable"
 730         except Exception as e: # pylint: disable=broad-except
 731             if isinstance(e, TestFailed):
 732                 result = "fail"
 733             else:
 734                 result = "exception"
 735             if isinstance(e, TestFailed):
 736                 header("Message")
 737                 print e.message
 738             header("Traceback")
 739             traceback.print_exc(file=sys.stdout)
 740             try:
 741                 self.postMortem()
 742             except Exception as e:  # pylint: disable=broad-except
 743                 header("postMortem Exception")
 744                 print e
 745                 traceback.print_exc(file=sys.stdout)
 746             return result
 747
 748         finally:
 749             for log in self.logs:
 750                 print_log(log)
 751             header("End of logs")
 752             self.classTeardown()
 753
 754         if not result:
 755             result = 'pass'
 756         return result
 757
 758 gdb_cmd = None
 759 class GdbTest(BaseTest):
 760     def __init__(self, target, hart=None):
 761         BaseTest.__init__(self, target, hart=hart)
 762         self.gdb = None
 763
 764     def classSetup(self):
 765         BaseTest.classSetup(self)
 766
 767         if gdb_cmd:
 768             self.gdb = Gdb(self.server.gdb_ports, gdb_cmd,
 769                     timeout=self.target.timeout_sec, binary=self.binary)
 770         else:
 771             self.gdb = Gdb(self.server.gdb_ports,
 772                     timeout=self.target.timeout_sec, binary=self.binary)
 773
 774         self.logs += self.gdb.lognames()
 775         self.gdb.connect()
 776
 777         self.gdb.global_command("set arch riscv:rv%d" % self.hart.xlen)
 778         self.gdb.global_command("set remotetimeout %d" %
 779             self.target.timeout_sec)
 780
 781         for cmd in self.target.gdb_setup:
 782             self.gdb.command(cmd)
 783
 784         self.gdb.select_hart(self.hart)
 785
 786         # FIXME: OpenOCD doesn't handle PRIV now
 787         #self.gdb.p("$priv=3")
 788
 789     def postMortem(self):
 790         if not self.gdb:
 791             return
 792         self.gdb.interrupt()
 793         self.gdb.command("info registers all", timeout=10)
 794
 795     def classTeardown(self):
 796         del self.gdb
 797         BaseTest.classTeardown(self)
 798
 799 class GdbSingleHartTest(GdbTest):
 800     def classSetup(self):
 801         GdbTest.classSetup(self)
 802
 803         for hart in self.target.harts:
 804             # Park all harts that we're not using in a safe place.
 805             if hart != self.hart:
 806                 self.gdb.select_hart(hart)
 807                 self.gdb.p("$pc=loop_forever")
 808         self.gdb.select_hart(self.hart)
 809
 810 class ExamineTarget(GdbTest):
 811     def test(self):
 812         for hart in self.target.harts:
 813             self.gdb.select_hart(hart)
 814
 815             hart.misa = self.gdb.p("$misa")
 816
 817             txt = "RV"
 818             if (hart.misa >> 30) == 1:
 819                 txt += "32"
 820             elif (hart.misa >> 62) == 2:
 821                 txt += "64"
 822             elif (hart.misa >> 126) == 3:
 823                 txt += "128"
 824             else:
 825                 raise TestFailed("Couldn't determine XLEN from $misa (0x%x)" %
 826                         self.hart.misa)
 827
 828             for i in range(26):
 829                 if hart.misa & (1<<i):
 830                     txt += chr(i + ord('A'))
 831             print txt,
 832
 833 class TestFailed(Exception):
 834     def __init__(self, message):
 835         Exception.__init__(self)
 836         self.message = message
 837
 838 class TestNotApplicable(Exception):
 839     def __init__(self, message):
 840         Exception.__init__(self)
 841         self.message = message
 842
 843 def assertEqual(a, b):
 844     if a != b:
 845         raise TestFailed("%r != %r" % (a, b))
 846
 847 def assertNotEqual(a, b):
 848     if a == b:
 849         raise TestFailed("%r == %r" % (a, b))
 850
 851 def assertIn(a, b):
 852     if a not in b:
 853         raise TestFailed("%r not in %r" % (a, b))
 854
 855 def assertNotIn(a, b):
 856     if a in b:
 857         raise TestFailed("%r in %r" % (a, b))
 858
 859 def assertGreater(a, b):
 860     if not a > b:
 861         raise TestFailed("%r not greater than %r" % (a, b))
 862
 863 def assertLess(a, b):
 864     if not a < b:
 865         raise TestFailed("%r not less than %r" % (a, b))
 866
 867 def assertTrue(a):
 868     if not a:
 869         raise TestFailed("%r is not True" % a)
 870
 871 def assertRegexpMatches(text, regexp):
 872     if not re.search(regexp, text):
 873         raise TestFailed("can't find %r in %r" % (regexp, text))