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
 315     def __init__(self, ports,
 316             cmd=os.path.expandvars("$RISCV/bin/riscv64-unknown-elf-gdb"),
 317             timeout=60,
 318             binary=None):
 319         assert ports
 320
 321         self.stack = []
 322
 323         self.logfiles = []
 324         self.children = []
 325         for port in ports:
 326             logfile = tempfile.NamedTemporaryFile(prefix="gdb@%d-" % port,
 327                     suffix=".log")
 328             self.logfiles.append(logfile)
 329             child = pexpect.spawn(cmd)
 330             child.logfile = logfile
 331             child.logfile.write("+ %s\n" % cmd)
 332             self.children.append(child)
 333         self.active_child = self.children[0]
 334
 335         self.harts = {}
 336         for port, child in zip(ports, self.children):
 337             self.select_child(child)
 338             self.wait()
 339             self.command("set confirm off")
 340             self.command("set width 0")
 341             self.command("set height 0")
 342             # Force consistency.
 343             self.command("set print entry-values no")
 344             self.command("set remotetimeout %d" % timeout)
 345             self.command("target extended-remote localhost:%d" % port)
 346             if binary:
 347                 self.command("file %s" % binary)
 348             threads = self.threads()
 349             for t in threads:
 350                 hartid = None
 351                 if t.name:
 352                     m = re.search(r"Hart (\d+)", t.name)
 353                     if m:
 354                         hartid = int(m.group(1))
 355                 if hartid is None:
 356                     if self.harts:
 357                         hartid = max(self.harts) + 1
 358                     else:
 359                         hartid = 0
 360                 self.harts[hartid] = (child, t)
 361
 362     def __del__(self):
 363         for child in self.children:
 364             del child
 365
 366     def lognames(self):
 367         return [logfile.name for logfile in self.logfiles]
 368
 369     def select_child(self, child):
 370         self.active_child = child
 371
 372     def select_hart(self, hart):
 373         child, thread = self.harts[hart.id]
 374         self.select_child(child)
 375         output = self.command("thread %s" % thread.id)
 376         assert "Unknown" not in output
 377
 378     def push_state(self):
 379         self.stack.append({
 380             'active_child': self.active_child
 381             })
 382
 383     def pop_state(self):
 384         state = self.stack.pop()
 385         self.active_child = state['active_child']
 386
 387     def wait(self):
 388         """Wait for prompt."""
 389         self.active_child.expect(r"\(gdb\)")
 390
 391     def command(self, command, timeout=6000):
 392         """timeout is in seconds"""
 393         self.active_child.sendline(command)
 394         self.active_child.expect("\n", timeout=timeout)
 395         self.active_child.expect(r"\(gdb\)", timeout=timeout)
 396         return self.active_child.before.strip()
 397
 398     def global_command(self, command):
 399         """Execute this command on every gdb that we control."""
 400         with PrivateState(self):
 401             for child in self.children:
 402                 self.select_child(child)
 403                 self.command(command)
 404
 405     def c(self, wait=True, timeout=-1, async=False):
 406         """
 407         Dumb c command.
 408         In RTOS mode, gdb will resume all harts.
 409         In multi-gdb mode, this command will just go to the current gdb, so
 410         will only resume one hart.
 411         """
 412         if async:
 413             async = "&"
 414         else:
 415             async = ""
 416         if wait:
 417             output = self.command("c%s" % async, timeout=timeout)
 418             assert "Continuing" in output
 419             return output
 420         else:
 421             self.active_child.sendline("c%s" % async)
 422             self.active_child.expect("Continuing")
 423
 424     def c_all(self):
 425         """
 426         Resume every hart.
 427
 428         This function works fine when using multiple gdb sessions, but the
 429         caller must be careful when using it nonetheless. gdb's behavior is to
 430         not set breakpoints until just before the hart is resumed, and then
 431         clears them as soon as the hart halts. That means that you can't set
 432         one software breakpoint, and expect multiple harts to hit it. It's
 433         possible that the first hart completes set/run/halt/clear before the
 434         second hart even gets to resume, so it will never hit the breakpoint.
 435         """
 436         with PrivateState(self):
 437             for child in self.children:
 438                 child.sendline("c")
 439                 child.expect("Continuing")
 440
 441             # Now wait for them all to halt
 442             for child in self.children:
 443                 child.expect(r"\(gdb\)")
 444
 445     def interrupt(self):
 446         self.active_child.send("\003")
 447         self.active_child.expect(r"\(gdb\)", timeout=6000)
 448         return self.active_child.before.strip()
 449
 450     def x(self, address, size='w'):
 451         output = self.command("x/%s %s" % (size, address))
 452         value = int(output.split(':')[1].strip(), 0)
 453         return value
 454
 455     def p_raw(self, obj):
 456         output = self.command("p %s" % obj)
 457         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 458         if m:
 459             raise CannotAccess(int(m.group(1), 0))
 460         return output.split('=')[-1].strip()
 461
 462     def parse_string(self, text):
 463         text = text.strip()
 464         if text.startswith("{") and text.endswith("}"):
 465             inner = text[1:-1]
 466             return [self.parse_string(t) for t in inner.split(", ")]
 467         elif text.startswith('"') and text.endswith('"'):
 468             return text[1:-1]
 469         else:
 470             return int(text, 0)
 471
 472     def p(self, obj, fmt="/x"):
 473         output = self.command("p%s %s" % (fmt, obj))
 474         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 475         if m:
 476             raise CannotAccess(int(m.group(1), 0))
 477         rhs = output.split('=')[-1]
 478         return self.parse_string(rhs)
 479
 480     def p_string(self, obj):
 481         output = self.command("p %s" % obj)
 482         value = shlex.split(output.split('=')[-1].strip())[1]
 483         return value
 484
 485     def stepi(self):
 486         output = self.command("stepi", timeout=60)
 487         return output
 488
 489     def load(self):
 490         output = self.command("load", timeout=6000)
 491         assert "failed" not in  output
 492         assert "Transfer rate" in output
 493
 494     def b(self, location):
 495         output = self.command("b %s" % location)
 496         assert "not defined" not in output
 497         assert "Breakpoint" in output
 498         return output
 499
 500     def hbreak(self, location):
 501         output = self.command("hbreak %s" % location)
 502         assert "not defined" not in output
 503         assert "Hardware assisted breakpoint" in output
 504         return output
 505
 506     def threads(self):
 507         output = self.command("info threads")
 508         threads = []
 509         for line in output.splitlines():
 510             m = re.match(
 511                     r"[\s\*]*(\d+)\s*"
 512                     r"(Remote target|Thread (\d+)\s*\(Name: ([^\)]+))"
 513                     r"\s*(.*)", line)
 514             if m:
 515                 threads.append(Thread(*m.groups()))
 516         assert threads
 517         #>>>if not threads:
 518         #>>>    threads.append(Thread('1', '1', 'Default', '???'))
 519         return threads
 520
 521     def thread(self, thread):
 522         return self.command("thread %s" % thread.id)
 523
 524     def where(self):
 525         return self.command("where 1")
 526
 527 class PrivateState(object):
 528     def __init__(self, gdb):
 529         self.gdb = gdb
 530
 531     def __enter__(self):
 532         self.gdb.push_state()
 533
 534     def __exit__(self, _type, _value, _traceback):
 535         self.gdb.pop_state()
 536
 537 def run_all_tests(module, target, parsed):
 538     if not os.path.exists(parsed.logs):
 539         os.makedirs(parsed.logs)
 540
 541     overall_start = time.time()
 542
 543     global gdb_cmd  # pylint: disable=global-statement
 544     gdb_cmd = parsed.gdb
 545
 546     todo = []
 547     examine_added = False
 548     for hart in target.harts:
 549         if parsed.misaval:
 550             hart.misa = int(parsed.misaval, 16)
 551             print "Using $misa from command line: 0x%x" % hart.misa
 552         elif hart.misa:
 553             print "Using $misa from hart definition: 0x%x" % hart.misa
 554         elif not examine_added:
 555             todo.append(("ExamineTarget", ExamineTarget, None))
 556             examine_added = True
 557
 558     for name in dir(module):
 559         definition = getattr(module, name)
 560         if isinstance(definition, type) and hasattr(definition, 'test') and \
 561                 (not parsed.test or any(test in name for test in parsed.test)):
 562             todo.append((name, definition, None))
 563
 564     results, count = run_tests(parsed, target, todo)
 565
 566     header("ran %d tests in %.0fs" % (count, time.time() - overall_start),
 567             dash=':')
 568
 569     return print_results(results)
 570
 571 good_results = set(('pass', 'not_applicable'))
 572 def run_tests(parsed, target, todo):
 573     results = {}
 574     count = 0
 575
 576     for name, definition, hart in todo:
 577         log_name = os.path.join(parsed.logs, "%s-%s-%s.log" %
 578                 (time.strftime("%Y%m%d-%H%M%S"), type(target).__name__, name))
 579         log_fd = open(log_name, 'w')
 580         print "[%s] Starting > %s" % (name, log_name)
 581         instance = definition(target, hart)
 582         sys.stdout.flush()
 583         log_fd.write("Test: %s\n" % name)
 584         log_fd.write("Target: %s\n" % type(target).__name__)
 585         start = time.time()
 586         real_stdout = sys.stdout
 587         sys.stdout = log_fd
 588         try:
 589             result = instance.run(real_stdout)
 590             log_fd.write("Result: %s\n" % result)
 591         finally:
 592             sys.stdout = real_stdout
 593             log_fd.write("Time elapsed: %.2fs\n" % (time.time() - start))
 594         print "[%s] %s in %.2fs" % (name, result, time.time() - start)
 595         if result not in good_results and parsed.print_failures:
 596             sys.stdout.write(open(log_name).read())
 597         sys.stdout.flush()
 598         results.setdefault(result, []).append((name, log_name))
 599         count += 1
 600         if result not in good_results and parsed.fail_fast:
 601             break
 602
 603     return results, count
 604
 605 def print_results(results):
 606     result = 0
 607     for key, value in results.iteritems():
 608         print "%d tests returned %s" % (len(value), key)
 609         if key not in good_results:
 610             result = 1
 611             for name, log_name in value:
 612                 print "   %s > %s" % (name, log_name)
 613
 614     return result
 615
 616 def add_test_run_options(parser):
 617     parser.add_argument("--logs", default="logs",
 618             help="Store logs in the specified directory.")
 619     parser.add_argument("--fail-fast", "-f", action="store_true",
 620             help="Exit as soon as any test fails.")
 621     parser.add_argument("--print-failures", action="store_true",
 622             help="When a test fails, print the log file to stdout.")
 623     parser.add_argument("test", nargs='*',
 624             help="Run only tests that are named here.")
 625     parser.add_argument("--gdb",
 626             help="The command to use to start gdb.")
 627     parser.add_argument("--misaval",
 628             help="Don't run ExamineTarget, just assume the misa value which is "
 629             "specified.")
 630
 631 def header(title, dash='-', length=78):
 632     if title:
 633         dashes = dash * (length - 4 - len(title))
 634         before = dashes[:len(dashes)/2]
 635         after = dashes[len(dashes)/2:]
 636         print "%s[ %s ]%s" % (before, title, after)
 637     else:
 638         print dash * length
 639
 640 def print_log(path):
 641     header(path)
 642     for l in open(path, "r"):
 643         sys.stdout.write(l)
 644     print
 645
 646 class BaseTest(object):
 647     compiled = {}
 648
 649     def __init__(self, target, hart=None):
 650         self.target = target
 651         if hart:
 652             self.hart = hart
 653         else:
 654             self.hart = random.choice(target.harts)
 655             self.hart = target.harts[-1]    #<<<
 656         self.server = None
 657         self.target_process = None
 658         self.binary = None
 659         self.start = 0
 660         self.logs = []
 661
 662     def early_applicable(self):
 663         """Return a false value if the test has determined it cannot run
 664         without ever needing to talk to the target or server."""
 665         # pylint: disable=no-self-use
 666         return True
 667
 668     def setup(self):
 669         pass
 670
 671     def compile(self):
 672         compile_args = getattr(self, 'compile_args', None)
 673         if compile_args:
 674             if compile_args not in BaseTest.compiled:
 675                 BaseTest.compiled[compile_args] = \
 676                         self.target.compile(self.hart, *compile_args)
 677         self.binary = BaseTest.compiled.get(compile_args)
 678
 679     def classSetup(self):
 680         self.compile()
 681         self.target_process = self.target.create()
 682         if self.target_process:
 683             self.logs.append(self.target_process.logname)
 684         try:
 685             self.server = self.target.server()
 686             self.logs.append(self.server.logname)
 687         except Exception:
 688             for log in self.logs:
 689                 print_log(log)
 690             raise
 691
 692     def classTeardown(self):
 693         del self.server
 694         del self.target_process
 695
 696     def postMortem(self):
 697         pass
 698
 699     def run(self, real_stdout):
 700         """
 701         If compile_args is set, compile a program and set self.binary.
 702
 703         Call setup().
 704
 705         Then call test() and return the result, displaying relevant information
 706         if an exception is raised.
 707         """
 708
 709         sys.stdout.flush()
 710
 711         if not self.early_applicable():
 712             return "not_applicable"
 713
 714         self.start = time.time()
 715
 716         try:
 717             self.classSetup()
 718             real_stdout.write("[%s] Temporary logs: %s\n" % (
 719                 type(self).__name__, ", ".join(self.logs)))
 720             self.setup()
 721             result = self.test()    # pylint: disable=no-member
 722         except TestNotApplicable:
 723             result = "not_applicable"
 724         except Exception as e: # pylint: disable=broad-except
 725             if isinstance(e, TestFailed):
 726                 result = "fail"
 727             else:
 728                 result = "exception"
 729             if isinstance(e, TestFailed):
 730                 header("Message")
 731                 print e.message
 732             header("Traceback")
 733             traceback.print_exc(file=sys.stdout)
 734             try:
 735                 self.postMortem()
 736             except Exception as e:  # pylint: disable=broad-except
 737                 header("postMortem Exception")
 738                 print e
 739                 traceback.print_exc(file=sys.stdout)
 740             return result
 741
 742         finally:
 743             for log in self.logs:
 744                 print_log(log)
 745             header("End of logs")
 746             self.classTeardown()
 747
 748         if not result:
 749             result = 'pass'
 750         return result
 751
 752 gdb_cmd = None
 753 class GdbTest(BaseTest):
 754     def __init__(self, target, hart=None):
 755         BaseTest.__init__(self, target, hart=hart)
 756         self.gdb = None
 757
 758     def classSetup(self):
 759         BaseTest.classSetup(self)
 760
 761         if gdb_cmd:
 762             self.gdb = Gdb(self.server.gdb_ports, gdb_cmd, timeout=self.target.timeout_sec, binary=self.binary)
 763         else:
 764             self.gdb = Gdb(self.server.gdb_ports, timeout=self.target.timeout_sec, binary=self.binary)
 765
 766         self.logs += self.gdb.lognames()
 767
 768         self.gdb.global_command("set arch riscv:rv%d" % self.hart.xlen)
 769         self.gdb.global_command("set remotetimeout %d" %
 770             self.target.timeout_sec)
 771
 772         for cmd in self.target.gdb_setup:
 773             self.gdb.command(cmd)
 774
 775         self.gdb.select_hart(self.hart)
 776
 777         # FIXME: OpenOCD doesn't handle PRIV now
 778         #self.gdb.p("$priv=3")
 779
 780     def postMortem(self):
 781         if not self.gdb:
 782             return
 783         self.gdb.interrupt()
 784         self.gdb.command("info registers all", timeout=10)
 785
 786     def classTeardown(self):
 787         del self.gdb
 788         BaseTest.classTeardown(self)
 789
 790 class GdbSingleHartTest(GdbTest):
 791     def classSetup(self):
 792         GdbTest.classSetup(self)
 793
 794         for hart in self.target.harts:
 795             # Park all harts that we're not using in a safe place.
 796             if hart != self.hart:
 797                 self.gdb.select_hart(hart)
 798                 self.gdb.p("$pc=loop_forever")
 799         self.gdb.select_hart(self.hart)
 800
 801 class ExamineTarget(GdbTest):
 802     def test(self):
 803         for hart in self.target.harts:
 804             self.gdb.select_hart(hart)
 805
 806             hart.misa = self.gdb.p("$misa")
 807
 808             txt = "RV"
 809             if (hart.misa >> 30) == 1:
 810                 txt += "32"
 811             elif (hart.misa >> 62) == 2:
 812                 txt += "64"
 813             elif (hart.misa >> 126) == 3:
 814                 txt += "128"
 815             else:
 816                 raise TestFailed("Couldn't determine XLEN from $misa (0x%x)" %
 817                         self.hart.misa)
 818
 819             for i in range(26):
 820                 if hart.misa & (1<<i):
 821                     txt += chr(i + ord('A'))
 822             print txt,
 823
 824 class TestFailed(Exception):
 825     def __init__(self, message):
 826         Exception.__init__(self)
 827         self.message = message
 828
 829 class TestNotApplicable(Exception):
 830     def __init__(self, message):
 831         Exception.__init__(self)
 832         self.message = message
 833
 834 def assertEqual(a, b):
 835     if a != b:
 836         raise TestFailed("%r != %r" % (a, b))
 837
 838 def assertNotEqual(a, b):
 839     if a == b:
 840         raise TestFailed("%r == %r" % (a, b))
 841
 842 def assertIn(a, b):
 843     if a not in b:
 844         raise TestFailed("%r not in %r" % (a, b))
 845
 846 def assertNotIn(a, b):
 847     if a in b:
 848         raise TestFailed("%r in %r" % (a, b))
 849
 850 def assertGreater(a, b):
 851     if not a > b:
 852         raise TestFailed("%r not greater than %r" % (a, b))
 853
 854 def assertLess(a, b):
 855     if not a < b:
 856         raise TestFailed("%r not less than %r" % (a, b))
 857
 858 def assertTrue(a):
 859     if not a:
 860         raise TestFailed("%r is not True" % a)
 861
 862 def assertRegexpMatches(text, regexp):
 863     if not re.search(regexp, text):
 864         raise TestFailed("can't find %r in %r" % (regexp, text))