test: Enable coverage in integration tests

integration-requirements.txt

@@ -11,3 +11,4 @@ pytest!=7.3.2
 
 packaging
 passlib
+coverage

tests/integration_tests/assets/enable_coverage.py

@@ -0,0 +1,28 @@
+from pathlib import Path
+
+services = [
+    "cloud-init-local.service",
+    "cloud-init.service",
+    "cloud-config.service",
+    "cloud-final.service",
+]
+service_dir = Path("/lib/systemd/system/")
+
+# Check for the existence of the service files
+for service in services:
+    if not (service_dir / service).is_file():
+        print(f"Error: {service} does not exist in {service_dir}")
+        exit(1)
+
+# Prepend the ExecStart= line with 'python3 -m coverage run'
+for service in services:
+    file_path = service_dir / service
+    content = file_path.read_text()
+    content = content.replace(
+        "ExecStart=/usr",
+        (
+            "ExecStart=python3 -m coverage run "
+            "--source=/usr/lib/python3/dist-packages/cloudinit --append /usr"
+        )
+    )
+    file_path.write_text(content)

tests/integration_tests/conftest.py

@@ -3,11 +3,13 @@
 import functools
 import logging
 import os
+import shutil
+import subprocess
 import sys
 from contextlib import contextmanager
 from pathlib import Path
 from tarfile import TarFile
-from typing import Dict, Generator, Iterator, Type
+from typing import Dict, Generator, Iterator, List, Type
 
 import pytest
 from pycloudlib.lxd.instance import LXDInstance
@@ -115,11 +117,18 @@ def setup_image(session_cloud: IntegrationCloud, request):
     So we can launch instances / run tests with the correct image
     """
     source = get_validated_source(session_cloud)
-    if not source.installs_new_version():
+    if not (
+        source.installs_new_version() or integration_settings.INCLUDE_COVERAGE
+    ):
         return
-    log.info("Setting up environment for %s", session_cloud.datasource)
+    log.info("Setting up source image")
     client = session_cloud.launch()
-    client.install_new_cloud_init(source)
+    if source.installs_new_version():
+        log.info("Installing cloud-init from %s", source.name)
+        client.install_new_cloud_init(source)
+    if integration_settings.INCLUDE_COVERAGE:
+        log.info("Installing coverage")
+        client.install_coverage()
     # All done customizing the image, so snapshot it and make it global
     snapshot_id = client.snapshot()
     client.cloud.snapshot_id = snapshot_id
@@ -134,28 +143,35 @@ def setup_image(session_cloud: IntegrationCloud, request):
     request.addfinalizer(session_cloud.delete_snapshot)
 
 
-def _collect_logs(
-    instance: IntegrationInstance, node_id: str, test_failed: bool
-):
-    """Collect logs from remote instance.
-
-    Args:
-        instance: The current IntegrationInstance to collect logs from
-        node_id: The pytest representation of this test, E.g.:
-            tests/integration_tests/test_example.py::TestExample.test_example
-        test_failed: If test failed or not
-    """
-    if any(
-        [
-            integration_settings.COLLECT_LOGS == "NEVER",
-            integration_settings.COLLECT_LOGS == "ON_ERROR"
-            and not test_failed,
-        ]
-    ):
-        return
+def _collect_logs(instance: IntegrationInstance, log_dir: Path):
     instance.execute(
         "cloud-init collect-logs -u -t /var/tmp/cloud-init.tar.gz"
     )
+    log.info("Writing logs to %s", log_dir)
+
+    tarball_path = log_dir / "cloud-init.tar.gz"
+    try:
+        instance.pull_file("/var/tmp/cloud-init.tar.gz", tarball_path)
+    except Exception as e:
+        log.error("Failed to pull logs: %s", e)
+        return
+
+    tarball = TarFile.open(str(tarball_path))
+    tarball.extractall(path=str(log_dir))
+    tarball_path.unlink()
+
+
+def _collect_coverage(instance: IntegrationInstance, log_dir: Path):
+    log.info("Writing coverage report to %s", log_dir)
+    try:
+        instance.pull_file("/.coverage", log_dir / ".coverage")
+    except Exception as e:
+        log.error("Failed to pull coverage for: %s", e)
+
+
+def _setup_artifact_paths(node_id: str):
+    parent_dir = Path(integration_settings.LOCAL_LOG_PATH, session_start_time)
+
     node_id_path = Path(
         node_id.replace(
             ".py", ""
@@ -164,32 +180,45 @@ def _collect_logs(
         .replace("[", "-")  # For parametrized names
         .replace("]", "")  # For parameterized names
     )
-    log_dir = (
-        Path(integration_settings.LOCAL_LOG_PATH)
-        / session_start_time
-        / node_id_path
-    )
-    log.info("Writing logs to %s", log_dir)
+    log_dir = parent_dir / node_id_path
 
+    # Create log dir if not exists
     if not log_dir.exists():
         log_dir.mkdir(parents=True)
 
     # Add a symlink to the latest log output directory
     last_symlink = Path(integration_settings.LOCAL_LOG_PATH) / "last"
     if os.path.islink(last_symlink):
         os.unlink(last_symlink)
-    os.symlink(log_dir.parent, last_symlink)
+    os.symlink(parent_dir, last_symlink)
+    return log_dir
 
-    tarball_path = log_dir / "cloud-init.tar.gz"
-    try:
-        instance.pull_file("/var/tmp/cloud-init.tar.gz", tarball_path)
-    except Exception as e:
-        log.error("Failed to pull logs: %s", e)
+
+def _collect_artifacts(
+    instance: IntegrationInstance, node_id: str, test_failed: bool
+):
+    """Collect artifacts from remote instance.
+
+    Args:
+        instance: The current IntegrationInstance to collect artifacts from
+        node_id: The pytest representation of this test, E.g.:
+            tests/integration_tests/test_example.py::TestExample.test_example
+        test_failed: If test failed or not
+    """
+    should_collect_logs = integration_settings.COLLECT_LOGS == "ALWAYS" or (
+        integration_settings.COLLECT_LOGS == "ON_ERROR" and test_failed
+    )
+    should_collect_coverage = integration_settings.INCLUDE_COVERAGE
+    if not (should_collect_logs or should_collect_coverage):
         return
 
-    tarball = TarFile.open(str(tarball_path))
-    tarball.extractall(path=str(log_dir))
-    tarball_path.unlink()
+    log_dir = _setup_artifact_paths(node_id)
+
+    if should_collect_logs:
+        _collect_logs(instance, log_dir)
+
+    if should_collect_coverage:
+        _collect_coverage(instance, log_dir)
 
 
 @contextmanager
@@ -240,7 +269,7 @@ def _client(
         previous_failures = request.session.testsfailed
         yield instance
         test_failed = request.session.testsfailed - previous_failures > 0
-        _collect_logs(instance, request.node.nodeid, test_failed)
+        _collect_artifacts(instance, request.node.nodeid, test_failed)
 
 
 @pytest.fixture
@@ -311,3 +340,45 @@ def pytest_configure(config):
         # If log_cli_level is available in this version of pytest and not set
         # to anything, set it to INFO.
         config.option.log_cli_level = "INFO"
+
+
+def _copy_coverage_files(parent_dir: Path) -> List[Path]:
+    combined_files = []
+    for dirpath in parent_dir.rglob("*"):
+        if (dirpath / ".coverage").exists():
+            # Construct the new filename
+            relative_dir = dirpath.relative_to(parent_dir)
+            new_filename = ".coverage." + str(relative_dir).replace(
+                os.sep, "-"
+            )
+            new_filepath = parent_dir / new_filename
+
+            # Copy the file
+            shutil.copy(dirpath / ".coverage", new_filepath)
+            combined_files.append(new_filepath)
+    return combined_files
+
+
+def _generate_coverage_report() -> None:
+    log.info("Generating coverage report")
+    parent_dir = Path(integration_settings.LOCAL_LOG_PATH, session_start_time)
+    coverage_files = _copy_coverage_files(parent_dir)
+    subprocess.run(
+        ["coverage", "combine"] + [str(f) for f in coverage_files],
+        check=True,
+        cwd=str(parent_dir),
+        stdout=subprocess.DEVNULL,
+    )
+    subprocess.run(
+        ["coverage", "html", "--ignore-errors"],
+        check=True,
+        cwd=str(parent_dir),
+        stdout=subprocess.DEVNULL,
+    )
+    log.info("Coverage report generated")
+
+
+def pytest_sessionfinish(session, exitstatus) -> None:
+    if not integration_settings.INCLUDE_COVERAGE:
+        return
+    _generate_coverage_report()

tests/integration_tests/instances.py

@@ -4,12 +4,14 @@
 import uuid
 from enum import Enum
 from tempfile import NamedTemporaryFile
+from typing import Union
 
 from pycloudlib.instance import BaseInstance
 from pycloudlib.result import Result
 
 from tests.integration_tests import integration_settings
 from tests.integration_tests.decorators import retry
+from tests.integration_tests.util import ASSETS_DIR
 
 try:
     from typing import TYPE_CHECKING
@@ -78,13 +80,21 @@ def execute(self, command, *, use_sudo=True) -> Result:
             raise RuntimeError("Root user cannot run unprivileged")
         return self.instance.execute(command, use_sudo=use_sudo)
 
-    def pull_file(self, remote_path, local_path):
+    def pull_file(
+        self,
+        remote_path: Union[str, os.PathLike],
+        local_path: Union[str, os.PathLike],
+    ):
         # First copy to a temporary directory because of permissions issues
         tmp_path = _get_tmp_path()
         self.instance.execute("cp {} {}".format(str(remote_path), tmp_path))
         self.instance.pull_file(tmp_path, str(local_path))
 
-    def push_file(self, local_path, remote_path):
+    def push_file(
+        self,
+        local_path: Union[str, os.PathLike],
+        remote_path: Union[str, os.PathLike],
+    ):
         # First push to a temporary directory because of permissions issues
         tmp_path = _get_tmp_path()
         self.instance.push_file(str(local_path), tmp_path)
@@ -124,6 +134,15 @@ def snapshot(self):
         log.info("Created new image: %s", image_id)
         return image_id
 
+    def install_coverage(self):
+        self._apt_update()
+        self.execute("apt-get install -qy python3-coverage")
+        self.push_file(
+            local_path=ASSETS_DIR / "enable_coverage.py",
+            remote_path="/var/tmp/enable_coverage.py",
+        )
+        assert self.execute("python3 /var/tmp/enable_coverage.py").ok
+
     def install_new_cloud_init(
         self,
         source: CloudInitSource,
@@ -148,30 +167,24 @@ def install_new_cloud_init(
         if clean:
             self.instance.clean()
 
-    # assert with retry because we can compete with apt already running in the
-    # background and get: E: Could not get lock /var/lib/apt/lists/lock - open
-    # (11: Resource temporarily unavailable)
-
-    @retry(tries=30, delay=1)
     def install_proposed_image(self):
         log.info("Installing proposed image")
         assert self.execute(
             'echo deb "http://archive.ubuntu.com/ubuntu '
             '$(lsb_release -sc)-proposed main" >> '
             "/etc/apt/sources.list.d/proposed.list"
         ).ok
-        assert self.execute("apt-get update -q").ok
+        self._apt_update()
         assert self.execute(
             "apt-get install -qy cloud-init -t=$(lsb_release -sc)-proposed"
         ).ok
 
-    @retry(tries=30, delay=1)
     def install_ppa(self):
         log.info("Installing PPA")
         assert self.execute(
             "add-apt-repository {} -y".format(self.settings.CLOUD_INIT_SOURCE)
         ).ok
-        assert self.execute("apt-get update -q").ok
+        self._apt_update()
         assert self.execute("apt-get install -qy cloud-init").ok
 
     @retry(tries=30, delay=1)
@@ -190,9 +203,35 @@ def install_deb(self):
     @retry(tries=30, delay=1)
     def upgrade_cloud_init(self):
         log.info("Upgrading cloud-init to latest version in archive")
-        assert self.execute("apt-get update -q").ok
+        self._apt_update()
         assert self.execute("apt-get install -qy cloud-init").ok
 
+    def _apt_update(self):
+        """Run an apt update.
+
+        `cloud-init single` allows us to ensure apt update is only run once
+        for this instance. It could be done with an lru_cache too, but
+        dogfooding is fun."""
+        self.write_to_file(
+            "/tmp/update-ci.yaml", "#cloud-config\npackage_update: true"
+        )
+        response = self.execute(
+            "cloud-init single --name package_update_upgrade_install "
+            "--frequency instance --file /tmp/update-ci.yaml"
+        )
+        if not response.ok:
+            if response.stderr.startswith("usage:"):
+                # https://github.com/canonical/cloud-init/pull/4559 hasn't
+                # landed yet, so we need to use the old syntax
+                response = self.execute(
+                    "cloud-init --file /tmp/update-ci.yaml single --name "
+                    "package_update_upgrade_install --frequency instance "
+                )
+            if response.stderr:
+                raise RuntimeError(
+                    f"Failed to update packages: {response.stderr}"
+                )
+
     def ip(self) -> str:
         if self._ip:
             return self._ip

tests/integration_tests/integration_settings.py

@@ -81,6 +81,12 @@
 COLLECT_LOGS = "ON_ERROR"
 LOCAL_LOG_PATH = "/tmp/cloud_init_test_logs"
 
+# We default our coverage to False because it involves modifying the
+# cloud-init systemd services, which is too intrusive of a change to
+# enable by default. If changed to true, the LOCAL_LOG_PATH defined
+# above will contain an `html` directory with the coverage report.
+INCLUDE_COVERAGE = False
+
 ##################################################################
 # USER SETTINGS OVERRIDES
 ##################################################################