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