diff --git a/.github/workflows/nightly-test.yaml b/.github/workflows/nightly-test.yaml index b4a57a9de..8a542622b 100644 --- a/.github/workflows/nightly-test.yaml +++ b/.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' diff --git a/tests/integration/tests/test_strict_interfaces.py b/tests/integration/tests/test_strict_interfaces.py new file mode 100644 index 000000000..58d5df75e --- /dev/null +++ b/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}", + ], + ) diff --git a/tests/integration/tests/test_util/config.py b/tests/integration/tests/test_util/config.py index 45c9ea5e6..07012288a 100644 --- a/tests/integration/tests/test_util/config.py +++ b/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 strict' to get the latest channels for strict. +STRICT_INTERFACE_CHANNELS = ( + os.environ.get("TEST_STRICT_INTERFACE_CHANNELS", "").strip().split() +) diff --git a/tests/integration/tests/test_util/snap.py b/tests/integration/tests/test_util/snap.py index 9ad77ab49..fe0219fe2 100644 --- a/tests/integration/tests/test_util/snap.py +++ b/tests/integration/tests/test_util/snap.py @@ -37,27 +37,9 @@ 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/` 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: @@ -65,29 +47,65 @@ def get_latest_channels( 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 diff --git a/tests/integration/tests/test_util/util.py b/tests/integration/tests/test_util/util.py index 391d73f6a..1e5b911de 100644 --- a/tests/integration/tests/test_util/util.py +++ b/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( diff --git a/tests/integration/tests/test_version_upgrades.py b/tests/integration/tests/test_version_upgrades.py index 3d9207759..6df9171ae 100644 --- a/tests/integration/tests/test_version_upgrades.py +++ b/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(