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     examine_added = False
 536     for hart in target.harts:
 537         if parsed.misaval:
 538             hart.misa = int(parsed.misaval, 16)
 539             print "Using $misa from command line: 0x%x" % hart.misa
 540         elif hart.misa:
 541             print "Using $misa from hart definition: 0x%x" % hart.misa
 542         elif not examine_added:
 543             todo.append(("ExamineTarget", ExamineTarget, None))
 544             examine_added = True
 545
 546     for name in dir(module):
 547         definition = getattr(module, name)
 548         if type(definition) == type and hasattr(definition, 'test') and \
 549                 (not parsed.test or any(test in name for test in parsed.test)):
 550             todo.append((name, definition, None))
 551
 552     results, count = run_tests(parsed, target, todo)
 553
 554     header("ran %d tests in %.0fs" % (count, time.time() - overall_start),
 555             dash=':')
 556
 557     return print_results(results)
 558
 559 good_results = set(('pass', 'not_applicable'))
 560 def run_tests(parsed, target, todo):
 561     results = {}
 562     count = 0
 563
 564     for name, definition, hart in todo:
 565         log_name = os.path.join(parsed.logs, "%s-%s-%s.log" %
 566                 (time.strftime("%Y%m%d-%H%M%S"), type(target).__name__, name))
 567         log_fd = open(log_name, 'w')
 568         print "[%s] Starting > %s" % (name, log_name)
 569         instance = definition(target, hart)
 570         sys.stdout.flush()
 571         log_fd.write("Test: %s\n" % name)
 572         log_fd.write("Target: %s\n" % type(target).__name__)
 573         start = time.time()
 574         real_stdout = sys.stdout
 575         sys.stdout = log_fd
 576         try:
 577             result = instance.run(real_stdout)
 578             log_fd.write("Result: %s\n" % result)
 579         finally:
 580             sys.stdout = real_stdout
 581             log_fd.write("Time elapsed: %.2fs\n" % (time.time() - start))
 582         print "[%s] %s in %.2fs" % (name, result, time.time() - start)
 583         if result not in good_results and parsed.print_failures:
 584             sys.stdout.write(open(log_name).read())
 585         sys.stdout.flush()
 586         results.setdefault(result, []).append((name, log_name))
 587         count += 1
 588         if result not in good_results and parsed.fail_fast:
 589             break
 590
 591     return results, count
 592
 593 def print_results(results):
 594     result = 0
 595     for key, value in results.iteritems():
 596         print "%d tests returned %s" % (len(value), key)
 597         if key not in good_results:
 598             result = 1
 599             for name, log_name in value:
 600                 print "   %s > %s" % (name, log_name)
 601
 602     return result
 603
 604 def add_test_run_options(parser):
 605     parser.add_argument("--logs", default="logs",
 606             help="Store logs in the specified directory.")
 607     parser.add_argument("--fail-fast", "-f", action="store_true",
 608             help="Exit as soon as any test fails.")
 609     parser.add_argument("--print-failures", action="store_true",
 610             help="When a test fails, print the log file to stdout.")
 611     parser.add_argument("test", nargs='*',
 612             help="Run only tests that are named here.")
 613     parser.add_argument("--gdb",
 614             help="The command to use to start gdb.")
 615     parser.add_argument("--misaval",
 616             help="Don't run ExamineTarget, just assume the misa value which is "
 617             "specified.")
 618
 619 def header(title, dash='-', length=78):
 620     if title:
 621         dashes = dash * (length - 4 - len(title))
 622         before = dashes[:len(dashes)/2]
 623         after = dashes[len(dashes)/2:]
 624         print "%s[ %s ]%s" % (before, title, after)
 625     else:
 626         print dash * length
 627
 628 def print_log(path):
 629     header(path)
 630     for l in open(path, "r"):
 631         sys.stdout.write(l)
 632     print
 633
 634 class BaseTest(object):
 635     compiled = {}
 636
 637     def __init__(self, target, hart=None):
 638         self.target = target
 639         if hart:
 640             self.hart = hart
 641         else:
 642             self.hart = random.choice(target.harts)
 643             self.hart = target.harts[-1]    #<<<
 644         self.server = None
 645         self.target_process = None
 646         self.binary = None
 647         self.start = 0
 648         self.logs = []
 649
 650     def early_applicable(self):
 651         """Return a false value if the test has determined it cannot run
 652         without ever needing to talk to the target or server."""
 653         # pylint: disable=no-self-use
 654         return True
 655
 656     def setup(self):
 657         pass
 658
 659     def compile(self):
 660         compile_args = getattr(self, 'compile_args', None)
 661         if compile_args:
 662             if compile_args not in BaseTest.compiled:
 663                 # pylint: disable=star-args
 664                 BaseTest.compiled[compile_args] = \
 665                         self.target.compile(self.hart, *compile_args)
 666         self.binary = BaseTest.compiled.get(compile_args)
 667
 668     def classSetup(self):
 669         self.compile()
 670         self.target_process = self.target.create()
 671         if self.target_process:
 672             self.logs.append(self.target_process.logname)
 673         try:
 674             self.server = self.target.server()
 675             self.logs.append(self.server.logname)
 676         except Exception:
 677             for log in self.logs:
 678                 print_log(log)
 679             raise
 680
 681     def classTeardown(self):
 682         del self.server
 683         del self.target_process
 684
 685     def postMortem(self):
 686         pass
 687
 688     def run(self, real_stdout):
 689         """
 690         If compile_args is set, compile a program and set self.binary.
 691
 692         Call setup().
 693
 694         Then call test() and return the result, displaying relevant information
 695         if an exception is raised.
 696         """
 697
 698         sys.stdout.flush()
 699
 700         if not self.early_applicable():
 701             return "not_applicable"
 702
 703         self.start = time.time()
 704
 705         try:
 706             self.classSetup()
 707             real_stdout.write("[%s] Temporary logs: %s\n" % (
 708                 type(self).__name__, ", ".join(self.logs)))
 709             self.setup()
 710             result = self.test()    # pylint: disable=no-member
 711         except TestNotApplicable:
 712             result = "not_applicable"
 713         except Exception as e: # pylint: disable=broad-except
 714             if isinstance(e, TestFailed):
 715                 result = "fail"
 716             else:
 717                 result = "exception"
 718             if isinstance(e, TestFailed):
 719                 header("Message")
 720                 print e.message
 721             header("Traceback")
 722             traceback.print_exc(file=sys.stdout)
 723             try:
 724                 self.postMortem()
 725             except Exception as e:  # pylint: disable=broad-except
 726                 header("postMortem Exception")
 727                 print e
 728                 traceback.print_exc(file=sys.stdout)
 729             return result
 730
 731         finally:
 732             for log in self.logs:
 733                 print_log(log)
 734             header("End of logs")
 735             self.classTeardown()
 736
 737         if not result:
 738             result = 'pass'
 739         return result
 740
 741 gdb_cmd = None
 742 class GdbTest(BaseTest):
 743     def __init__(self, target, hart=None):
 744         BaseTest.__init__(self, target, hart=hart)
 745         self.gdb = None
 746
 747     def classSetup(self):
 748         BaseTest.classSetup(self)
 749
 750         if gdb_cmd:
 751             self.gdb = Gdb(self.server.gdb_ports, gdb_cmd, binary=self.binary)
 752         else:
 753             self.gdb = Gdb(self.server.gdb_ports, binary=self.binary)
 754
 755         self.logs += self.gdb.lognames()
 756
 757         if self.target:
 758             self.gdb.global_command("set arch riscv:rv%d" % self.hart.xlen)
 759             self.gdb.global_command("set remotetimeout %d" %
 760                     self.target.timeout_sec)
 761
 762         for cmd in self.target.gdb_setup:
 763             self.gdb.command(cmd)
 764
 765         self.gdb.select_hart(self.hart)
 766
 767         # FIXME: OpenOCD doesn't handle PRIV now
 768         #self.gdb.p("$priv=3")
 769
 770     def postMortem(self):
 771         if not self.gdb:
 772             return
 773         self.gdb.interrupt()
 774         self.gdb.command("info registers all", timeout=10)
 775
 776     def classTeardown(self):
 777         del self.gdb
 778         BaseTest.classTeardown(self)
 779
 780 class GdbSingleHartTest(GdbTest):
 781     def classSetup(self):
 782         GdbTest.classSetup(self)
 783
 784         for hart in self.target.harts:
 785             # Park all harts that we're not using in a safe place.
 786             if hart != self.hart:
 787                 self.gdb.select_hart(hart)
 788                 self.gdb.p("$pc=loop_forever")
 789         self.gdb.select_hart(self.hart)
 790
 791 class ExamineTarget(GdbTest):
 792     def test(self):
 793         for hart in self.target.harts:
 794             self.gdb.select_hart(hart)
 795
 796             hart.misa = self.gdb.p("$misa")
 797
 798             txt = "RV"
 799             if (hart.misa >> 30) == 1:
 800                 txt += "32"
 801             elif (hart.misa >> 62) == 2:
 802                 txt += "64"
 803             elif (hart.misa >> 126) == 3:
 804                 txt += "128"
 805             else:
 806                 raise TestFailed("Couldn't determine XLEN from $misa (0x%x)" %
 807                         self.hart.misa)
 808
 809             for i in range(26):
 810                 if hart.misa & (1<<i):
 811                     txt += chr(i + ord('A'))
 812             print txt,
 813
 814 class TestFailed(Exception):
 815     def __init__(self, message):
 816         Exception.__init__(self)
 817         self.message = message
 818
 819 class TestNotApplicable(Exception):
 820     def __init__(self, message):
 821         Exception.__init__(self)
 822         self.message = message
 823
 824 def assertEqual(a, b):
 825     if a != b:
 826         raise TestFailed("%r != %r" % (a, b))
 827
 828 def assertNotEqual(a, b):
 829     if a == b:
 830         raise TestFailed("%r == %r" % (a, b))
 831
 832 def assertIn(a, b):
 833     if a not in b:
 834         raise TestFailed("%r not in %r" % (a, b))
 835
 836 def assertNotIn(a, b):
 837     if a in b:
 838         raise TestFailed("%r in %r" % (a, b))
 839
 840 def assertGreater(a, b):
 841     if not a > b:
 842         raise TestFailed("%r not greater than %r" % (a, b))
 843
 844 def assertLess(a, b):
 845     if not a < b:
 846         raise TestFailed("%r not less than %r" % (a, b))
 847
 848 def assertTrue(a):
 849     if not a:
 850         raise TestFailed("%r is not True" % a)
 851
 852 def assertRegexpMatches(text, regexp):
 853     if not re.search(regexp, text):
 854         raise TestFailed("can't find %r in %r" % (regexp, text))