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         """Resume every hart."""
 424         with PrivateState(self):
 425             for child in self.children:
 426                 child.sendline("c")
 427                 child.expect("Continuing")
 428
 429             # Now wait for them all to halt
 430             for child in self.children:
 431                 child.expect(r"\(gdb\)")
 432
 433     def interrupt(self):
 434         self.active_child.send("\003")
 435         self.active_child.expect(r"\(gdb\)", timeout=6000)
 436         return self.active_child.before.strip()
 437
 438     def x(self, address, size='w'):
 439         output = self.command("x/%s %s" % (size, address))
 440         value = int(output.split(':')[1].strip(), 0)
 441         return value
 442
 443     def p_raw(self, obj):
 444         output = self.command("p %s" % obj)
 445         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 446         if m:
 447             raise CannotAccess(int(m.group(1), 0))
 448         return output.split('=')[-1].strip()
 449
 450     def parse_string(self, text):
 451         text = text.strip()
 452         if text.startswith("{") and text.endswith("}"):
 453             inner = text[1:-1]
 454             return [self.parse_string(t) for t in inner.split(", ")]
 455         elif text.startswith('"') and text.endswith('"'):
 456             return text[1:-1]
 457         else:
 458             return int(text, 0)
 459
 460     def p(self, obj, fmt="/x"):
 461         output = self.command("p%s %s" % (fmt, obj))
 462         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 463         if m:
 464             raise CannotAccess(int(m.group(1), 0))
 465         rhs = output.split('=')[-1]
 466         return self.parse_string(rhs)
 467
 468     def p_string(self, obj):
 469         output = self.command("p %s" % obj)
 470         value = shlex.split(output.split('=')[-1].strip())[1]
 471         return value
 472
 473     def stepi(self):
 474         output = self.command("stepi", timeout=60)
 475         return output
 476
 477     def load(self):
 478         output = self.command("load", timeout=6000)
 479         assert "failed" not in  output
 480         assert "Transfer rate" in output
 481
 482     def b(self, location):
 483         output = self.command("b %s" % location)
 484         assert "not defined" not in output
 485         assert "Breakpoint" in output
 486         return output
 487
 488     def hbreak(self, location):
 489         output = self.command("hbreak %s" % location)
 490         assert "not defined" not in output
 491         assert "Hardware assisted breakpoint" in output
 492         return output
 493
 494     def threads(self):
 495         output = self.command("info threads")
 496         threads = []
 497         for line in output.splitlines():
 498             m = re.match(
 499                     r"[\s\*]*(\d+)\s*"
 500                     r"(Remote target|Thread (\d+)\s*\(Name: ([^\)]+))"
 501                     r"\s*(.*)", line)
 502             if m:
 503                 threads.append(Thread(*m.groups()))
 504         assert threads
 505         #>>>if not threads:
 506         #>>>    threads.append(Thread('1', '1', 'Default', '???'))
 507         return threads
 508
 509     def thread(self, thread):
 510         return self.command("thread %s" % thread.id)
 511
 512     def where(self):
 513         return self.command("where 1")
 514
 515 class PrivateState(object):
 516     def __init__(self, gdb):
 517         self.gdb = gdb
 518
 519     def __enter__(self):
 520         self.gdb.push_state()
 521
 522     def __exit__(self, _type, _value, _traceback):
 523         self.gdb.pop_state()
 524
 525 def run_all_tests(module, target, parsed):
 526     if not os.path.exists(parsed.logs):
 527         os.makedirs(parsed.logs)
 528
 529     overall_start = time.time()
 530
 531     global gdb_cmd  # pylint: disable=global-statement
 532     gdb_cmd = parsed.gdb
 533
 534     todo = []
 535     for hart in target.harts:
 536         if parsed.misaval:
 537             hart.misa = int(parsed.misaval, 16)
 538             print "Using $misa from command line: 0x%x" % hart.misa
 539         elif hart.misa:
 540             print "Using $misa from hart definition: 0x%x" % hart.misa
 541         else:
 542             todo.append(("ExamineTarget", ExamineTarget, hart))
 543
 544     for name in dir(module):
 545         definition = getattr(module, name)
 546         if type(definition) == type and hasattr(definition, 'test') and \
 547                 (not parsed.test or any(test in name for test in parsed.test)):
 548             todo.append((name, definition, None))
 549
 550     results, count = run_tests(parsed, target, todo)
 551
 552     header("ran %d tests in %.0fs" % (count, time.time() - overall_start),
 553             dash=':')
 554
 555     return print_results(results)
 556
 557 good_results = set(('pass', 'not_applicable'))
 558 def run_tests(parsed, target, todo):
 559     results = {}
 560     count = 0
 561
 562     for name, definition, hart in todo:
 563         log_name = os.path.join(parsed.logs, "%s-%s-%s.log" %
 564                 (time.strftime("%Y%m%d-%H%M%S"), type(target).__name__, name))
 565         log_fd = open(log_name, 'w')
 566         print "[%s] Starting > %s" % (name, log_name)
 567         instance = definition(target, hart)
 568         sys.stdout.flush()
 569         log_fd.write("Test: %s\n" % name)
 570         log_fd.write("Target: %s\n" % type(target).__name__)
 571         start = time.time()
 572         real_stdout = sys.stdout
 573         sys.stdout = log_fd
 574         try:
 575             result = instance.run(real_stdout)
 576             log_fd.write("Result: %s\n" % result)
 577         finally:
 578             sys.stdout = real_stdout
 579             log_fd.write("Time elapsed: %.2fs\n" % (time.time() - start))
 580         print "[%s] %s in %.2fs" % (name, result, time.time() - start)
 581         if result not in good_results and parsed.print_failures:
 582             sys.stdout.write(open(log_name).read())
 583         sys.stdout.flush()
 584         results.setdefault(result, []).append((name, log_name))
 585         count += 1
 586         if result not in good_results and parsed.fail_fast:
 587             break
 588
 589     return results, count
 590
 591 def print_results(results):
 592     result = 0
 593     for key, value in results.iteritems():
 594         print "%d tests returned %s" % (len(value), key)
 595         if key not in good_results:
 596             result = 1
 597             for name, log_name in value:
 598                 print "   %s > %s" % (name, log_name)
 599
 600     return result
 601
 602 def add_test_run_options(parser):
 603     parser.add_argument("--logs", default="logs",
 604             help="Store logs in the specified directory.")
 605     parser.add_argument("--fail-fast", "-f", action="store_true",
 606             help="Exit as soon as any test fails.")
 607     parser.add_argument("--print-failures", action="store_true",
 608             help="When a test fails, print the log file to stdout.")
 609     parser.add_argument("test", nargs='*',
 610             help="Run only tests that are named here.")
 611     parser.add_argument("--gdb",
 612             help="The command to use to start gdb.")
 613     parser.add_argument("--misaval",
 614             help="Don't run ExamineTarget, just assume the misa value which is "
 615             "specified.")
 616
 617 def header(title, dash='-', length=78):
 618     if title:
 619         dashes = dash * (length - 4 - len(title))
 620         before = dashes[:len(dashes)/2]
 621         after = dashes[len(dashes)/2:]
 622         print "%s[ %s ]%s" % (before, title, after)
 623     else:
 624         print dash * length
 625
 626 def print_log(path):
 627     header(path)
 628     for l in open(path, "r"):
 629         sys.stdout.write(l)
 630     print
 631
 632 class BaseTest(object):
 633     compiled = {}
 634
 635     def __init__(self, target, hart=None):
 636         self.target = target
 637         if hart:
 638             self.hart = hart
 639         else:
 640             self.hart = random.choice(target.harts)
 641             self.hart = target.harts[-1]    #<<<
 642         self.server = None
 643         self.target_process = None
 644         self.binary = None
 645         self.start = 0
 646         self.logs = []
 647
 648     def early_applicable(self):
 649         """Return a false value if the test has determined it cannot run
 650         without ever needing to talk to the target or server."""
 651         # pylint: disable=no-self-use
 652         return True
 653
 654     def setup(self):
 655         pass
 656
 657     def compile(self):
 658         compile_args = getattr(self, 'compile_args', None)
 659         if compile_args:
 660             if compile_args not in BaseTest.compiled:
 661                 # pylint: disable=star-args
 662                 BaseTest.compiled[compile_args] = \
 663                         self.target.compile(self.hart, *compile_args)
 664         self.binary = BaseTest.compiled.get(compile_args)
 665
 666     def classSetup(self):
 667         self.compile()
 668         self.target_process = self.target.create()
 669         if self.target_process:
 670             self.logs.append(self.target_process.logname)
 671         try:
 672             self.server = self.target.server()
 673             self.logs.append(self.server.logname)
 674         except Exception:
 675             for log in self.logs:
 676                 print_log(log)
 677             raise
 678
 679     def classTeardown(self):
 680         del self.server
 681         del self.target_process
 682
 683     def postMortem(self):
 684         pass
 685
 686     def run(self, real_stdout):
 687         """
 688         If compile_args is set, compile a program and set self.binary.
 689
 690         Call setup().
 691
 692         Then call test() and return the result, displaying relevant information
 693         if an exception is raised.
 694         """
 695
 696         sys.stdout.flush()
 697
 698         if not self.early_applicable():
 699             return "not_applicable"
 700
 701         self.start = time.time()
 702
 703         try:
 704             self.classSetup()
 705             real_stdout.write("[%s] Temporary logs: %s\n" % (
 706                 type(self).__name__, ", ".join(self.logs)))
 707             self.setup()
 708             result = self.test()    # pylint: disable=no-member
 709         except TestNotApplicable:
 710             result = "not_applicable"
 711         except Exception as e: # pylint: disable=broad-except
 712             if isinstance(e, TestFailed):
 713                 result = "fail"
 714             else:
 715                 result = "exception"
 716             if isinstance(e, TestFailed):
 717                 header("Message")
 718                 print e.message
 719             header("Traceback")
 720             traceback.print_exc(file=sys.stdout)
 721             try:
 722                 self.postMortem()
 723             except Exception as e:  # pylint: disable=broad-except
 724                 header("postMortem Exception")
 725                 print e
 726                 traceback.print_exc(file=sys.stdout)
 727             return result
 728
 729         finally:
 730             for log in self.logs:
 731                 print_log(log)
 732             header("End of logs")
 733             self.classTeardown()
 734
 735         if not result:
 736             result = 'pass'
 737         return result
 738
 739 gdb_cmd = None
 740 class GdbTest(BaseTest):
 741     def __init__(self, target, hart=None):
 742         BaseTest.__init__(self, target, hart=hart)
 743         self.gdb = None
 744
 745     def classSetup(self):
 746         BaseTest.classSetup(self)
 747
 748         if gdb_cmd:
 749             self.gdb = Gdb(self.server.gdb_ports, gdb_cmd, binary=self.binary)
 750         else:
 751             self.gdb = Gdb(self.server.gdb_ports, binary=self.binary)
 752
 753         self.logs += self.gdb.lognames()
 754
 755         if self.target:
 756             self.gdb.global_command("set arch riscv:rv%d" % self.hart.xlen)
 757             self.gdb.global_command("set remotetimeout %d" %
 758                     self.target.timeout_sec)
 759
 760         for cmd in self.target.gdb_setup:
 761             self.gdb.command(cmd)
 762
 763         self.gdb.select_hart(self.hart)
 764
 765         # FIXME: OpenOCD doesn't handle PRIV now
 766         #self.gdb.p("$priv=3")
 767
 768     def postMortem(self):
 769         if not self.gdb:
 770             return
 771         self.gdb.interrupt()
 772         self.gdb.command("info registers all", timeout=10)
 773
 774     def classTeardown(self):
 775         del self.gdb
 776         BaseTest.classTeardown(self)
 777
 778 class GdbSingleHartTest(GdbTest):
 779     def classSetup(self):
 780         GdbTest.classSetup(self)
 781
 782         for hart in self.target.harts:
 783             # Park all harts that we're not using in a safe place.
 784             if hart != self.hart:
 785                 self.gdb.select_hart(hart)
 786                 self.gdb.p("$pc=loop_forever")
 787         self.gdb.select_hart(self.hart)
 788
 789 class ExamineTarget(GdbTest):
 790     def test(self):
 791         self.hart.misa = self.gdb.p("$misa")
 792
 793         txt = "RV"
 794         if (self.hart.misa >> 30) == 1:
 795             txt += "32"
 796         elif (self.hart.misa >> 62) == 2:
 797             txt += "64"
 798         elif (self.hart.misa >> 126) == 3:
 799             txt += "128"
 800         else:
 801             raise TestFailed("Couldn't determine XLEN from $misa (0x%x)" %
 802                     self.hart.misa)
 803
 804         for i in range(26):
 805             if self.hart.misa & (1<<i):
 806                 txt += chr(i + ord('A'))
 807         print txt,
 808
 809 class TestFailed(Exception):
 810     def __init__(self, message):
 811         Exception.__init__(self)
 812         self.message = message
 813
 814 class TestNotApplicable(Exception):
 815     def __init__(self, message):
 816         Exception.__init__(self)
 817         self.message = message
 818
 819 def assertEqual(a, b):
 820     if a != b:
 821         raise TestFailed("%r != %r" % (a, b))
 822
 823 def assertNotEqual(a, b):
 824     if a == b:
 825         raise TestFailed("%r == %r" % (a, b))
 826
 827 def assertIn(a, b):
 828     if a not in b:
 829         raise TestFailed("%r not in %r" % (a, b))
 830
 831 def assertNotIn(a, b):
 832     if a in b:
 833         raise TestFailed("%r in %r" % (a, b))
 834
 835 def assertGreater(a, b):
 836     if not a > b:
 837         raise TestFailed("%r not greater than %r" % (a, b))
 838
 839 def assertLess(a, b):
 840     if not a < b:
 841         raise TestFailed("%r not less than %r" % (a, b))
 842
 843 def assertTrue(a):
 844     if not a:
 845         raise TestFailed("%r is not True" % a)
 846
 847 def assertRegexpMatches(text, regexp):
 848     if not re.search(regexp, text):
 849         raise TestFailed("can't find %r in %r" % (regexp, text))