Strict interfaces test

.github/workflows/nightly-test.yaml

@@ -2,7 +2,7 @@ name: Nightly Latest/Edge Tests
 
 on:
   schedule:
-    - cron: '0 0 * * *' # Runs every midnight
+    - cron: "0 0 * * *" # Runs every midnight
 
 permissions:
   contents: read
@@ -45,6 +45,7 @@ jobs:
           # Test the latest (up to) 6 releases for the flavour
           # TODO(ben): upgrade nightly to run all flavours
           TEST_VERSION_UPGRADE_CHANNELS: "recent 6 classic"
+          TEST_STRICT_INTERFACE_CHANNELS: "recent 6 strict"
         run: |
           export PATH="/home/runner/.local/bin:$PATH"
           cd tests/integration && sg lxd -c 'tox -vve integration'

tests/integration/tests/test_strict_interfaces.py

@@ -0,0 +1,75 @@
+#
+# Copyright 2024 Canonical, Ltd.
+#
+import logging
+from typing import List
+
+import pytest
+from test_util import config, harness, snap, util
+
+LOG = logging.getLogger(__name__)
+
+
+@pytest.mark.node_count(1)
+@pytest.mark.no_setup()
+@pytest.mark.skipif(
+    not config.STRICT_INTERFACE_CHANNELS, reason="No strict channels configured"
+)
+def test_strict_interfaces(instances: List[harness.Instance], tmp_path):
+    channels = config.STRICT_INTERFACE_CHANNELS
+    cp = instances[0]
+    current_channel = channels[0]
+
+    if current_channel.lower() == "recent":
+        if len(channels) != 3:
+            pytest.fail(
+                "'recent' requires the number of releases as second argument and the flavour as third argument"
+            )
+        _, num_channels, flavour = channels
+        channels = snap.get_channels(int(num_channels), flavour, cp.arch, "edge", True)
+
+    for channel in channels:
+        util.setup_k8s_snap(cp, tmp_path, channel, connect_interfaces=False)
+
+        # Log the current snap version on the node.
+        out = cp.exec(["snap", "list", config.SNAP_NAME], capture_output=True)
+        LOG.info(f"Current snap version: {out.stdout.decode().strip()}")
+
+        check_snap_interfaces(cp, config.SNAP_NAME)
+
+        cp.exec(["snap", "remove", config.SNAP_NAME, "--purge"])
+
+
+def check_snap_interfaces(cp, snap_name):
+    """Check the strict snap interfaces."""
+    interfaces = [
+        "docker-privileged",
+        "kubernetes-support",
+        "network",
+        "network-bind",
+        "network-control",
+        "network-observe",
+        "firewall-control",
+        "process-control",
+        "kernel-module-observe",
+        "cilium-module-load",
+        "mount-observe",
+        "hardware-observe",
+        "system-observe",
+        "home",
+        "opengl",
+        "home-read-all",
+        "login-session-observe",
+        "log-observe",
+    ]
+    for interface in interfaces:
+        cp.exec(
+            [
+                "snap",
+                "run",
+                "--shell",
+                snap_name,
+                "-c",
+                f"snapctl is-connected {interface}",
+            ],
+        )

tests/integration/tests/test_util/config.py

@@ -101,3 +101,8 @@
 VERSION_UPGRADE_CHANNELS = (
     os.environ.get("TEST_VERSION_UPGRADE_CHANNELS", "").strip().split()
 )
+# A list of space-separated channels for which the strict interface tests should be run in sequential order.
+# Alternatively, use 'recent <num> strict' to get the latest <num> channels for strict.
+STRICT_INTERFACE_CHANNELS = (
+    os.environ.get("TEST_STRICT_INTERFACE_CHANNELS", "").strip().split()
+)

tests/integration/tests/test_util/snap.py

@@ -37,57 +37,75 @@ def get_snap_info(snap_name=SNAP_NAME):
         raise
 
 
-def get_latest_channels(
-    num_of_channels: int, flavor: str, arch: str, include_latest=True
-) -> List[str]:
-    """Get an ascending list of latest channels based on the number of channels and flavour.
-
-    e.g. get_latest_release_channels(3, "classic") -> ['1.31-classic/candidate', '1.30-classic/candidate']
-    if there are less than num_of_channels available, return all available channels.
-    Only the most stable risk level is returned for each major.minor version.
-    By default, the `latest/edge/<flavor>` channel is included in the list.
-    """
-    snap_info = get_snap_info()
-
-    # Extract channel information
-    channels = snap_info.get("channel-map", [])
-    available_channels = [
-        ch["channel"]["name"]
-        for ch in channels
-        if ch["channel"]["architecture"] == arch
-    ]
-
-    # Define regex pattern to match channels in the format 'major.minor-flavour'
+def filter_arch_and_flavor(channels: List[dict], arch: str, flavor: str) -> List[tuple]:
+    """Filter available channels by architecture and match them with a given regex pattern
+    for a flavor."""
     if flavor == "strict":
         pattern = re.compile(r"(\d+)\.(\d+)\/(" + "|".join(RISK_LEVELS) + ")")
     else:
         pattern = re.compile(
             r"(\d+)\.(\d+)-" + re.escape(flavor) + r"\/(" + "|".join(RISK_LEVELS) + ")"
         )
 
-    # Dictionary to store the highest risk level for each major.minor
+    matched_channels = []
+    for ch in channels:
+        if ch["channel"]["architecture"] == arch:
+            channel_name = ch["channel"]["name"]
+            match = pattern.match(channel_name)
+            if match:
+                major, minor, risk = match.groups()
+                matched_channels.append((channel_name, int(major), int(minor), risk))
+
+    return matched_channels
+
+
+def get_most_stable_channels(
+    num_of_channels: int, flavor: str, arch: str, include_latest=True
+) -> List[str]:
+    """Get an ascending list of latest channels based on the number of channels
+    flavour and architecture."""
+    snap_info = get_snap_info()
+
+    # Extract channel information and filter by architecture and flavor
+    arch_flavor_channels = filter_arch_and_flavor(
+        snap_info.get("channel-map", []), arch, flavor
+    )
+
+    # Dictionary to store the most stable channels for each version
     channel_map = {}
+    for channel, major, minor, risk in arch_flavor_channels:
+        version_key = (int(major), int(minor))
+
+        if version_key not in channel_map or RISK_LEVELS.index(
+            risk
+        ) < RISK_LEVELS.index(channel_map[version_key][1]):
+            channel_map[version_key] = (channel, risk)
+
+    # Sort channels by major and minor version (ascending order)
+    sorted_versions = sorted(channel_map.keys(), key=lambda v: (v[0], v[1]))
+
+    # Extract only the channel names
+    final_channels = [channel_map[v][0] for v in sorted_versions[:num_of_channels]]
 
-    for channel in available_channels:
-        match = pattern.match(channel)
-        if match:
-            major, minor, risk = match.groups()
-            major_minor = (int(major), int(minor))
+    if include_latest:
+        final_channels.append(f"latest/edge/{flavor}")
 
-            # Store only the highest risk level channel for each major.minor
-            if major_minor not in channel_map or RISK_LEVELS.index(
-                risk
-            ) < RISK_LEVELS.index(channel_map[major_minor][1]):
-                channel_map[major_minor] = (channel, risk)
+    return final_channels
 
-    # Sort channels by major and minor version in descending order
-    sorted_channels = sorted(channel_map.keys(), reverse=False)
 
-    # Prepare final channel list
-    final_channels = [channel_map[mm][0] for mm in sorted_channels[:num_of_channels]]
+def get_channels(
+    num_of_channels: int, flavor: str, arch: str, risk_level: str, include_latest=True
+) -> List[str]:
+    """Get channels based on the risk level, architecture and flavour."""
+    snap_info = get_snap_info()
+    arch_flavor_channels = filter_arch_and_flavor(
+        snap_info.get("channel-map", []), arch, flavor
+    )
 
+    matching_channels = [ch[0] for ch in arch_flavor_channels if ch[3] == risk_level]
+    matching_channels = matching_channels[:num_of_channels]
     if include_latest:
         latest_channel = f"latest/edge/{flavor}"
-        final_channels.append(latest_channel)
+        matching_channels.append(latest_channel)
 
-    return final_channels
+    return matching_channels

tests/integration/tests/test_util/util.py

@@ -141,7 +141,10 @@ def _as_int(value: Optional[str]) -> Optional[int]:
 
 
 def setup_k8s_snap(
-    instance: harness.Instance, tmp_path: Path, snap: Optional[str] = None
+    instance: harness.Instance,
+    tmp_path: Path,
+    snap: Optional[str] = None,
+    connect_interfaces=True,
 ):
     """Installs and sets up the snap on the given instance and connects the interfaces.
 
@@ -176,8 +179,9 @@ def setup_k8s_snap(
         cmd += [config.SNAP_NAME, "--channel", channel]
 
     instance.exec(cmd)
-    LOG.info("Ensure k8s interfaces and network requirements")
-    instance.exec(["/snap/k8s/current/k8s/hack/init.sh"], stdout=subprocess.DEVNULL)
+    if connect_interfaces:
+        LOG.info("Ensure k8s interfaces and network requirements")
+        instance.exec(["/snap/k8s/current/k8s/hack/init.sh"], stdout=subprocess.DEVNULL)
 
 
 def wait_until_k8s_ready(

tests/integration/tests/test_version_upgrades.py

@@ -26,7 +26,7 @@ def test_version_upgrades(instances: List[harness.Instance], tmp_path):
                 "'recent' requires the number of releases as second argument and the flavour as third argument"
             )
         _, num_channels, flavour = channels
-        channels = snap.get_latest_channels(int(num_channels), flavour, cp.arch)
+        channels = snap.get_most_stable_channels(int(num_channels), flavour, cp.arch)
         current_channel = channels[0]
 
     LOG.info(