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