Add abort() for benefit of benchmark code
[riscv-tests.git] / debug / testlib.py
index 100b124b8cdcf3738605daa67bb206ee581ec20b..a762174df18240a2309d3f8def20f14ff998d3c5 100644 (file)
@@ -27,9 +27,17 @@ def compile(args, xlen=32): # pylint: disable=redefined-builtin
             cmd.append(found)
         else:
             cmd.append(arg)
-    cmd = " ".join(cmd)
-    result = os.system(cmd)
-    assert result == 0, "%r failed" % cmd
+    process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE)
+    stdout, stderr = process.communicate()
+    if process.returncode:
+        print
+        header("Compile failed")
+        print "+", " ".join(cmd)
+        print stdout,
+        print stderr,
+        header("")
+        raise Exception("Compile failed!")
 
 def unused_port():
     # http://stackoverflow.com/questions/2838244/get-open-tcp-port-in-python/2838309#2838309
@@ -130,11 +138,20 @@ class Openocd(object):
         if debug:
             cmd.append("-d")
 
-        # Assign port
-        self.port = unused_port()
         # This command needs to come before any config scripts on the command
         # line, since they are executed in order.
-        cmd[1:1] = ["--command", "gdb_port %d" % self.port]
+        cmd[1:1] = [
+            # Tell OpenOCD to bind gdb to an unused, ephemeral port.
+            "--command",
+            "gdb_port 0",
+            # Disable tcl and telnet servers, since they are unused and because
+            # the port numbers will conflict if multiple OpenOCD processes are
+            # running on the same server.
+            "--command",
+            "tcl_port disabled",
+            "--command",
+            "telnet_port disabled",
+        ]
 
         logfile = open(Openocd.logname, "w")
         logfile.write("+ %s\n" % " ".join(cmd))
@@ -158,6 +175,37 @@ class Openocd(object):
                 messaged = True
                 print "Waiting for OpenOCD to examine RISCV core..."
 
+        self.port = self._get_gdb_server_port()
+
+    def _get_gdb_server_port(self):
+        """Get port that OpenOCD's gdb server is listening on."""
+        MAX_ATTEMPTS = 50
+        PORT_REGEX = re.compile(r'(?P<port>\d+) \(LISTEN\)')
+        for _ in range(MAX_ATTEMPTS):
+            with open(os.devnull, 'w') as devnull:
+                try:
+                    output = subprocess.check_output([
+                        'lsof',
+                        '-a',  # Take the AND of the following selectors
+                        '-p{}'.format(self.process.pid),  # Filter on PID
+                        '-iTCP',  # Filter only TCP sockets
+                    ], stderr=devnull)
+                except subprocess.CalledProcessError:
+                    output = ""
+            matches = list(PORT_REGEX.finditer(output))
+            matches = [m for m in matches
+                    if m.group('port') not in ('6666', '4444')]
+            if len(matches) > 1:
+                print output
+                raise Exception(
+                    "OpenOCD listening on multiple ports. Cannot uniquely "
+                    "identify gdb server port.")
+            elif matches:
+                [match] = matches
+                return int(match.group('port'))
+            time.sleep(0.1)
+        raise Exception("Timed out waiting for gdb server to obtain port.")
+
     def __del__(self):
         try:
             self.process.kill()
@@ -173,9 +221,23 @@ class OpenocdCli(object):
 
     def command(self, cmd):
         self.child.sendline(cmd)
+        self.child.expect(cmd)
         self.child.expect("\n")
         self.child.expect("> ")
-        return self.child.before.strip()
+        return self.child.before.strip("\t\r\n \0")
+
+    def reg(self, reg=''):
+        output = self.command("reg %s" % reg)
+        matches = re.findall(r"(\w+) \(/\d+\): (0x[0-9A-F]+)", output)
+        values = {r: int(v, 0) for r, v in matches}
+        if reg:
+            return values[reg]
+        return values
+
+    def load_image(self, image):
+        output = self.command("load_image %s" % image)
+        if 'invalid ELF file, only 32bits files are supported' in output:
+            raise TestNotApplicable(output)
 
 class CannotAccess(Exception):
     def __init__(self, address):
@@ -224,6 +286,13 @@ class Gdb(object):
         value = int(output.split(':')[1].strip(), 0)
         return value
 
+    def p_raw(self, obj):
+        output = self.command("p %s" % obj)
+        m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
+        if m:
+            raise CannotAccess(int(m.group(1), 0))
+        return output.split('=')[-1].strip()
+
     def p(self, obj):
         output = self.command("p/x %s" % obj)
         m = re.search("Cannot access memory at address (0x[0-9a-f]+)", output)
@@ -258,23 +327,31 @@ class Gdb(object):
         assert "Hardware assisted breakpoint" in output
         return output
 
-def run_all_tests(module, target, tests, fail_fast):
+def run_all_tests(module, target, parsed):
     good_results = set(('pass', 'not_applicable'))
 
     start = time.time()
 
     results = {}
     count = 0
+
+    global gdb_cmd  # pylint: disable=global-statement
+    gdb_cmd = parsed.gdb
+
+    todo = [("ExamineTarget", ExamineTarget)]
     for name in dir(module):
         definition = getattr(module, name)
         if type(definition) == type and hasattr(definition, 'test') and \
-                (not tests or any(test in name for test in tests)):
-            instance = definition(target)
-            result = instance.run()
-            results.setdefault(result, []).append(name)
-            count += 1
-            if result not in good_results and fail_fast:
-                break
+                (not parsed.test or any(test in name for test in parsed.test)):
+            todo.append((name, definition))
+
+    for name, definition in todo:
+        instance = definition(target)
+        result = instance.run()
+        results.setdefault(result, []).append(name)
+        count += 1
+        if result not in good_results and parsed.fail_fast:
+            break
 
     header("ran %d tests in %.0fs" % (count, time.time() - start), dash=':')
 
@@ -293,16 +370,33 @@ def add_test_run_options(parser):
             help="Exit as soon as any test fails.")
     parser.add_argument("test", nargs='*',
             help="Run only tests that are named here.")
+    parser.add_argument("--gdb",
+            help="The command to use to start gdb.")
 
 def header(title, dash='-'):
-    dashes = dash * (36 - len(title))
-    before = dashes[:len(dashes)/2]
-    after = dashes[len(dashes)/2:]
-    print "%s[ %s ]%s" % (before, title, after)
+    if title:
+        dashes = dash * (36 - len(title))
+        before = dashes[:len(dashes)/2]
+        after = dashes[len(dashes)/2:]
+        print "%s[ %s ]%s" % (before, title, after)
+    else:
+        print dash * 40
+
+def print_log(path):
+    header(path)
+    lines = open(path, "r").readlines()
+    if len(lines) > 1000:
+        for l in lines[:500]:
+            sys.stdout.write(l)
+        print "..."
+        for l in lines[-500:]:
+            sys.stdout.write(l)
+    else:
+        for l in lines:
+            sys.stdout.write(l)
 
 class BaseTest(object):
     compiled = {}
-    logs = []
 
     def __init__(self, target):
         self.target = target
@@ -310,6 +404,7 @@ class BaseTest(object):
         self.target_process = None
         self.binary = None
         self.start = 0
+        self.logs = []
 
     def early_applicable(self):
         """Return a false value if the test has determined it cannot run
@@ -324,18 +419,9 @@ class BaseTest(object):
         compile_args = getattr(self, 'compile_args', None)
         if compile_args:
             if compile_args not in BaseTest.compiled:
-                try:
-                    # pylint: disable=star-args
-                    BaseTest.compiled[compile_args] = \
-                            self.target.compile(*compile_args)
-                except Exception: # pylint: disable=broad-except
-                    print "exception while compiling in %.2fs" % (
-                            time.time() - self.start)
-                    print "=" * 40
-                    header("Traceback")
-                    traceback.print_exc(file=sys.stdout)
-                    print "/" * 40
-                    return "exception"
+                # pylint: disable=star-args
+                BaseTest.compiled[compile_args] = \
+                        self.target.compile(*compile_args)
         self.binary = BaseTest.compiled.get(compile_args)
 
     def classSetup(self):
@@ -372,6 +458,8 @@ class BaseTest(object):
         try:
             self.setup()
             result = self.test()    # pylint: disable=no-member
+        except TestNotApplicable:
+            result = "not_applicable"
         except Exception as e: # pylint: disable=broad-except
             if isinstance(e, TestFailed):
                 result = "fail"
@@ -385,8 +473,7 @@ class BaseTest(object):
             header("Traceback")
             traceback.print_exc(file=sys.stdout)
             for log in self.logs:
-                header(log)
-                print open(log, "r").read()
+                print_log(log)
             print "/" * 40
             return result
 
@@ -398,11 +485,65 @@ class BaseTest(object):
         print "%s in %.2fs" % (result, time.time() - self.start)
         return result
 
+gdb_cmd = None
+class GdbTest(BaseTest):
+    def __init__(self, target):
+        BaseTest.__init__(self, target)
+        self.gdb = None
+
+    def classSetup(self):
+        BaseTest.classSetup(self)
+        self.logs.append("gdb.log")
+
+        if gdb_cmd:
+            self.gdb = Gdb(gdb_cmd)
+        else:
+            self.gdb = Gdb()
+
+        if self.binary:
+            self.gdb.command("file %s" % self.binary)
+        if self.target:
+            self.gdb.command("set arch riscv:rv%d" % self.target.xlen)
+            self.gdb.command("set remotetimeout %d" % self.target.timeout_sec)
+        if self.server.port:
+            self.gdb.command(
+                    "target extended-remote localhost:%d" % self.server.port)
+
+        self.gdb.p("$priv=3")
+
+    def classTeardown(self):
+        del self.gdb
+        BaseTest.classTeardown(self)
+
+class ExamineTarget(GdbTest):
+    def test(self):
+        self.target.misa = self.gdb.p("$misa")
+
+        txt = "RV"
+        if (self.target.misa >> 30) == 1:
+            txt += "32"
+        elif (self.target.misa >> 62) == 2:
+            txt += "64"
+        elif (self.target.misa >> 126) == 3:
+            txt += "128"
+        else:
+            txt += "??"
+
+        for i in range(26):
+            if self.target.misa & (1<<i):
+                txt += chr(i + ord('A'))
+        print txt,
+
 class TestFailed(Exception):
     def __init__(self, message):
         Exception.__init__(self)
         self.message = message
 
+class TestNotApplicable(Exception):
+    def __init__(self, message):
+        Exception.__init__(self)
+        self.message = message
+
 def assertEqual(a, b):
     if a != b:
         raise TestFailed("%r != %r" % (a, b))
@@ -423,6 +564,10 @@ def assertGreater(a, b):
     if not a > b:
         raise TestFailed("%r not greater than %r" % (a, b))
 
+def assertLess(a, b):
+    if not a < b:
+        raise TestFailed("%r not less than %r" % (a, b))
+
 def assertTrue(a):
     if not a:
         raise TestFailed("%r is not True" % a)