twister: coverage: Data collection and reporting per-test instance

With this change, the coverage data (GCOV dump) is extracted from the
test log files after each of the test instance parallel execution,
instead of post processing all the logs at the end of the Twister run.

The new `--coverage-per-instance` mode extends Twister coverage operations
to report coverage statistics on each test instance execution individually
in addition to the default reporting mode which aggregates data to one
report with all the test instances in the current scope of the Twister run.
The split mode allows to identify precisely what amount of code coverage
each test suite provides and to analyze its contribution to the overall
test plan's coverage. Each test configuration's output directory will have
its own coverage report and data files, so the overall disk space and
the total execution time increase.

Another new `--disable-coverage-aggregation` option allows to execute
only the `--coverage-per-instance` mode when the aggregate coverage
report for the whole Twister run scope is not needed.

Signed-off-by: Dmitrii Golovanov <[email protected]>

Author: golowanow

scripts/pylib/twister/twisterlib/coverage.py

@@ -1,6 +1,6 @@
 # vim: set syntax=python ts=4 :
 #
-# Copyright (c) 2018-2022 Intel Corporation
+# Copyright (c) 2018-2025 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
 
 import contextlib
@@ -31,6 +31,8 @@ def __init__(self):
         self.gcov_tool = None
         self.base_dir = None
         self.output_formats = None
+        self.coverage_capture = True
+        self.coverage_report = True
 
     @staticmethod
     def factory(tool, jobs=None):
@@ -109,7 +111,7 @@ def merge_hexdumps(self, hexdumps):
 
     def create_gcda_files(self, extracted_coverage_info):
         gcda_created = True
-        logger.debug("Generating gcda files")
+        logger.debug(f"Generating {len(extracted_coverage_info)} gcda files")
         for filename, hexdumps in extracted_coverage_info.items():
             # if kobject_hash is given for coverage gcovr fails
             # hence skipping it problem only in gcovr v4.1
@@ -132,7 +134,7 @@ def create_gcda_files(self, extracted_coverage_info):
                 gcda_created = False
         return gcda_created
 
-    def generate(self, outdir):
+    def capture_data(self, outdir):
         coverage_completed = True
         for filename in glob.glob(f"{outdir}/**/handler.log", recursive=True):
             gcov_data = self.__class__.retrieve_gcov_data(filename)
@@ -148,9 +150,15 @@ def generate(self, outdir):
             else:
                 logger.error(f"Gcov data capture incomplete: {filename}")
                 coverage_completed = False
+        return coverage_completed
 
+    def generate(self, outdir):
+        coverage_completed = self.capture_data(outdir) if self.coverage_capture else True
+        if not coverage_completed or not self.coverage_report:
+            return coverage_completed, {}
+        reports = {}
         with open(os.path.join(outdir, "coverage.log"), "a") as coveragelog:
-            ret = self._generate(outdir, coveragelog)
+            ret, reports = self._generate(outdir, coveragelog)
             if ret == 0:
                 report_log = {
                     "html": "HTML report generated: {}".format(
@@ -180,7 +188,7 @@ def generate(self, outdir):
             else:
                 coverage_completed = False
         logger.debug(f"All coverage data processed: {coverage_completed}")
-        return coverage_completed
+        return coverage_completed, reports
 
 
 class Lcov(CoverageTool):
@@ -274,6 +282,7 @@ def _generate(self, outdir, coveragelog):
                "--output-file", ztestfile]
         self.run_lcov(cmd, coveragelog)
 
+        files = []
         if os.path.exists(ztestfile) and os.path.getsize(ztestfile) > 0:
             cmd = ["--remove", ztestfile,
                    os.path.join(self.base_dir, "tests/ztest/test/*"),
@@ -289,12 +298,15 @@ def _generate(self, outdir, coveragelog):
             self.run_lcov(cmd, coveragelog)
 
         if 'html' not in self.output_formats.split(','):
-            return 0
+            return 0, {}
 
         cmd = ["genhtml", "--legend", "--branch-coverage",
                "--prefix", self.base_dir,
                "-output-directory", os.path.join(outdir, "coverage")] + files
-        return self.run_command(cmd, coveragelog)
+        ret = self.run_command(cmd, coveragelog)
+
+        # TODO: Add LCOV summary coverage report.
+        return ret, { 'report': coveragefile, 'ztest': ztestfile, 'summary': None }
 
 
 class Gcovr(CoverageTool):
@@ -344,8 +356,9 @@ def _flatten_list(list):
         return [a for b in list for a in b]
 
     def _generate(self, outdir, coveragelog):
-        coveragefile = os.path.join(outdir, "coverage.json")
-        ztestfile = os.path.join(outdir, "ztest.json")
+        coverage_file = os.path.join(outdir, "coverage.json")
+        coverage_summary = os.path.join(outdir, "coverage_summary.json")
+        ztest_file = os.path.join(outdir, "ztest.json")
 
         excludes = Gcovr._interleave_list("-e", self.ignores)
         if len(self.ignore_branch_patterns) > 0:
@@ -358,24 +371,37 @@ def _generate(self, outdir, coveragelog):
         mode_options = ["--merge-mode-functions=separate"]
 
         # We want to remove tests/* and tests/ztest/test/* but save tests/ztest
-        cmd = ["gcovr", "-r", self.base_dir,
+        cmd = ["gcovr", "-v", "-r", self.base_dir,
                "--gcov-ignore-parse-errors=negative_hits.warn_once_per_file",
                "--gcov-executable", self.gcov_tool,
                "-e", "tests/*"]
-        cmd += excludes + mode_options + ["--json", "-o", coveragefile, outdir]
+        cmd += excludes + mode_options + ["--json", "-o", coverage_file, outdir]
         cmd_str = " ".join(cmd)
-        logger.debug(f"Running {cmd_str}...")
-        subprocess.call(cmd, stdout=coveragelog)
-
-        subprocess.call(["gcovr", "-r", self.base_dir, "--gcov-executable",
-                         self.gcov_tool, "-f", "tests/ztest", "-e",
-                         "tests/ztest/test/*", "--json", "-o", ztestfile,
-                         outdir] + mode_options, stdout=coveragelog)
-
-        if os.path.exists(ztestfile) and os.path.getsize(ztestfile) > 0:
-            files = [coveragefile, ztestfile]
+        logger.debug(f"Running: {cmd_str}")
+        coveragelog.write(f"Running: {cmd_str}\n")
+        coveragelog.flush()
+        ret = subprocess.call(cmd, stdout=coveragelog, stderr=coveragelog)
+        if ret:
+            logger.error(f"GCOVR failed with {ret}")
+            return ret, {}
+
+        cmd = ["gcovr", "-v", "-r", self.base_dir] + mode_options
+        cmd += ["--gcov-executable", self.gcov_tool,
+                "-f", "tests/ztest", "-e", "tests/ztest/test/*",
+                "--json", "-o", ztest_file, outdir]
+        cmd_str = " ".join(cmd)
+        logger.debug(f"Running: {cmd_str}")
+        coveragelog.write(f"Running: {cmd_str}\n")
+        coveragelog.flush()
+        ret = subprocess.call(cmd, stdout=coveragelog, stderr=coveragelog)
+        if ret:
+            logger.error(f"GCOVR ztest stage failed with {ret}")
+            return ret, {}
+
+        if os.path.exists(ztest_file) and os.path.getsize(ztest_file) > 0:
+            files = [coverage_file, ztest_file]
         else:
-            files = [coveragefile]
+            files = [coverage_file]
 
         subdir = os.path.join(outdir, "coverage")
         os.makedirs(subdir, exist_ok=True)
@@ -396,21 +422,21 @@ def _generate(self, outdir, coveragelog):
             [report_options[r] for r in self.output_formats.split(',')]
         )
 
-        return subprocess.call(
-            ["gcovr", "-r", self.base_dir] \
-                + mode_options + gcovr_options + tracefiles, stdout=coveragelog
-        )
+        cmd = ["gcovr", "-v", "-r", self.base_dir] + mode_options + gcovr_options + tracefiles
+        cmd += ["--json-summary-pretty", "--json-summary", coverage_summary]
+        cmd_str = " ".join(cmd)
+        logger.debug(f"Running: {cmd_str}")
+        coveragelog.write(f"Running: {cmd_str}\n")
+        coveragelog.flush()
+        ret = subprocess.call(cmd, stdout=coveragelog, stderr=coveragelog)
+        if ret:
+            logger.error(f"GCOVR merge report stage failed with {ret}")
 
+        return ret, { 'report': coverage_file, 'ztest': ztest_file, 'summary': coverage_summary }
 
 
-def run_coverage(testplan, options):
-    use_system_gcov = False
+def choose_gcov_tool(options, is_system_gcov):
     gcov_tool = None
-
-    for plat in options.coverage_platform:
-        _plat = testplan.get_platform(plat)
-        if _plat and (_plat.type in {"native", "unit"}):
-            use_system_gcov = True
     if not options.gcov_tool:
         zephyr_sdk_gcov_tool = os.path.join(
             os.environ.get("ZEPHYR_SDK_INSTALL_DIR", default=""),
@@ -427,7 +453,7 @@ def run_coverage(testplan, options):
             except OSError:
                 shutil.copy(llvm_cov, gcov_lnk)
             gcov_tool = gcov_lnk
-        elif use_system_gcov:
+        elif is_system_gcov:
             gcov_tool = "gcov"
         elif os.path.exists(zephyr_sdk_gcov_tool):
             gcov_tool = zephyr_sdk_gcov_tool
@@ -439,10 +465,19 @@ def run_coverage(testplan, options):
     else:
         gcov_tool = str(options.gcov_tool)
 
-    logger.info("Generating coverage files...")
-    logger.info(f"Using gcov tool: {gcov_tool}")
+    return gcov_tool
+
+
+def run_coverage_tool(options, outdir, is_system_gcov, coverage_capture, coverage_report):
     coverage_tool = CoverageTool.factory(options.coverage_tool, jobs=options.jobs)
-    coverage_tool.gcov_tool = gcov_tool
+    if not coverage_tool:
+        return False, {}
+
+    coverage_tool.gcov_tool = str(choose_gcov_tool(options, is_system_gcov))
+    logger.debug(f"Using gcov tool: {coverage_tool.gcov_tool}")
+
+    coverage_tool.coverage_capture = coverage_capture
+    coverage_tool.coverage_report = coverage_report
     coverage_tool.base_dir = os.path.abspath(options.coverage_basedir)
     # Apply output format default
     if options.coverage_formats is not None:
@@ -456,5 +491,32 @@ def run_coverage(testplan, options):
     # Ignore branch coverage on __ASSERT* macros
     # Covering the failing case is not desirable as it will immediately terminate the test.
     coverage_tool.add_ignore_branch_pattern(r"^\s*__ASSERT(?:_EVAL|_NO_MSG|_POST_ACTION)?\(.*")
-    coverage_completed = coverage_tool.generate(options.outdir)
-    return coverage_completed
+    return coverage_tool.generate(outdir)
+
+
+def has_system_gcov(platform):
+    return platform and (platform.type in {"native", "unit"})
+
+
+def run_coverage(options, testplan):
+    """ Summary code coverage over the full test plan's scope.
+    """
+    is_system_gcov = False
+
+    for plat in options.coverage_platform:
+        if has_system_gcov(testplan.get_platform(plat)):
+            is_system_gcov = True
+            break
+
+    return run_coverage_tool(options, options.outdir, is_system_gcov,
+                             coverage_capture=False,
+                             coverage_report=True)
+
+
+def run_coverage_instance(options, instance):
+    """ Per-instance code coverage called by ProjectBuilder ('coverage' operation).
+    """
+    is_system_gcov = has_system_gcov(instance.platform)
+    return run_coverage_tool(options, instance.build_dir, is_system_gcov,
+                             coverage_capture=True,
+                             coverage_report=options.coverage_per_instance)

scripts/pylib/twister/twisterlib/environment.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 # vim: set syntax=python ts=4 :
 #
-# Copyright (c) 2018-2024 Intel Corporation
+# Copyright (c) 2018-2025 Intel Corporation
 # Copyright 2022 NXP
 # Copyright (c) 2024 Arm Limited (or its affiliates). All rights reserved.
 #
@@ -380,7 +380,7 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
                          "Default to what was selected with --platform.")
 
     coverage_group.add_argument("--coverage-tool", choices=['lcov', 'gcovr'], default='gcovr',
-                    help="Tool to use to generate coverage report (%(default)s - default).")
+                    help="Tool to use to generate coverage reports (%(default)s - default).")
 
     coverage_group.add_argument("--coverage-formats", action="store", default=None,
                     help="Output formats to use for generated coverage reports " +
@@ -390,6 +390,19 @@ def add_parse_arguments(parser = None) -> argparse.ArgumentParser:
                          " Valid options for 'lcov' tool are: " +
                          ','.join(supported_coverage_formats['lcov']) + " (html,lcov - default).")
 
+    coverage_group.add_argument("--coverage-per-instance", action="store_true", default=False,
+                help="""Compose individual coverage reports for each test suite
+                        when coverage reporting is enabled; it might run in addition to
+                        the default aggregation mode which produces one coverage report for
+                        all executed test suites. Default: %(default)s""")
+
+    coverage_group.add_argument("--disable-coverage-aggregation",
+                action="store_true", default=False,
+                help="""Don't aggregate coverage report statistics for all the test suites
+                        selected to run with enabled coverage. Requires another reporting mode to be
+                        active (`--coverage-split`) to have at least one of these reporting modes.
+                        Default: %(default)s""")
+
     parser.add_argument(
         "--test-config",
         action="store",
@@ -908,6 +921,21 @@ def parse_arguments(
     if options.enable_coverage and not options.coverage_platform:
         options.coverage_platform = options.platform
 
+    if (
+        (not options.coverage)
+        and (options.disable_coverage_aggregation or options.coverage_per_instance)
+    ):
+        logger.error("Enable coverage reporting to set its aggregation mode.")
+        sys.exit(1)
+
+    if (
+        options.coverage
+        and options.disable_coverage_aggregation and (not options.coverage_per_instance)
+    ):
+        logger.error("At least one coverage reporting mode should be enabled: "
+                     "either aggregation, or per-instance, or both.")
+        sys.exit(1)
+
     if options.coverage_formats:
         for coverage_format in options.coverage_formats.split(','):
             if coverage_format not in supported_coverage_formats[options.coverage_tool]:

scripts/pylib/twister/twisterlib/reports.py

@@ -58,6 +58,7 @@ def __init__(self, plan, env) -> None:
         self.outdir = os.path.abspath(env.options.outdir)
         self.instance_fail_count = plan.instance_fail_count
         self.footprint = None
+        self.coverage_status = None
 
 
     @staticmethod

scripts/pylib/twister/twisterlib/runner.py

@@ -1,6 +1,6 @@
 # vim: set syntax=python ts=4 :
 #
-# Copyright (c) 2018-2024 Intel Corporation
+# Copyright (c) 2018-2025 Intel Corporation
 # Copyright 2022 NXP
 # SPDX-License-Identifier: Apache-2.0
 
@@ -42,6 +42,7 @@
 
 sys.path.insert(0, os.path.join(ZEPHYR_BASE, "scripts/pylib/build_helpers"))
 from domains import Domains
+from twisterlib.coverage import run_coverage_instance
 from twisterlib.environment import TwisterEnv
 from twisterlib.harness import Ctest, HarnessImporter, Pytest
 from twisterlib.log_helper import log_command
@@ -1130,7 +1131,7 @@ def process(self, pipeline, done, message, lock, results):
                 self.instance.handler.thread = None
                 self.instance.handler.duts = None
 
-                next_op = 'report'
+                next_op = "coverage" if self.options.coverage else "report"
                 additionals = {
                     "status": self.instance.status,
                     "reason": self.instance.reason
@@ -1146,6 +1147,28 @@ def process(self, pipeline, done, message, lock, results):
             finally:
                 self._add_to_pipeline(pipeline, next_op, additionals)
 
+        # Run per-instance code coverage
+        elif op == "coverage":
+            try:
+                logger.debug(f"Run coverage for '{self.instance.name}'")
+                self.instance.coverage_status, self.instance.coverage = \
+                        run_coverage_instance(self.options, self.instance)
+                next_op = 'report'
+                additionals = {
+                    "status": self.instance.status,
+                    "reason": self.instance.reason
+                }
+            except StatusAttributeError as sae:
+                logger.error(str(sae))
+                self.instance.status = TwisterStatus.ERROR
+                reason = f"Incorrect status assignment on {op}"
+                self.instance.reason = reason
+                self.instance.add_missing_case_status(TwisterStatus.BLOCK, reason)
+                next_op = 'report'
+                additionals = {}
+            finally:
+                self._add_to_pipeline(pipeline, next_op, additionals)
+
         # Report results and output progress to screen
         elif op == "report":
             try:

scripts/pylib/twister/twisterlib/testinstance.py

@@ -1,6 +1,6 @@
 # vim: set syntax=python ts=4 :
 #
-# Copyright (c) 2018-2024 Intel Corporation
+# Copyright (c) 2018-2025 Intel Corporation
 # Copyright 2022 NXP
 # Copyright (c) 2024 Arm Limited (or its affiliates). All rights reserved.
 #
@@ -59,6 +59,8 @@ def __init__(self, testsuite, platform, toolchain, outdir):
         self.metrics = dict()
         self.handler = None
         self.recording = None
+        self.coverage = None
+        self.coverage_status = None
         self.outdir = outdir
         self.execution_time = 0
         self.build_time = 0