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