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         print "[%s] %s in %.2fs" % (name, result, time.time() - start)
 610         if result not in good_results and parsed.print_failures:
 611             sys.stdout.write(open(log_name).read())
 612         sys.stdout.flush()
 613         results.setdefault(result, []).append((name, log_name))
 614         count += 1
 615         if result not in good_results and parsed.fail_fast:
 616             break
 617
 618     return results, count
 619
 620 def print_results(results):
 621     result = 0
 622     for key, value in results.iteritems():
 623         print "%d tests returned %s" % (len(value), key)
 624         if key not in good_results:
 625             result = 1
 626             for name, log_name in value:
 627                 print "   %s > %s" % (name, log_name)
 628
 629     return result
 630
 631 def add_test_run_options(parser):
 632     parser.add_argument("--logs", default="logs",
 633             help="Store logs in the specified directory.")
 634     parser.add_argument("--fail-fast", "-f", action="store_true",
 635             help="Exit as soon as any test fails.")
 636     parser.add_argument("--print-failures", action="store_true",
 637             help="When a test fails, print the log file to stdout.")
 638     parser.add_argument("--print-log-names", "--pln", action="store_true",
 639             help="Print names of temporary log files as soon as they are "
 640             "created.")
 641     parser.add_argument("test", nargs='*',
 642             help="Run only tests that are named here.")
 643     parser.add_argument("--gdb",
 644             help="The command to use to start gdb.")
 645     parser.add_argument("--misaval",
 646             help="Don't run ExamineTarget, just assume the misa value which is "
 647             "specified.")
 648
 649 def header(title, dash='-', length=78):
 650     if title:
 651         dashes = dash * (length - 4 - len(title))
 652         before = dashes[:len(dashes)/2]
 653         after = dashes[len(dashes)/2:]
 654         print "%s[ %s ]%s" % (before, title, after)
 655     else:
 656         print dash * length
 657
 658 def print_log(path):
 659     header(path)
 660     for l in open(path, "r"):
 661         sys.stdout.write(l)
 662     print
 663
 664 class BaseTest(object):
 665     compiled = {}
 666
 667     def __init__(self, target, hart=None):
 668         self.target = target
 669         if hart:
 670             self.hart = hart
 671         else:
 672             self.hart = random.choice(target.harts)
 673             self.hart = target.harts[-1]    #<<<
 674         self.server = None
 675         self.target_process = None
 676         self.binary = None
 677         self.start = 0
 678         self.logs = []
 679
 680     def early_applicable(self):
 681         """Return a false value if the test has determined it cannot run
 682         without ever needing to talk to the target or server."""
 683         # pylint: disable=no-self-use
 684         return True
 685
 686     def setup(self):
 687         pass
 688
 689     def compile(self):
 690         compile_args = getattr(self, 'compile_args', None)
 691         if compile_args:
 692             if compile_args not in BaseTest.compiled:
 693                 BaseTest.compiled[compile_args] = \
 694                         self.target.compile(self.hart, *compile_args)
 695         self.binary = BaseTest.compiled.get(compile_args)
 696
 697     def classSetup(self):
 698         self.compile()
 699         self.target_process = self.target.create()
 700         if self.target_process:
 701             self.logs.append(self.target_process.logname)
 702         try:
 703             self.server = self.target.server()
 704             self.logs.append(self.server.logname)
 705         except Exception:
 706             for log in self.logs:
 707                 print_log(log)
 708             raise
 709
 710     def classTeardown(self):
 711         del self.server
 712         del self.target_process
 713
 714     def postMortem(self):
 715         pass
 716
 717     def run(self):
 718         """
 719         If compile_args is set, compile a program and set self.binary.
 720
 721         Call setup().
 722
 723         Then call test() and return the result, displaying relevant information
 724         if an exception is raised.
 725         """
 726
 727         sys.stdout.flush()
 728
 729         if not self.early_applicable():
 730             return "not_applicable"
 731
 732         self.start = time.time()
 733
 734         try:
 735             self.classSetup()
 736             self.setup()
 737             result = self.test()    # pylint: disable=no-member
 738         except TestNotApplicable:
 739             result = "not_applicable"
 740         except Exception as e: # pylint: disable=broad-except
 741             if isinstance(e, TestFailed):
 742                 result = "fail"
 743             else:
 744                 result = "exception"
 745             if isinstance(e, TestFailed):
 746                 header("Message")
 747                 print e.message
 748             header("Traceback")
 749             traceback.print_exc(file=sys.stdout)
 750             try:
 751                 self.postMortem()
 752             except Exception as e:  # pylint: disable=broad-except
 753                 header("postMortem Exception")
 754                 print e
 755                 traceback.print_exc(file=sys.stdout)
 756             return result
 757
 758         finally:
 759             for log in self.logs:
 760                 print_log(log)
 761             header("End of logs")
 762             self.classTeardown()
 763
 764         if not result:
 765             result = 'pass'
 766         return result
 767
 768 gdb_cmd = None
 769 class GdbTest(BaseTest):
 770     def __init__(self, target, hart=None):
 771         BaseTest.__init__(self, target, hart=hart)
 772         self.gdb = None
 773
 774     def classSetup(self):
 775         BaseTest.classSetup(self)
 776
 777         if gdb_cmd:
 778             self.gdb = Gdb(self.server.gdb_ports, gdb_cmd,
 779                     timeout=self.target.timeout_sec, binary=self.binary)
 780         else:
 781             self.gdb = Gdb(self.server.gdb_ports,
 782                     timeout=self.target.timeout_sec, binary=self.binary)
 783
 784         self.logs += self.gdb.lognames()
 785         self.gdb.connect()
 786
 787         self.gdb.global_command("set arch riscv:rv%d" % self.hart.xlen)
 788         self.gdb.global_command("set remotetimeout %d" %
 789             self.target.timeout_sec)
 790
 791         for cmd in self.target.gdb_setup:
 792             self.gdb.command(cmd)
 793
 794         self.gdb.select_hart(self.hart)
 795
 796         # FIXME: OpenOCD doesn't handle PRIV now
 797         #self.gdb.p("$priv=3")
 798
 799     def postMortem(self):
 800         if not self.gdb:
 801             return
 802         self.gdb.interrupt()
 803         self.gdb.command("info registers all", timeout=10)
 804
 805     def classTeardown(self):
 806         del self.gdb
 807         BaseTest.classTeardown(self)
 808
 809 class GdbSingleHartTest(GdbTest):
 810     def classSetup(self):
 811         GdbTest.classSetup(self)
 812
 813         for hart in self.target.harts:
 814             # Park all harts that we're not using in a safe place.
 815             if hart != self.hart:
 816                 self.gdb.select_hart(hart)
 817                 self.gdb.p("$pc=loop_forever")
 818         self.gdb.select_hart(self.hart)
 819
 820 class ExamineTarget(GdbTest):
 821     def test(self):
 822         for hart in self.target.harts:
 823             self.gdb.select_hart(hart)
 824
 825             hart.misa = self.gdb.p("$misa")
 826
 827             txt = "RV"
 828             if (hart.misa >> 30) == 1:
 829                 txt += "32"
 830             elif (hart.misa >> 62) == 2:
 831                 txt += "64"
 832             elif (hart.misa >> 126) == 3:
 833                 txt += "128"
 834             else:
 835                 raise TestFailed("Couldn't determine XLEN from $misa (0x%x)" %
 836                         self.hart.misa)
 837
 838             for i in range(26):
 839                 if hart.misa & (1<<i):
 840                     txt += chr(i + ord('A'))
 841             print txt,
 842
 843 class TestFailed(Exception):
 844     def __init__(self, message):
 845         Exception.__init__(self)
 846         self.message = message
 847
 848 class TestNotApplicable(Exception):
 849     def __init__(self, message):
 850         Exception.__init__(self)
 851         self.message = message
 852
 853 def assertEqual(a, b):
 854     if a != b:
 855         raise TestFailed("%r != %r" % (a, b))
 856
 857 def assertNotEqual(a, b):
 858     if a == b:
 859         raise TestFailed("%r == %r" % (a, b))
 860
 861 def assertIn(a, b):
 862     if a not in b:
 863         raise TestFailed("%r not in %r" % (a, b))
 864
 865 def assertNotIn(a, b):
 866     if a in b:
 867         raise TestFailed("%r in %r" % (a, b))
 868
 869 def assertGreater(a, b):
 870     if not a > b:
 871         raise TestFailed("%r not greater than %r" % (a, b))
 872
 873 def assertLess(a, b):
 874     if not a < b:
 875         raise TestFailed("%r not less than %r" % (a, b))
 876
 877 def assertTrue(a):
 878     if not a:
 879         raise TestFailed("%r is not True" % a)
 880
 881 def assertRegexpMatches(text, regexp):
 882     if not re.search(regexp, text):
 883         raise TestFailed("can't find %r in %r" % (regexp, text))