Tools/scripts/run_coverage.py - master - Sangak/ardupilot

Tools/scripts/run_coverage.py
315 строк14 KB
Pierre Kancir
Tools: remove all modules from coverage
16 июн 2026, 01:40
16 июн 2026, 01:40c566a9c
#!/usr/bin/env python3

"""
Runs tests with gcov coverage support.

 AP_FLAKE8_CLEAN
"""
import argparse
import os
import shutil
import subprocess
import sys
import tempfile
import time

os.environ['PYTHONUNBUFFERED'] = '1'
os.set_blocking(sys.stdout.fileno(), True)
os.set_blocking(sys.stderr.fileno(), True)

tools_dir = os.path.dirname(os.path.realpath(__file__))
root_dir = os.path.realpath(os.path.join(tools_dir, '../..'))


class CoverageRunner(object):
    """Coverage Runner Class."""

    def __init__(self, verbose=False, check_tests=True) -> None:
        """Set the files Path."""
        self.REPORT_DIR = os.path.join(root_dir, "reports/lcov-report")
        self.INFO_FILE = os.path.join(root_dir, self.REPORT_DIR, "lcov.info")
        self.INFO_FILE_BASE = os.path.join(root_dir, self.REPORT_DIR, "lcov_base.info")
        self.LCOV_LOG = os.path.join(root_dir, "GCOV_lcov.log")
        self.GENHTML_LOG = os.path.join(root_dir, "GCOV_genhtml.log")

        self.autotest = os.path.join(root_dir, "Tools/autotest/autotest.py")
        self.verbose = verbose
        self.check_tests = check_tests
        self.start_time = time.time()

    def progress(self, text) -> None:
        """Pretty printer."""
        delta_time = time.time() - self.start_time
        formatted_text = "****** AT-%06.1f: %s" % (delta_time, text)
        print(formatted_text)

    def init_coverage(self, use_example=False) -> None:
        """Initialize ArduPilot for coverage.

        This needs to be run with the binaries built.
        """
        self.progress("Initializing Coverage...")
        self.progress("Removing previous reports")
        try:
            shutil.rmtree(self.REPORT_DIR)
        except FileNotFoundError:
            pass

        try:
            os.makedirs(self.REPORT_DIR)
        except FileExistsError:
            pass

        self.progress("Checking that vehicles binaries are set up and built")
        binaries_dir = os.path.join(root_dir, 'build/sitl/bin')
        dirs_to_check = [["binaries", binaries_dir], ["tests", os.path.join(root_dir, 'build/linux/tests')]]
        if use_example:
            self.progress("Adding examples")
            dirs_to_check.append(["examples", os.path.join(root_dir, 'build/linux/examples')])
        for dirc in dirs_to_check:
            if not (self.check_build(dirc[0], dirc[1])):
                self.run_build()
                break

        self.progress("Zeroing previous build")
        retcode = subprocess.call(["lcov", "--zerocounters", "--directory", root_dir])
        if retcode != 0:
            self.progress("Failed with retcode (%s)" % retcode)
            exit(1)

        self.progress("Initializing Coverage with current build")
        try:
            result = subprocess.run(["lcov",
                                     "--no-external",
                                     "--initial",
                                     "--capture",
                                     "--exclude", root_dir + "/build/sitl/modules/*",
                                     "--directory", root_dir,
                                     "-o", self.INFO_FILE_BASE,
                                     ], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=True)
            if self.verbose:
                print(*result.args)
                print(result.stdout)
            with open(self.LCOV_LOG, 'w') as log_file:
                log_file.write(result.stdout)
        except subprocess.CalledProcessError as err:
            print("ERROR :")
            print(err.cmd)
            print(err.output)
            exit(1)
        self.progress("Initialization done")

    def check_build(self, name, path) -> bool:
        """Check that build directory is not empty and that binaries are built with the coverage flags."""
        self.progress("Checking that %s are set up and built" % name)
        if os.path.exists(path):
            if not os.listdir(path):
                return False
        else:
            return False
        self.progress("Checking %s build configuration for coverage" % name)
        with open(os.path.join(path, "../compile_commands.json"), "r") as searchfile:
            for line in searchfile:
                if "-ftest-coverage" in line:
                    return True
            self.progress("%s wasn't built with coverage support" % name)
            return False

    def run_build(self, use_example=False) -> None:
        """Clean the build directory and build binaries for coverage."""

        os.chdir(root_dir)
        waf_light = os.path.join(root_dir, "modules/waf/waf-light")
        self.progress("Removing previous build binaries")
        subprocess.run([waf_light, "configure", "--debug"], check=True)
        subprocess.run([waf_light, "clean"], check=True)

        self.progress("Building examples and SITL binaries")

        try:
            if use_example:
                self.progress("Building examples")
                subprocess.run([waf_light, "configure", "--board=linux", "--debug", "--coverage"], check=True)
                subprocess.run([waf_light, "examples"], check=True)
            subprocess.run(
                [self.autotest,
                 "--debug",
                 "--coverage",
                 "build.unit_tests"],
                check=True)
            subprocess.run([waf_light, "configure", "--debug", "--coverage"], check=True)
            subprocess.run([waf_light], check=True)
        except subprocess.CalledProcessError as err:
            print("ERROR :")
            print(err.cmd)
            print(err.output)
            exit(1)
        self.progress("Build examples and vehicle binaries done !")

    def run_full(self, use_example=False) -> None:
        """Run full coverage on maximum of ArduPilot binaries and test functions."""
        self.progress("Running full test suite...")
        self.run_build()
        self.init_coverage()
        self.progress("Running tests")
        SPEEDUP = 5
        TIMEOUT = 14400

        if use_example:
            self.progress("Running run.examples")
            subprocess.run([self.autotest,
                            "--timeout=" + str(TIMEOUT),
                            "--debug",
                            "--coverage",
                            "--no-clean",
                            "--speedup=" + str(SPEEDUP),
                            "run.examples"], check=self.check_tests)
        self.progress("Running run.unit_tests")
        subprocess.run(
            [self.autotest,
             "--timeout=" + str(TIMEOUT),
             "--debug",
             "--no-clean",
             "run.unit_tests"], check=self.check_tests)
        subprocess.run(["reset"], check=True)
        os.set_blocking(sys.stdout.fileno(), True)
        os.set_blocking(sys.stderr.fileno(), True)
        test_list = ["Plane", "QuadPlane", "Sub", "Copter", "Helicopter", "Rover", "Tracker", "BalanceBot", "Sailboat"]
        for test in test_list:
            self.progress("Running test.%s" % test)
            try:
                subprocess.run([self.autotest,
                                "--timeout=" + str(TIMEOUT),
                                "--debug",
                                "--no-clean",
                                "test.%s" % test], check=self.check_tests)
            except subprocess.CalledProcessError:
                # pass in case of failing tests
                pass
        # TODO add any other execution path/s we can to maximise the actually
        # used code, can we run other tests or things?  Replay, perhaps?
        self.update_stats()

    def update_stats(self) -> None:
        """Update Coverage statistics only.

        Assumes that coverage tests have been run.
        """
        self.progress("Generating Coverage statistics")
        with open(self.LCOV_LOG, 'a') as log_file:
            # we cannot use subprocess.PIPE and result.stdout to get the output as it will be too long and trigger
            # BlockingIOError: [Errno 11] write could not complete without blocking
            # thus we output to temp file, and print the file line by line...
            with tempfile.NamedTemporaryFile(mode="w+") as tmp_file:
                try:
                    self.progress("Capturing Coverage statistics")
                    subprocess.run(["lcov",
                                    "--no-external",
                                    "--capture",
                                    "--directory", root_dir,
                                    "-o", self.INFO_FILE,
                                    ], stdout=tmp_file, stderr=subprocess.STDOUT, text=True, check=True)
                    if self.verbose:
                        tmp_file.seek(0)
                        content = tmp_file.read().splitlines()
                        for line in content:
                            print(line, flush=True)
                            log_file.write(line)

                    self.progress("Matching Coverage with binaries")
                    subprocess.run(["lcov",
                                    "--add-tracefile", self.INFO_FILE_BASE,
                                    "--add-tracefile", self.INFO_FILE,
                                    ], stdout=tmp_file, stderr=subprocess.STDOUT, text=True, check=True)
                    if self.verbose:
                        tmp_file.seek(0)
                        content = tmp_file.read().splitlines()
                        for line in content:
                            # print(line, flush=True)  # not useful to print
                            log_file.write(line)
                    # remove files we do not intentionally test:
                    self.progress("Removing unwanted coverage statistics")
                    subprocess.run(["lcov",
                                    "--remove", self.INFO_FILE,
                                    ".waf*",
                                    root_dir + "/modules/*",
                                    root_dir + "/build/linux/libraries/*",
                                    root_dir + "/build/linux/modules/*",
                                    root_dir + "/build/sitl/libraries/*",
                                    root_dir + "/build/sitl/modules/*",
                                    root_dir + "/build/sitl_periph_universal/libraries/*",
                                    root_dir + "/build/sitl_periph_universal/modules/*",
                                    root_dir + "/libraries/*/examples/*",
                                    root_dir + "/libraries/*/tests/*",
                                    "-o", self.INFO_FILE
                                    ], stdout=tmp_file, stderr=subprocess.STDOUT, text=True, check=True)
                    if self.verbose:
                        tmp_file.seek(0)
                        content = tmp_file.read().splitlines()
                        for line in content:
                            # print(line, flush=True)  # not useful to print
                            log_file.write(line)

                except subprocess.CalledProcessError as err:
                    print("ERROR :")
                    print(err.cmd)
                    print(err.output)
                    sys.exit(1)

        with open(self.GENHTML_LOG, 'w+') as log_file:
            try:
                self.progress("Generating HTML files")
                subprocess.run(["genhtml", self.INFO_FILE,
                                "-o", self.REPORT_DIR,
                                "--demangle-cpp",
                                ], stdout=log_file, stderr=subprocess.STDOUT, text=True, check=True)

                log_file.seek(0)
                content = log_file.read().splitlines()
                if self.verbose:
                    for line in content[1: -3]:
                        print(line, flush=True)
                for line in content[-3:]:
                    print(line, flush=True)
            except subprocess.CalledProcessError as err:
                print("ERROR :")
                print(err.cmd)
                print(err.output)
                exit(1)
        self.progress("Coverage successful. Open " + self.REPORT_DIR + "/index.html")


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Runs tests with gcov coverage support.')
    parser.add_argument('-v', '--verbose', action='store_true',
                        help='Output everything on terminal.')
    parser.add_argument('-c', '--no-check-tests', action='store_true',
                        help='Do not fail if tests do not run.')
    parser.add_argument('--add-examples', action='store_true',
                        help='Add examples to coverage.')
    group = parser.add_mutually_exclusive_group()
    group.add_argument('-i', '--init', action='store_true',
                       help='Initialise ArduPilot for coverage. It should be run after building the binaries.')
    group.add_argument('-f', '--full', action='store_true',
                       help='Run ArduPilot full coverage. This will run all tests and example. It is really long.')
    group.add_argument('-b', '--build', action='store_true',
                       help='Clean the build directory and build binaries for coverage.')
    group.add_argument('-u', '--update', action='store_true',
                       help='Update coverage statistics. To be used after running some tests.')
    args = parser.parse_args()

    runner = CoverageRunner(verbose=args.verbose, check_tests=not args.no_check_tests)
    if args.init:
        runner.init_coverage(args.add_examples)
        sys.exit(0)
    if args.full:
        runner.run_full(args.add_examples)
        sys.exit(0)
    if args.build:
        runner.run_build(args.add_examples)
        sys.exit(0)
    if args.update:
        runner.update_stats()
        sys.exit(0)
    parser.print_help()
    sys.exit(0)