[elbe-devel] [PATCH 01/11] cov: Introduce elbe-ci coverage

[Olivier Dion]

Coverage of source code is done by using python-coverage [1].  The
version 4.x is very important to use because this is the only version
available on Buster for the initvm (see Notes).

Coverage is initiated by a top process.  This process set an
environment variable to tell others processes (mainly recursive
execution of Elbe) that coverage should be activated.  The descendants
of the top process can only do coverage, but they will never combine
nor do report of the measurement they have made.  This is done by the
top process only.

The coverage is a per-thread attribute.  python-coverage [1] should be
thread aware, but I haven't tested it yet.  The current problem with
this implementation however is that it's using os.environ.  Which
means that if thread A starts coverage, then children forked from
thread B will also start coverage of them self.  This could be fix by
using a per-thread environment and merging this environment in
functions in 'elbepack/shellhelper.py' and ensuring that all recursive
call to Elbe use these functions.

[1] python-coverage - https://coverage.readthedocs.io/en/coverage-4.5.4/

Signed-off-by: Olivier Dion <dion at linutronix.de>
---
 elbepack/cov.py | 173 ++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 173 insertions(+)
 create mode 100644 elbepack/cov.py

diff --git a/elbepack/cov.py b/elbepack/cov.py
new file mode 100644
index 00000000..5190fe1f
--- /dev/null
+++ b/elbepack/cov.py
@@ -0,0 +1,173 @@
+# ELBE - Debian Based Embedded Rootfilesystem Builder
+# Copyright (c) 2020 Olivier Dion <dion at linutronix.de>
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# elbepack/cov.py - Wrapper around coverage.py
+#
+# Coverage is activated in Elbe by using the ElbeCoverage class as a
+# context manager.  Doing so will set the ELBE_COVERAGE environment
+# variable to the path where the coverage files are saved to.
+#
+# This is only done by what I call the 'top' process.  This is the
+# process, or thread, that started the coverage of Elbe.  Only the top
+# process can combine coverage files and make a report.
+#
+# This is useful because Elbe execve itself a lot.  And so, when Elbe
+# starts, if it sees the ELBE_COVERAGE environment variable set, it
+# will automatically start coverage of itself as an inferior process,
+# i.e. not the top process.
+
+import os
+import threading
+import tempfile
+
+import coverage
+
+# TODO:py3 Remove object inheritance
+# pylint: disable=useless-object-inheritance
+class ElbeCoverage(object):
+
+    local = threading.local()
+
+    def __init__(self,
+                 en_coverage=False,
+                 coverage_path=None,
+                 sources=None,
+                 report=None,
+                 ctx=None):
+
+        self.en_coverage   = en_coverage
+        self.sources       = sources or ["."]
+        self.is_top        = False
+        self.report        = report
+        self.ctx           = ctx
+        self.coverage_path = coverage_path
+
+        self._ensure_local_attrs()
+
+        # This is not thread aware!!!  This _will_ result in mixed coverage of
+        # differents projects, if Elbe can do parallel build!
+        #
+        # For example, if coverage is enable for build A but not B,
+        # setting ELBE_COVERAGE in thread A might will result in coverage
+        # for thread B and its children.
+        #
+        # To fix this, make functions in elbepack/shellhelper.py aware
+        # of local.env
+        #
+        # TODO - Replace the following line with the commented one whenever Elbe
+        # can do parallel build.
+        #
+        # self.env = local.env
+        self.env = os.environ
+
+    def _ensure_local_attrs(self):
+        """Make sure that thread local variables exists before doing anything"""
+        for attr in ("cov", "coverage_dir", "coverage_path"):
+            if not hasattr(self.local, attr):
+                setattr(self.local, attr, None)
+
+        if not hasattr(self.local, "env"):
+            self.local.env = {}
+
+    def _setup_ctx(self):
+        """Setup the coverage context. This setup is initiated by the top
+        process, e.g. the one that started the coverage"""
+
+        if not self.coverage_path:
+            self.local.coverage_dir  = tempfile.TemporaryDirectory(prefix="elbe-cov")
+            self.local.coverage_path = self.local.coverage_dir.name
+        else:
+            self.local.coverage_dir = None
+            self.local.coverage_path = self.coverage_path
+
+        # Force coverage on recursive execution of Elbe
+        self.env["ELBE_COVERAGE"] = self.local.coverage_path
+
+    def __enter__(self):
+
+        # NO OP
+        if not self.en_coverage:
+            return self
+
+        # We're the first process to activate the coverage
+        self.is_top = "ELBE_COVERAGE" not in self.env
+
+        if self.is_top:
+            self._setup_ctx()
+
+        if self.local.cov is None:
+
+            # Note here that data_suffix is set to True for thread
+            # that are not the top process.  This will generate a
+            # .coverage* file with unique identification
+            self.local.cov = coverage.Coverage(branch=False,
+                                               source=self.sources,
+
+                                               # TODO:sid - Uncomment the following to
+                                               # allow context in coverage
+                                               #
+                                               # context=self.ctx,
+
+                                               data_file=os.path.join(self.env["ELBE_COVERAGE"], ".coverage"),
+                                               data_suffix=not self.is_top)
+            self.local.cov.load()
+            self.local.cov.start()
+
+        return self
+
+    def __exit__(self, _typ, _value, _tb):
+
+        if not self.local.cov:
+            return
+
+        # Stop coverage of this thread
+        self.local.cov.stop()
+        self.local.cov.save()
+
+        # Only the top process can go further
+        if not self.is_top:
+            return
+
+        self.local.cov.combine([self.local.coverage_path])
+
+        if self.report:
+            with open(self.report, "w") as f:
+                self.local.cov.report(show_missing=True, skip_covered=True,
+                                      file=f)
+
+        self.local.cov = None
+
+        # Don't do coverage on children anymore
+        del self.env["ELBE_COVERAGE"]
+
+        if self.local.coverage_dir:
+            self.local.coverage_dir.cleanup()
+            self.local.coverage_dir = None
+
+    @classmethod
+    def merge(cls, paths):
+        if cls.local.cov:
+            cls.local.cov.combine(paths)
+
+    @staticmethod
+    def rename_files(path, from_str, to_str):
+        """Replace from_str with to_str in all measured files found in path.
+        This is useful for converting coverage in different
+        filesystems.  e.g. on host and initvm
+        """
+
+        data = coverage.CoverageData()
+        data.read_file(path)
+        files = data.measured_files()
+
+        line_data = {}
+
+        for old in files:
+            new = old.replace(from_str, to_str)
+            line_data[new] = data.lines(old)
+
+        data.erase()
+        data.add_lines(line_data)
+        data.write_file(path)
-- 
2.28.0