debug/testlib.py

   1 import os.path
   2 import re
   3 import shlex
   4 import subprocess
   5 import sys
   6 import time
   7 import traceback
   8
   9 import pexpect
  10
  11 # Note that gdb comes with its own testsuite. I was unable to figure out how to
  12 # run that testsuite against the spike simulator.
  13
  14 def find_file(path):
  15     for directory in (os.getcwd(), os.path.dirname(__file__)):
  16         fullpath = os.path.join(directory, path)
  17         if os.path.exists(fullpath):
  18             return fullpath
  19     return None
  20
  21 def compile(args, xlen=32): # pylint: disable=redefined-builtin
  22     cc = os.path.expandvars("$RISCV/bin/riscv64-unknown-elf-gcc")
  23     cmd = [cc, "-g"]
  24     if xlen == 32:
  25         cmd.append("-march=rv32imac")
  26         cmd.append("-mabi=ilp32")
  27     else:
  28         cmd.append("-march=rv64imac")
  29         cmd.append("-mabi=lp64")
  30     for arg in args:
  31         found = find_file(arg)
  32         if found:
  33             cmd.append(found)
  34         else:
  35             cmd.append(arg)
  36     process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
  37                                stderr=subprocess.PIPE)
  38     stdout, stderr = process.communicate()
  39     if process.returncode:
  40         print
  41         header("Compile failed")
  42         print "+", " ".join(cmd)
  43         print stdout,
  44         print stderr,
  45         header("")
  46         raise Exception("Compile failed!")
  47
  48 def unused_port():
  49     # http://stackoverflow.com/questions/2838244/get-open-tcp-port-in-python/2838309#2838309
  50     import socket
  51     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  52     s.bind(("", 0))
  53     port = s.getsockname()[1]
  54     s.close()
  55     return port
  56
  57 class Spike(object):
  58     logname = "spike-%d.log" % os.getpid()
  59
  60     def __init__(self, target, halted=False, timeout=None, with_jtag_gdb=True):
  61         """Launch spike. Return tuple of its process and the port it's running
  62         on."""
  63         if target.sim_cmd:
  64             cmd = shlex.split(target.sim_cmd)
  65         else:
  66             spike = os.path.expandvars("$RISCV/bin/spike")
  67             cmd = [spike]
  68         if target.xlen == 32:
  69             cmd += ["--isa", "RV32G"]
  70         else:
  71             cmd += ["--isa", "RV64G"]
  72         cmd += ["-m0x%x:0x%x" % (target.ram, target.ram_size)]
  73
  74         if timeout:
  75             cmd = ["timeout", str(timeout)] + cmd
  76
  77         if halted:
  78             cmd.append('-H')
  79         if with_jtag_gdb:
  80             cmd += ['--rbb-port', '0']
  81             os.environ['REMOTE_BITBANG_HOST'] = 'localhost'
  82         self.infinite_loop = target.compile(
  83                 "programs/checksum.c", "programs/tiny-malloc.c",
  84                 "programs/infinite_loop.c", "-DDEFINE_MALLOC", "-DDEFINE_FREE")
  85         cmd.append(self.infinite_loop)
  86         logfile = open(self.logname, "w")
  87         logfile.write("+ %s\n" % " ".join(cmd))
  88         logfile.flush()
  89         self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
  90                 stdout=logfile, stderr=logfile)
  91
  92         if with_jtag_gdb:
  93             self.port = None
  94             for _ in range(30):
  95                 m = re.search(r"Listening for remote bitbang connection on "
  96                         r"port (\d+).", open(self.logname).read())
  97                 if m:
  98                     self.port = int(m.group(1))
  99                     os.environ['REMOTE_BITBANG_PORT'] = m.group(1)
 100                     break
 101                 time.sleep(0.11)
 102             assert self.port, "Didn't get spike message about bitbang " \
 103                     "connection"
 104
 105     def __del__(self):
 106         try:
 107             self.process.kill()
 108             self.process.wait()
 109         except OSError:
 110             pass
 111
 112     def wait(self, *args, **kwargs):
 113         return self.process.wait(*args, **kwargs)
 114
 115 class VcsSim(object):
 116     def __init__(self, sim_cmd=None, debug=False):
 117         if sim_cmd:
 118             cmd = shlex.split(sim_cmd)
 119         else:
 120             cmd = ["simv"]
 121         cmd += ["+jtag_vpi_enable"]
 122         if debug:
 123             cmd[0] = cmd[0] + "-debug"
 124             cmd += ["+vcdplusfile=output/gdbserver.vpd"]
 125         logfile = open("simv.log", "w")
 126         logfile.write("+ %s\n" % " ".join(cmd))
 127         logfile.flush()
 128         listenfile = open("simv.log", "r")
 129         listenfile.seek(0, 2)
 130         self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
 131                 stdout=logfile, stderr=logfile)
 132         done = False
 133         while not done:
 134             # Fail if VCS exits early
 135             exit_code = self.process.poll()
 136             if exit_code is not None:
 137                 raise RuntimeError('VCS simulator exited early with status %d'
 138                                    % exit_code)
 139
 140             line = listenfile.readline()
 141             if not line:
 142                 time.sleep(1)
 143             match = re.match(r"^Listening on port (\d+)$", line)
 144             if match:
 145                 done = True
 146                 self.port = int(match.group(1))
 147                 os.environ['JTAG_VPI_PORT'] = str(self.port)
 148
 149     def __del__(self):
 150         try:
 151             self.process.kill()
 152             self.process.wait()
 153         except OSError:
 154             pass
 155
 156 class Openocd(object):
 157     logname = "openocd-%d.log" % os.getpid()
 158
 159     def __init__(self, server_cmd=None, config=None, debug=False):
 160         if server_cmd:
 161             cmd = shlex.split(server_cmd)
 162         else:
 163             openocd = os.path.expandvars("$RISCV/bin/openocd")
 164             cmd = [openocd]
 165             if debug:
 166                 cmd.append("-d")
 167
 168         # This command needs to come before any config scripts on the command
 169         # line, since they are executed in order.
 170         cmd += [
 171             # Tell OpenOCD to bind gdb to an unused, ephemeral port.
 172             "--command",
 173             "gdb_port 0",
 174             # Disable tcl and telnet servers, since they are unused and because
 175             # the port numbers will conflict if multiple OpenOCD processes are
 176             # running on the same server.
 177             "--command",
 178             "tcl_port disabled",
 179             "--command",
 180             "telnet_port disabled",
 181         ]
 182
 183         if config:
 184             f = find_file(config)
 185             if f is None:
 186                 print "Unable to read file " + config
 187                 exit(1)
 188
 189             cmd += ["-f", f]
 190         if debug:
 191             cmd.append("-d")
 192
 193         logfile = open(Openocd.logname, "w")
 194         logfile.write("+ %s\n" % " ".join(cmd))
 195         logfile.flush()
 196         self.process = subprocess.Popen(cmd, stdin=subprocess.PIPE,
 197                 stdout=logfile, stderr=logfile)
 198
 199         # Wait for OpenOCD to have made it through riscv_examine(). When using
 200         # OpenOCD to communicate with a simulator this may take a long time,
 201         # and gdb will time out when trying to connect if we attempt too early.
 202         start = time.time()
 203         messaged = False
 204         while True:
 205             log = open(Openocd.logname).read()
 206             if "Ready for Remote Connections" in log:
 207                 break
 208             if not self.process.poll() is None:
 209                 raise Exception(
 210                         "OpenOCD exited before completing riscv_examine()")
 211             if not messaged and time.time() - start > 1:
 212                 messaged = True
 213                 print "Waiting for OpenOCD to examine RISCV core..."
 214             if time.time() - start > 60:
 215                 raise Exception("ERROR: Timed out waiting for OpenOCD to "
 216                         "examine RISCV core")
 217
 218         self.port = self._get_gdb_server_port()
 219
 220     def _get_gdb_server_port(self):
 221         """Get port that OpenOCD's gdb server is listening on."""
 222         MAX_ATTEMPTS = 50
 223         PORT_REGEX = re.compile(r'(?P<port>\d+) \(LISTEN\)')
 224         for _ in range(MAX_ATTEMPTS):
 225             with open(os.devnull, 'w') as devnull:
 226                 try:
 227                     output = subprocess.check_output([
 228                         'lsof',
 229                         '-a',  # Take the AND of the following selectors
 230                         '-p{}'.format(self.process.pid),  # Filter on PID
 231                         '-iTCP',  # Filter only TCP sockets
 232                     ], stderr=devnull)
 233                 except subprocess.CalledProcessError:
 234                     output = ""
 235             matches = list(PORT_REGEX.finditer(output))
 236             matches = [m for m in matches
 237                     if m.group('port') not in ('6666', '4444')]
 238             if len(matches) > 1:
 239                 print output
 240                 raise Exception(
 241                     "OpenOCD listening on multiple ports. Cannot uniquely "
 242                     "identify gdb server port.")
 243             elif matches:
 244                 [match] = matches
 245                 return int(match.group('port'))
 246             time.sleep(1)
 247         raise Exception("Timed out waiting for gdb server to obtain port.")
 248
 249     def __del__(self):
 250         try:
 251             self.process.kill()
 252             self.process.wait()
 253         except OSError:
 254             pass
 255
 256 class OpenocdCli(object):
 257     def __init__(self, port=4444):
 258         self.child = pexpect.spawn(
 259                 "sh -c 'telnet localhost %d | tee openocd-cli.log'" % port)
 260         self.child.expect("> ")
 261
 262     def command(self, cmd):
 263         self.child.sendline(cmd)
 264         self.child.expect(cmd)
 265         self.child.expect("\n")
 266         self.child.expect("> ")
 267         return self.child.before.strip("\t\r\n \0")
 268
 269     def reg(self, reg=''):
 270         output = self.command("reg %s" % reg)
 271         matches = re.findall(r"(\w+) \(/\d+\): (0x[0-9A-F]+)", output)
 272         values = {r: int(v, 0) for r, v in matches}
 273         if reg:
 274             return values[reg]
 275         return values
 276
 277     def load_image(self, image):
 278         output = self.command("load_image %s" % image)
 279         if 'invalid ELF file, only 32bits files are supported' in output:
 280             raise TestNotApplicable(output)
 281
 282 class CannotAccess(Exception):
 283     def __init__(self, address):
 284         Exception.__init__(self)
 285         self.address = address
 286
 287 class Gdb(object):
 288     def __init__(self,
 289             cmd=os.path.expandvars("$RISCV/bin/riscv64-unknown-elf-gdb")):
 290         self.child = pexpect.spawn(cmd)
 291         self.child.logfile = open("gdb.log", "w")
 292         self.child.logfile.write("+ %s\n" % cmd)
 293         self.wait()
 294         self.command("set confirm off")
 295         self.command("set width 0")
 296         self.command("set height 0")
 297         # Force consistency.
 298         self.command("set print entry-values no")
 299
 300     def wait(self):
 301         """Wait for prompt."""
 302         self.child.expect(r"\(gdb\)")
 303
 304     def command(self, command, timeout=6000):
 305         self.child.sendline(command)
 306         self.child.expect("\n", timeout=timeout)
 307         self.child.expect(r"\(gdb\)", timeout=timeout)
 308         return self.child.before.strip()
 309
 310     def c(self, wait=True, timeout=-1):
 311         if wait:
 312             output = self.command("c", timeout=timeout)
 313             assert "Continuing" in output
 314             return output
 315         else:
 316             self.child.sendline("c")
 317             self.child.expect("Continuing")
 318
 319     def interrupt(self):
 320         self.child.send("\003")
 321         self.child.expect(r"\(gdb\)", timeout=6000)
 322         return self.child.before.strip()
 323
 324     def x(self, address, size='w'):
 325         output = self.command("x/%s %s" % (size, address))
 326         value = int(output.split(':')[1].strip(), 0)
 327         return value
 328
 329     def p_raw(self, obj):
 330         output = self.command("p %s" % obj)
 331         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 332         if m:
 333             raise CannotAccess(int(m.group(1), 0))
 334         return output.split('=')[-1].strip()
 335
 336     def p(self, obj):
 337         output = self.command("p/x %s" % obj)
 338         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
 339         if m:
 340             raise CannotAccess(int(m.group(1), 0))
 341         value = int(output.split('=')[-1].strip(), 0)
 342         return value
 343
 344     def p_string(self, obj):
 345         output = self.command("p %s" % obj)
 346         value = shlex.split(output.split('=')[-1].strip())[1]
 347         return value
 348
 349     def stepi(self):
 350         output = self.command("stepi")
 351         return output
 352
 353     def load(self):
 354         output = self.command("load", timeout=6000)
 355         assert "failed" not in  output
 356         assert "Transfer rate" in output
 357
 358     def b(self, location):
 359         output = self.command("b %s" % location)
 360         assert "not defined" not in output
 361         assert "Breakpoint" in output
 362         return output
 363
 364     def hbreak(self, location):
 365         output = self.command("hbreak %s" % location)
 366         assert "not defined" not in output
 367         assert "Hardware assisted breakpoint" in output
 368         return output
 369
 370 def run_all_tests(module, target, parsed):
 371     good_results = set(('pass', 'not_applicable'))
 372
 373     start = time.time()
 374
 375     results = {}
 376     count = 0
 377
 378     global gdb_cmd  # pylint: disable=global-statement
 379     gdb_cmd = parsed.gdb
 380
 381     todo = []
 382     if parsed.misaval:
 383         target.misa = int(parsed.misaval, 16)
 384         print "Assuming $MISA value of 0x%x. Skipping ExamineTarget." % \
 385                 target.misa
 386     else:
 387         todo.append(("ExamineTarget", ExamineTarget))
 388
 389     for name in dir(module):
 390         definition = getattr(module, name)
 391         if type(definition) == type and hasattr(definition, 'test') and \
 392                 (not parsed.test or any(test in name for test in parsed.test)):
 393             todo.append((name, definition))
 394
 395     for name, definition in todo:
 396         instance = definition(target)
 397         result = instance.run()
 398         results.setdefault(result, []).append(name)
 399         count += 1
 400         if result not in good_results and parsed.fail_fast:
 401             break
 402
 403     header("ran %d tests in %.0fs" % (count, time.time() - start), dash=':')
 404
 405     result = 0
 406     for key, value in results.iteritems():
 407         print "%d tests returned %s" % (len(value), key)
 408         if key not in good_results:
 409             result = 1
 410             for test in value:
 411                 print "   ", test
 412
 413     return result
 414
 415 def add_test_run_options(parser):
 416
 417     parser.add_argument("--fail-fast", "-f", action="store_true",
 418             help="Exit as soon as any test fails.")
 419     parser.add_argument("test", nargs='*',
 420             help="Run only tests that are named here.")
 421     parser.add_argument("--gdb",
 422             help="The command to use to start gdb.")
 423     parser.add_argument("--misaval",
 424             help="Don't run ExamineTarget, just assume the misa value which is "
 425             "specified.")
 426
 427 def header(title, dash='-'):
 428     if title:
 429         dashes = dash * (36 - len(title))
 430         before = dashes[:len(dashes)/2]
 431         after = dashes[len(dashes)/2:]
 432         print "%s[ %s ]%s" % (before, title, after)
 433     else:
 434         print dash * 40
 435
 436 def print_log(path):
 437     header(path)
 438     lines = open(path, "r").readlines()
 439     if len(lines) > 1000:
 440         for l in lines[:500]:
 441             sys.stdout.write(l)
 442         print "..."
 443         for l in lines[-500:]:
 444             sys.stdout.write(l)
 445     else:
 446         for l in lines:
 447             sys.stdout.write(l)
 448
 449 class BaseTest(object):
 450     compiled = {}
 451
 452     def __init__(self, target):
 453         self.target = target
 454         self.server = None
 455         self.target_process = None
 456         self.binary = None
 457         self.start = 0
 458         self.logs = []
 459
 460     def early_applicable(self):
 461         """Return a false value if the test has determined it cannot run
 462         without ever needing to talk to the target or server."""
 463         # pylint: disable=no-self-use
 464         return True
 465
 466     def setup(self):
 467         pass
 468
 469     def compile(self):
 470         compile_args = getattr(self, 'compile_args', None)
 471         if compile_args:
 472             if compile_args not in BaseTest.compiled:
 473                 # pylint: disable=star-args
 474                 BaseTest.compiled[compile_args] = \
 475                         self.target.compile(*compile_args)
 476         self.binary = BaseTest.compiled.get(compile_args)
 477
 478     def classSetup(self):
 479         self.compile()
 480         self.target_process = self.target.target()
 481         self.server = self.target.server()
 482         self.logs.append(self.server.logname)
 483
 484     def classTeardown(self):
 485         del self.server
 486         del self.target_process
 487
 488     def run(self):
 489         """
 490         If compile_args is set, compile a program and set self.binary.
 491
 492         Call setup().
 493
 494         Then call test() and return the result, displaying relevant information
 495         if an exception is raised.
 496         """
 497
 498         print "Running", type(self).__name__, "...",
 499         sys.stdout.flush()
 500
 501         if not self.early_applicable():
 502             print "not_applicable"
 503             return "not_applicable"
 504
 505         self.start = time.time()
 506
 507         self.classSetup()
 508
 509         try:
 510             self.setup()
 511             result = self.test()    # pylint: disable=no-member
 512         except TestNotApplicable:
 513             result = "not_applicable"
 514         except Exception as e: # pylint: disable=broad-except
 515             if isinstance(e, TestFailed):
 516                 result = "fail"
 517             else:
 518                 result = "exception"
 519             print "%s in %.2fs" % (result, time.time() - self.start)
 520             print "=" * 40
 521             if isinstance(e, TestFailed):
 522                 header("Message")
 523                 print e.message
 524             header("Traceback")
 525             traceback.print_exc(file=sys.stdout)
 526             for log in self.logs:
 527                 print_log(log)
 528             print "/" * 40
 529             return result
 530
 531         finally:
 532             self.classTeardown()
 533
 534         if not result:
 535             result = 'pass'
 536         print "%s in %.2fs" % (result, time.time() - self.start)
 537         return result
 538
 539 gdb_cmd = None
 540 class GdbTest(BaseTest):
 541     def __init__(self, target):
 542         BaseTest.__init__(self, target)
 543         self.gdb = None
 544
 545     def classSetup(self):
 546         BaseTest.classSetup(self)
 547         self.logs.append("gdb.log")
 548
 549         if gdb_cmd:
 550             self.gdb = Gdb(gdb_cmd)
 551         else:
 552             self.gdb = Gdb()
 553
 554         if self.binary:
 555             self.gdb.command("file %s" % self.binary)
 556         if self.target:
 557             self.gdb.command("set arch riscv:rv%d" % self.target.xlen)
 558             self.gdb.command("set remotetimeout %d" % self.target.timeout_sec)
 559         if self.server.port:
 560             self.gdb.command(
 561                     "target extended-remote localhost:%d" % self.server.port)
 562
 563         # FIXME: OpenOCD doesn't handle PRIV now
 564         #self.gdb.p("$priv=3")
 565
 566     def classTeardown(self):
 567         del self.gdb
 568         BaseTest.classTeardown(self)
 569
 570 class ExamineTarget(GdbTest):
 571     def test(self):
 572         self.target.misa = self.gdb.p("$misa")
 573
 574         txt = "RV"
 575         if (self.target.misa >> 30) == 1:
 576             txt += "32"
 577         elif (self.target.misa >> 62) == 2:
 578             txt += "64"
 579         elif (self.target.misa >> 126) == 3:
 580             txt += "128"
 581         else:
 582             raise TestFailed("Couldn't determine XLEN from $misa (0x%x)" %
 583                     self.target.misa)
 584
 585         for i in range(26):
 586             if self.target.misa & (1<<i):
 587                 txt += chr(i + ord('A'))
 588         print txt,
 589
 590 class TestFailed(Exception):
 591     def __init__(self, message):
 592         Exception.__init__(self)
 593         self.message = message
 594
 595 class TestNotApplicable(Exception):
 596     def __init__(self, message):
 597         Exception.__init__(self)
 598         self.message = message
 599
 600 def assertEqual(a, b):
 601     if a != b:
 602         raise TestFailed("%r != %r" % (a, b))
 603
 604 def assertNotEqual(a, b):
 605     if a == b:
 606         raise TestFailed("%r == %r" % (a, b))
 607
 608 def assertIn(a, b):
 609     if a not in b:
 610         raise TestFailed("%r not in %r" % (a, b))
 611
 612 def assertNotIn(a, b):
 613     if a in b:
 614         raise TestFailed("%r in %r" % (a, b))
 615
 616 def assertGreater(a, b):
 617     if not a > b:
 618         raise TestFailed("%r not greater than %r" % (a, b))
 619
 620 def assertLess(a, b):
 621     if not a < b:
 622         raise TestFailed("%r not less than %r" % (a, b))
 623
 624 def assertTrue(a):
 625     if not a:
 626         raise TestFailed("%r is not True" % a)
 627
 628 def assertRegexpMatches(text, regexp):
 629     if not re.search(regexp, text):
 630         raise TestFailed("can't find %r in %r" % (regexp, text))