android-12.0.0_r34/s

#!/usr/bin/env python3
# Copyright 2021 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
#
# Test runner for crosvm:
# - Selects which tests to run based on local environment
# - Can run some tests single-threaded
# - Can run some tests using the VM provided by the builders.
# - Can generate junit xml files for integration with sponge
#
# The crates and feature to test are configured in ./run_tests

from typing import Iterable, List, Dict, Set, Optional, Union
import argparse
import enum
import os
import platform
import subprocess
import sys
import re
import xml.etree.ElementTree as ET
import pathlib

# Print debug info. Overriden by -v or -vv
VERBOSE = False
VERY_VERBOSE = False

# Runs tests using the exec_file wrapper, which will run the test inside the
# builders built-in VM.
VM_TEST_RUNNER = (
    os.path.abspath("./ci/vm_tools/exec_binary_in_vm") + " --no-sync"
)

# Runs tests using QEMU user-space emulation.
QEMU_TEST_RUNNER = (
    "qemu-aarch64-static -E LD_LIBRARY_PATH=/workspace/scratch/lib"
)

# Kill a test after 5 minutes to prevent frozen tests from running too long.
TEST_TIMEOUT_SECS = 300


class Requirements(enum.Enum):
    # Test can only be built for aarch64.
    AARCH64 = "aarch64"

    # Test can only be built for x86_64.
    X86_64 = "x86_64"

    # Requires ChromeOS build environment.
    CROS_BUILD = "cros_build"

    # Test is disabled explicitly.
    DISABLED = "disabled"

    # Test needs to be executed with expanded privileges for device access and
    # will be run inside a VM.
    PRIVILEGED = "privileged"

    # Test needs to run single-threaded
    SINGLE_THREADED = "single_threaded"

    # Separate workspaces that have dev-dependencies cannot be built from the
    # crosvm workspace and need to be built separately.
    # Note: Separate workspaces are built with no features enabled.
    SEPARATE_WORKSPACE = "separate_workspace"

    # Build, but do not run.
    DO_NOT_RUN = "do_not_run"


BUILD_TIME_REQUIREMENTS = [
    Requirements.AARCH64,
    Requirements.X86_64,
    Requirements.CROS_BUILD,
    Requirements.DISABLED,
]


class CrateInfo(object):
    """Informaton about whether a crate can be built or run on this host."""

    def __init__(
        self,
        name: str,
        requirements: Set[Requirements],
        capabilities: Set[Requirements],
    ):
        self.name = name
        self.requirements = requirements
        self.single_threaded = Requirements.SINGLE_THREADED in requirements
        self.needs_privilege = Requirements.PRIVILEGED in requirements

        build_reqs = requirements.intersection(BUILD_TIME_REQUIREMENTS)
        self.can_build = all(req in capabilities for req in build_reqs)

        self.can_run = (
            self.can_build
            and (
                not self.needs_privilege
                or Requirements.PRIVILEGED in capabilities
            )
            and not Requirements.DO_NOT_RUN in self.requirements
        )

    def __repr__(self):
        return f"{self.name} {self.requirements}"


def target_arch():
    """Returns architecture cargo is set up to build for."""
    if "CARGO_BUILD_TARGET" in os.environ:
        target = os.environ["CARGO_BUILD_TARGET"]
        return target.split("-")[0]
    else:
        return platform.machine()


def get_test_runner_env(use_vm: bool):
    """Sets the target.*.runner cargo setting to use the correct test runner."""
    env = os.environ.copy()
    key = f"CARGO_TARGET_{target_arch().upper()}_UNKNOWN_LINUX_GNU_RUNNER"
    if use_vm:
        env[key] = VM_TEST_RUNNER
    else:
        if target_arch() == "aarch64":
            env[key] = QEMU_TEST_RUNNER
        else:
            if key in env:
                del env[key]
    return env


class TestResult(enum.Enum):
    PASS = "Pass"
    FAIL = "Fail"
    SKIP = "Skip"
    UNKNOWN = "Unknown"


class CrateResults(object):
    """Container for results of a single cargo test call."""

    def __init__(self, crate_name: str, success: bool, cargo_test_log: str):
        self.crate_name = crate_name
        self.success = success
        self.cargo_test_log = cargo_test_log

        # Parse "test test_name... ok|ignored|FAILED" messages from cargo log.
        test_regex = re.compile(r"^test ([\w\/_\-\.:() ]+) \.\.\. (\w+)$")
        self.tests: Dict[str, TestResult] = {}
        for line in cargo_test_log.split(os.linesep):
            match = test_regex.match(line)
            if match:
                name = match.group(1)
                result = match.group(2)
                if result == "ok":
                    self.tests[name] = TestResult.PASS
                elif result == "ignored":
                    self.tests[name] = TestResult.SKIP
                elif result == "FAILED":
                    self.tests[name] = TestResult.FAIL
                else:
                    self.tests[name] = TestResult.UNKNOWN

    def total(self):
        return len(self.tests)

    def count(self, result: TestResult):
        return sum(r == result for r in self.tests.values())

    def to_junit(self):
        testsuite = ET.Element(
            "testsuite",
            {
                "name": self.crate_name,
                "tests": str(self.total()),
                "failures": str(self.count(TestResult.FAIL)),
            },
        )
        for (test, result) in self.tests.items():
            testcase = ET.SubElement(
                testsuite, "testcase", {"name": f"{self.crate_name} - ${test}"}
            )
            if result == TestResult.SKIP:
                ET.SubElement(
                    testcase, "skipped", {"message": "Disabled in rust code."}
                )
            else:
                testcase.set("status", "run")
                if result == TestResult.FAIL:
                    failure = ET.SubElement(
                        testcase, "failure", {"message": "Test failed."}
                    )
                    failure.text = self.cargo_test_log

        return testsuite


class RunResults(object):
    """Container for results of the whole test run."""

    def __init__(self, crate_results: Iterable[CrateResults]):
        self.crate_results = list(crate_results)
        self.success: bool = (
            len(self.crate_results) > 0 and self.count(TestResult.FAIL) == 0
        )

    def total(self):
        return sum(r.total() for r in self.crate_results)

    def count(self, result: TestResult):
        return sum(r.count(result) for r in self.crate_results)

    def to_junit(self):
        testsuites = ET.Element("testsuites", {"name": "Cargo Tests"})
        for crate_result in self.crate_results:
            testsuites.append(crate_result.to_junit())
        return testsuites


def results_summary(results: Union[RunResults, CrateResults]):
    """Returns a concise 'N passed, M failed' summary of `results`"""
    num_pass = results.count(TestResult.PASS)
    num_skip = results.count(TestResult.SKIP)
    num_fail = results.count(TestResult.FAIL)
    msg: List[str] = []
    if num_pass:
        msg.append(f"{num_pass} passed")
    if num_skip:
        msg.append(f"{num_skip} skipped")
    if num_fail:
        msg.append(f"{num_fail} failed")
    return ", ".join(msg)


def cargo_build_process(
    cwd: str = ".", crates: List[CrateInfo] = [], features: Set[str] = set()
):
    """Builds the main crosvm crate."""
    cmd = [
        "cargo",
        "build",
        "--color=never",
        "--no-default-features",
        "--features",
        ",".join(features),
    ]

    for crate in sorted(crate.name for crate in crates):
        cmd += ["-p", crate]

    if VERY_VERBOSE:
        print("CMD", " ".join(cmd))

    process = subprocess.run(
        cmd,
        cwd=cwd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
    )
    if process.returncode != 0 or VERBOSE:
        print()
        print(process.stdout)
    return process


def cargo_test_process(
    cwd: str,
    crates: List[CrateInfo] = [],
    features: Set[str] = set(),
    run: bool = True,
    single_threaded: bool = False,
    use_vm: bool = False,
    timeout: Optional[int] = None,
):
    """Creates the subprocess to run `cargo test`."""
    cmd = ["cargo", "test", "--color=never"]
    if not run:
        cmd += ["--no-run"]
    if features:
        cmd += ["--no-default-features", "--features", ",".join(features)]

    # Skip doc tests as these cannot be run in the VM.
    if use_vm:
        cmd += ["--bins", "--tests"]

    for crate in sorted(crate.name for crate in crates):
        cmd += ["-p", crate]

    cmd += ["--", "--color=never"]
    if single_threaded:
        cmd += ["--test-threads=1"]
    env = get_test_runner_env(use_vm)

    if VERY_VERBOSE:
        print("ENV", env)
        print("CMD", " ".join(cmd))

    process = subprocess.run(
        cmd,
        cwd=cwd,
        env=env,
        timeout=timeout,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
    )
    if process.returncode != 0 or VERBOSE:
        print()
        print(process.stdout)
    return process


def cargo_build_tests(crates: List[CrateInfo], features: Set[str]):
    """Runs cargo test --no-run to build all listed `crates`."""
    separate_workspace_crates = [
        crate
        for crate in crates
        if Requirements.SEPARATE_WORKSPACE in crate.requirements
    ]
    workspace_crates = [
        crate
        for crate in crates
        if Requirements.SEPARATE_WORKSPACE not in crate.requirements
    ]

    print(
        "Building workspace: ",
        ", ".join(crate.name for crate in workspace_crates),
    )
    build_process = cargo_build_process(
        cwd=".", crates=workspace_crates, features=features
    )
    if build_process.returncode != 0:
        return False
    test_process = cargo_test_process(
        cwd=".", crates=workspace_crates, features=features, run=False
    )
    if test_process.returncode != 0:
        return False

    for crate in separate_workspace_crates:
        print("Building crate:", crate.name)
        build_process = cargo_build_process(cwd=crate.name)
        if build_process.returncode != 0:
            return False
        test_process = cargo_test_process(cwd=crate.name, run=False)
        if test_process.returncode != 0:
            return False
    return True


def cargo_test(
    crates: List[CrateInfo],
    features: Set[str],
    single_threaded: bool = False,
    use_vm: bool = False,
) -> Iterable[CrateResults]:
    """Runs cargo test for all listed `crates`."""
    for crate in crates:
        msg = ["Testing crate", crate.name]
        if use_vm:
            msg.append("in vm")
        if single_threaded:
            msg.append("(single-threaded)")
        if Requirements.SEPARATE_WORKSPACE in crate.requirements:
            msg.append("(separate workspace)")
        sys.stdout.write(f"{' '.join(msg)}... ")
        sys.stdout.flush()

        if Requirements.SEPARATE_WORKSPACE in crate.requirements:
            process = cargo_test_process(
                cwd=crate.name,
                run=True,
                single_threaded=single_threaded,
                use_vm=use_vm,
                timeout=TEST_TIMEOUT_SECS,
            )
        else:
            process = cargo_test_process(
                cwd=".",
                crates=[crate],
                features=features,
                run=True,
                single_threaded=single_threaded,
                use_vm=use_vm,
                timeout=TEST_TIMEOUT_SECS,
            )
        results = CrateResults(
            crate.name, process.returncode == 0, process.stdout
        )
        print(results_summary(results))
        yield results


def execute_batched_by_parallelism(
    crates: List[CrateInfo], features: Set[str], use_vm: bool
) -> Iterable[CrateResults]:
    """Batches tests by single-threaded and parallel, then executes them."""
    run_single = [crate for crate in crates if crate.single_threaded]
    yield from cargo_test(
        run_single, features, single_threaded=True, use_vm=use_vm
    )

    run_parallel = [crate for crate in crates if not crate.single_threaded]
    yield from cargo_test(run_parallel, features, use_vm=use_vm)


def execute_batched_by_privilege(
    crates: List[CrateInfo], features: Set[str], use_vm: bool
) -> Iterable[CrateResults]:
    """
    Batches tests by whether or not a test needs privileged access to run.

    Non-privileged tests are run first. Privileged tests are executed in
    a VM if use_vm is set.
    """
    build_crates = [crate for crate in crates if crate.can_build]
    if not cargo_build_tests(build_crates, features):
        return []

    simple_crates = [
        crate for crate in crates if crate.can_run and not crate.needs_privilege
    ]
    yield from execute_batched_by_parallelism(
        simple_crates, features, use_vm=False
    )

    privileged_crates = [
        crate for crate in crates if crate.can_run and crate.needs_privilege
    ]
    if privileged_crates:
        if use_vm:
            subprocess.run("./ci/vm_tools/sync_deps", check=True)
            yield from execute_batched_by_parallelism(
                privileged_crates, features, use_vm=True
            )
        else:
            yield from execute_batched_by_parallelism(
                privileged_crates, features, use_vm=False
            )


def results_report(
    feature_requirements: Dict[str, List[Requirements]],
    crates: List[CrateInfo],
    features: Set[str],
    run_results: RunResults,
):
    """Prints a summary report of all test results."""
    print()

    if len(run_results.crate_results) == 0:
        print("Could not build tests.")
        return

    crates_not_built = [crate.name for crate in crates if not crate.can_build]
    print(f"Crates not built: {', '.join(crates_not_built)}")

    crates_not_run = [
        crate.name for crate in crates if crate.can_build and not crate.can_run
    ]
    print(f"Crates not tested: {', '.join(crates_not_run)}")

    disabled_features: Set[str] = set(feature_requirements.keys()).difference(
        features
    )
    print(f"Disabled features: {', '.join(disabled_features)}")

    print()
    if not run_results.success:
        for crate_results in run_results.crate_results:
            if crate_results.success:
                continue
            print(f"Test failures in {crate_results.crate_name}:")
            for (test, result) in crate_results.tests.items():
                if result == TestResult.FAIL:
                    print(f"  {test}")
        print()
        print("Some tests failed:", results_summary(run_results))
    else:
        print("All tests passed:", results_summary(run_results))


def execute_tests(
    crate_requirements: Dict[str, List[Requirements]],
    feature_requirements: Dict[str, List[Requirements]],
    capabilities: Set[Requirements],
    use_vm: bool,
    junit_file: Optional[str] = None,
):
    print("Capabilities:", ", ".join(cap.value for cap in capabilities))

    # Select all features where capabilities meet the requirements
    features = set(
        feature
        for (feature, requirements) in feature_requirements.items()
        if all(r in capabilities for r in requirements)
    )

    # Disable sandboxing for tests until our builders are set up to run with
    # sandboxing.
    features.add("default-no-sandbox")
    print("Features:", ", ".join(features))

    crates = [
        CrateInfo(crate, set(requirements), capabilities)
        for (crate, requirements) in crate_requirements.items()
    ]
    run_results = RunResults(
        execute_batched_by_privilege(crates, features, use_vm)
    )

    if junit_file:
        pathlib.Path(junit_file).parent.mkdir(parents=True, exist_ok=True)
        ET.ElementTree(run_results.to_junit()).write(junit_file)

    results_report(feature_requirements, crates, features, run_results)
    if not run_results.success:
        exit(-1)


DESCRIPTION = """\
Runs tests for crosvm based on the capabilities of the local host.

This script can be run directly on a worksation to run a limited number of tests
that can be built and run on a standard debian system.

It can also be run via the CI builder: `./ci/builder --vm ./run_tests`. This
will build all tests and runs tests that require special privileges inside the
virtual machine provided by the builder.
"""


def main(
    crate_requirements: Dict[str, List[Requirements]],
    feature_requirements: Dict[str, List[Requirements]],
):
    parser = argparse.ArgumentParser(description=DESCRIPTION)
    parser.add_argument(
        "--verbose",
        "-v",
        action="store_true",
        default=False,
        help="Print all test output.",
    )
    parser.add_argument(
        "--very-verbose",
        "-vv",
        action="store_true",
        default=False,
        help="Print debug information and commands executed.",
    )
    parser.add_argument(
        "--run-privileged",
        action="store_true",
        default=False,
        help="Enable tests that requires privileged access to the system.",
    )
    parser.add_argument(
        "--cros-build",
        action="store_true",
        default=False,
        help=(
            "Enables tests that require a ChromeOS build environment. "
            "Can also be set by CROSVM_CROS_BUILD"
        ),
    )
    parser.add_argument(
        "--use-vm",
        action="store_true",
        default=False,
        help=(
            "Enables privileged tests to run in a VM. "
            "Can also be set by CROSVM_USE_VM"
        ),
    )
    parser.add_argument(
        "--require-all",
        action="store_true",
        default=False,
        help="Requires all tests to run, fail if tests would be disabled.",
    )
    parser.add_argument(
        "--junit-file",
        default=None,
        help="Path to file where to store junit xml results",
    )
    args = parser.parse_args()

    global VERBOSE, VERY_VERBOSE
    VERBOSE = args.verbose or args.very_verbose  # type: ignore
    VERY_VERBOSE = args.very_verbose  # type: ignore

    use_vm = os.environ.get("CROSVM_USE_VM") != None or args.use_vm
    cros_build = os.environ.get("CROSVM_CROS_BUILD") != None or args.cros_build

    capabilities = set()
    if target_arch() == "aarch64":
        capabilities.add(Requirements.AARCH64)
    elif target_arch() == "x86_64":
        capabilities.add(Requirements.X86_64)

    if cros_build:
        capabilities.add(Requirements.CROS_BUILD)

    if use_vm:
        if not os.path.exists("/workspace/vm"):
            print("--use-vm can only be used within the ./ci/builder's.")
            exit(1)
        capabilities.add(Requirements.PRIVILEGED)

    if args.run_privileged:
        capabilities.add(Requirements.PRIVILEGED)

    if args.require_all and not Requirements.PRIVILEGED in capabilities:
        print("--require-all needs to be run with --use-vm or --run-privileged")
        exit(1)

    execute_tests(
        crate_requirements,
        feature_requirements,
        capabilities,
        use_vm,
        args.junit_file,
    )