debug/testlib.py

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