test/hftest/hftest.py

#!/usr/bin/env python
#
# Copyright 2018 The Hafnium Authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Run tests.

Runs tests on QEMU.
"""

from __future__ import print_function

import xml.etree.ElementTree as ET

import argparse
import datetime
import json
import os
import re
import subprocess
import sys

HFTEST_LOG_PREFIX = "[hftest] "
HFTEST_LOG_FAILURE_PREFIX = "Failure:"
HFTEST_LOG_FINISHED = "FINISHED"

HF_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
DTC_SCRIPT = os.path.join(HF_ROOT, "build", "image", "dtc.py")

def log_timeout_returncode(f, returncode):
    if returncode == 0:
        return
    elif returncode == 124:
        f.write("\r\n{}{} timed out\r\n".format(HFTEST_LOG_PREFIX,
                                                HFTEST_LOG_FAILURE_PREFIX))
    else:
        f.write("\r\n{}{} process return code {}\r\n".format(
            HFTEST_LOG_PREFIX, HFTEST_LOG_FAILURE_PREFIX, returncode))

def qemu(image, initrd, args, log):
    qemu_args = [
        "timeout", "--foreground", "10s",
        "./prebuilts/linux-x64/qemu/qemu-system-aarch64", "-M", "virt,gic_version=3",
        "-cpu", "cortex-a57", "-smp", "4", "-m", "64M", "-machine", "virtualization=true",
        "-nographic", "-nodefaults", "-serial", "stdio", "-kernel", image,
    ]
    if initrd:
        qemu_args += ["-initrd", initrd]
    if args:
        qemu_args += ["-append", args]
    # Save the log to a file.
    with open(log, "w") as f:
        f.write("$ {}\r\n".format(" ".join(qemu_args)))
        f.flush()
        try:
            subprocess.check_call(qemu_args, stdout=f, stderr=f)
        except subprocess.CalledProcessError as e:
            log_timeout_returncode(f, e.returncode)
    # Return that log for processing.
    with open(log, "r") as f:
        return f.read()


def fvp(image, initrd, args, log):
    uart0_log = log + ".uart0"
    uart1_log = log + ".uart1"
    fdt = log + ".dtb"
    dtc_args = [ DTC_SCRIPT, "-o", fdt ]
    fvp_args = [
        "timeout", "--foreground", "40s",
        "../fvp/Base_RevC_AEMv8A_pkg/models/Linux64_GCC-4.9/FVP_Base_RevC-2xAEMv8A",
        "-C", "pctl.startup=0.0.0.0",
        "-C", "bp.secure_memory=0",
        "-C", "cluster0.NUM_CORES=4",
        "-C", "cluster1.NUM_CORES=4",
        "-C", "cache_state_modelled=0",
        "-C", "bp.vis.disable_visualisation=true",
        "-C", "bp.vis.rate_limit-enable=false",
        "-C", "bp.terminal_0.start_telnet=false",
        "-C", "bp.terminal_1.start_telnet=false",
        "-C", "bp.terminal_2.start_telnet=false",
        "-C", "bp.terminal_3.start_telnet=false",
        "-C", "bp.pl011_uart0.untimed_fifos=1",
        "-C", "bp.pl011_uart0.unbuffered_output=1",
        "-C", "bp.pl011_uart0.out_file=" + uart0_log,
        "-C", "bp.pl011_uart1.out_file=" + uart1_log,
        "-C", "cluster0.cpu0.RVBAR=0x04020000",
        "-C", "cluster0.cpu1.RVBAR=0x04020000",
        "-C", "cluster0.cpu2.RVBAR=0x04020000",
        "-C", "cluster0.cpu3.RVBAR=0x04020000",
        "-C", "cluster1.cpu0.RVBAR=0x04020000",
        "-C", "cluster1.cpu1.RVBAR=0x04020000",
        "-C", "cluster1.cpu2.RVBAR=0x04020000",
        "-C", "cluster1.cpu3.RVBAR=0x04020000",
        "--data", "cluster0.cpu0=prebuilts/linux-aarch64/arm-trusted-firmware/bl31.bin@0x04020000",
        "--data", "cluster0.cpu0=" + fdt + "@0x82000000",
        "--data", "cluster0.cpu0=" + image + "@0x80000000",
        "-C", "bp.ve_sysregs.mmbSiteDefault=0",
        "-C", "bp.ve_sysregs.exit_on_shutdown=1",
    ]
    initrd_start = 0x84000000
    initrd_end = 0x85000000  # Default value
    if initrd:
        fvp_args += ["--data", "cluster0.cpu0={}@{}".format(initrd, hex(initrd_start))]
        initrd_end = initrd_start + os.path.getsize(initrd)


    with open(log, "w") as f:
        f.write("$ {}\r\n".format(" ".join(dtc_args)))
        f.flush()
        dtc = subprocess.Popen(
            dtc_args, stdout=f, stderr=f, stdin=subprocess.PIPE)
        with open(
                "prebuilts/linux-aarch64/arm-trusted-firmware/fvp-base-gicv3-psci-1t.dts",
                "r") as base_dts:
            dtc.stdin.write(base_dts.read())
        dtc.stdin.write("/ {\n")
        dtc.stdin.write("        chosen {\n")
        dtc.stdin.write("                bootargs = \"" + args + "\";\n")
        dtc.stdin.write("                stdout-path = \"serial0:115200n8\";\n")
        dtc.stdin.write("                linux,initrd-start = <{}>;\n".format(initrd_start))
        dtc.stdin.write("                linux,initrd-end = <{}>;\n".format(initrd_end))
        dtc.stdin.write("        };\n")
        dtc.stdin.write("};\n")
        dtc.stdin.close()
        dtc.wait()

        f.write("$ {}\r\n".format(" ".join(fvp_args)))
        f.flush()
        returncode = subprocess.call(fvp_args, stdout=f, stderr=f)
        with open(uart0_log, "r") as g:
            f.write(g.read())
        log_timeout_returncode(f, returncode)

    with open(log, "r") as f:
        return f.read()


def emulator(use_fvp, image, initrd, args, log):
    if use_fvp:
        return fvp(image, initrd, args, log)
    else:
        return qemu(image, initrd, args, log)


def ensure_dir(path):
    try:
        os.makedirs(path)
    except OSError:
        if not os.path.isdir(path):
            raise


def hftest_lines(raw):
    lines = []
    for line in raw.splitlines():
        if line.startswith("VM "):
            line = line[len("VM 0: "):]
        if line.startswith(HFTEST_LOG_PREFIX):
            lines.append(line[len(HFTEST_LOG_PREFIX):])
    return lines


def Main():
    parser = argparse.ArgumentParser()
    parser.add_argument("image")
    parser.add_argument("--out", required=True)
    parser.add_argument("--log", required=True)
    parser.add_argument("--out_initrd")
    parser.add_argument("--initrd")
    parser.add_argument("--suite")
    parser.add_argument("--test")
    parser.add_argument("--vm_args")
    parser.add_argument("--fvp", type=bool)
    args = parser.parse_args()
    # Resolve some paths.
    image = os.path.join(args.out, args.image + ".bin")
    initrd = None
    suite = args.image
    if args.initrd:
        initrd = os.path.join(args.out_initrd, "obj", args.initrd, "initrd.img")
        suite += "_" + args.initrd
    vm_args = args.vm_args or ""
    log = os.path.join(args.log, suite)
    ensure_dir(log)
    print("Logs saved under", log)
    log_file = os.path.join(log, "sponge_log.log")
    with open(log_file, "w") as full_log:
        # Query the tests in the image.
        out = emulator(args.fvp, image, initrd, vm_args + " json",
                       os.path.join(log, "json.log"))
        full_log.write(out)
        full_log.write("\r\n\r\n")
        hftest_json = "\n".join(hftest_lines(out))
        try:
            tests = json.loads(hftest_json)
        except ValueError:
            print(hftest_json)
            return 2
        # Run the selected tests.
        tests_run = 0
        failures = 0
        suite_re = re.compile(args.suite or ".*")
        test_re = re.compile(args.test or ".*")
        suites_xml = ET.Element("testsuites")
        suites_xml.set("name", suite)
        suites_xml.set(
            "timestamp",
            datetime.datetime.now().replace(microsecond=0).isoformat())
        for suite in tests["suites"]:
            if not suite_re.match(suite["name"]):
                continue
            tests_run_from_suite = 0
            failures_from_suite = 0
            suite_xml = ET.SubElement(suites_xml, "testsuite")
            suite_xml.set("name", suite["name"])
            for test in suite["tests"]:
                if not test_re.match(test):
                    continue
                test_xml = ET.SubElement(suite_xml, "testcase")
                test_xml.set("name", test)
                test_xml.set("classname", suite['name'])
                test_xml.set("status", "run")
                tests_run_from_suite += 1
                if tests_run_from_suite == 1:
                    print("    SUITE", suite["name"])
                print("      RUN", test)
                test_log = os.path.join(log,
                                        suite["name"] + "." + test + ".log")
                out = emulator(
                    args.fvp, image, initrd,
                    vm_args + " run {} {}".format(suite["name"], test),
                    test_log)
                full_log.write(out)
                full_log.write("\r\n\r\n")
                hftest_out = hftest_lines(out)
                if len(
                        hftest_out
                ) > 0 and hftest_out[-1] == HFTEST_LOG_FINISHED and not any(
                        l.startswith(HFTEST_LOG_FAILURE_PREFIX)
                        for l in hftest_out):
                    print("        PASS")
                else:
                    failures_from_suite += 1
                    failure_xml = ET.SubElement(test_xml, "failure")
                    # TODO: set a meaningful message and put log in CDATA
                    failure_xml.set("message", "Test failed")
                    print("[x]     FAIL --", test_log)
            tests_run += tests_run_from_suite
            failures += failures_from_suite
            suite_xml.set("tests", str(tests_run_from_suite))
            suite_xml.set("failures", str(failures_from_suite))
    suites_xml.set("tests", str(tests_run))
    suites_xml.set("failures", str(failures))
    with open(os.path.join(log, "sponge_log.xml"), "w") as f:
        ET.ElementTree(suites_xml).write(
            f, encoding='utf-8', xml_declaration=True)
    # If none were run, this is probably a mistake.
    if tests_run == 0:
        print("Error: no tests match")
        return 10
    # Exit with 0 on success and 1 if any test failed.
    if failures:
        print("[x] FAIL:", failures, "of", tests_run, "tests failed")
        return 1
    else:
        print("    PASS: all", tests_run, "tests passed")
    return 0


if __name__ == "__main__":
    sys.exit(Main())