Implement version_downgrade_with_rollback test

.github/workflows/e2e-tests.yaml

@@ -65,6 +65,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_VERSION_DOWNGRADE_CHANNELS: "recent 6 classic"
           # Upgrading from 1.30 is not supported.
           TEST_VERSION_UPGRADE_MIN_RELEASE: "1.31"
           TEST_STRICT_INTERFACE_CHANNELS: "recent 6 strict"

tests/integration/tests/test_util/config.py

@@ -141,6 +141,11 @@
 # Only relevant when using 'recent' in VERSION_UPGRADE_CHANNELS.
 VERSION_UPGRADE_MIN_RELEASE = os.environ.get("TEST_VERSION_UPGRADE_MIN_RELEASE")
 
+# Same usage as VERSION_UPGRADE_MIN_RELEASE but for downgrades.
+VERSION_DOWNGRADE_CHANNELS = (
+    os.environ.get("TEST_VERSION_DOWNGRADE_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 = (

tests/integration/tests/test_util/snap.py

@@ -67,6 +67,7 @@ def get_most_stable_channels(
     arch: str,
     include_latest: bool = True,
     min_release: Optional[str] = None,
+    reverse: bool = False,
 ) -> List[str]:
     """Get an ascending list of latest channels based on the number of channels
     flavour and architecture."""
@@ -93,7 +94,9 @@ def get_most_stable_channels(
             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]))
+    sorted_versions = sorted(
+        channel_map.keys(), key=lambda v: (v[0], v[1]), reverse=reverse
+    )
 
     # Extract only the channel names
     final_channels = [channel_map[v][0] for v in sorted_versions[:num_of_channels]]

tests/integration/tests/test_util/util.py

@@ -262,6 +262,7 @@ def wait_until_k8s_ready(
     retries: int = config.DEFAULT_WAIT_RETRIES,
     delay_s: int = config.DEFAULT_WAIT_DELAY_S,
     node_names: Mapping[str, str] = {},
+    output: bool = True,
 ):
     """
     Validates that the K8s node is in Ready state.
@@ -282,13 +283,15 @@ def wait_until_k8s_ready(
             .until(lambda p: " Ready" in p.stdout.decode())
             .exec(["k8s", "kubectl", "get", "node", node_name, "--no-headers"])
         )
-        LOG.info(f"Kubelet registered successfully on instance '{instance.id}'")
-        LOG.info("%s", result.stdout.decode())
+        if output:
+            LOG.info(f"Kubelet registered successfully on instance '{instance.id}'")
+            LOG.info("%s", result.stdout.decode().strip())
         instance_id_node_name_map[instance.id] = node_name
-    LOG.info(
-        "Successfully checked Kubelet registered on all harness instances: "
-        f"{instance_id_node_name_map}"
-    )
+    if output:
+        LOG.info(
+            "Successfully checked Kubelet registered on all harness instances: "
+            f"{instance_id_node_name_map}"
+        )
 
 
 def wait_for_dns(instance: harness.Instance):

tests/integration/tests/test_version_upgrades.py

@@ -55,7 +55,8 @@ def test_version_upgrades(instances: List[harness.Instance], tmp_path):
 
         # 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()}")
+        latest_version = out.stdout.decode().strip().split("\n")[-1]
+        LOG.info(f"Current snap version: {latest_version}")
 
         # note: the `--classic` flag will be ignored by snapd for strict snaps.
         cp.exec(
@@ -64,3 +65,106 @@ def test_version_upgrades(instances: List[harness.Instance], tmp_path):
         util.wait_until_k8s_ready(cp, instances)
         current_channel = channel
         LOG.info(f"Upgraded {cp.id} on channel {channel}")
+
+
+@pytest.mark.node_count(1)
+@pytest.mark.no_setup()
+@pytest.mark.skipif(
+    not config.VERSION_DOWNGRADE_CHANNELS, reason="No downgrade channels configured"
+)
+@pytest.mark.tags(tags.NIGHTLY)
+def test_version_downgrades_with_rollback(instances: List[harness.Instance], tmp_path):
+    channels = config.VERSION_DOWNGRADE_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_most_stable_channels(
+            int(num_channels),
+            flavour,
+            cp.arch,
+            min_release=config.VERSION_UPGRADE_MIN_RELEASE,
+            reverse=True,
+            include_latest=False,
+        )
+        if len(channels) < 2:
+            pytest.fail(
+                f"Need at least 2 channels to downgrade, got {len(channels)} for flavour {flavour}"
+            )
+        current_channel = channels[0]
+
+    LOG.info(
+        f"Bootstrap node on {current_channel} and downgrade through channels: {channels[1:]}"
+    )
+
+    # Setup the k8s snap from the bootstrap channel and setup basic configuration.
+    util.setup_k8s_snap(cp, tmp_path, current_channel)
+    cp.exec(["k8s", "bootstrap"])
+
+    util.wait_until_k8s_ready(cp, instances)
+    LOG.info(f"Installed {cp.id} on channel {current_channel}")
+
+    # This test will downgrade the snap through the channels, and at each downgrade, attempt a rollback.
+    # Example of downgrading while rolling back through channels:
+    # Channels:  1.32-classic/stable, 1.31-classic/stable
+    # Segment 1: 1.32-classic/stable -> 1.31-classic/stable -> 1.32-classic/stable
+    # Segment 2: 1.31-classic/stable -> 1.30-classic/stable -> 1.31-classic/stable
+
+    for channel in channels[1:]:
+        LOG.info(
+            f">>> Initiating downgrade + rollback segment from {current_channel} → {channel}"
+        )
+        out = cp.exec(["snap", "list", config.SNAP_NAME], capture_output=True)
+        latest_version = out.stdout.decode().strip().split("\n")[-1]
+        LOG.info(f"Current snap version: {latest_version}")
+
+        LOG.info(f"Step 1. Downgrade {cp.id} from {current_channel} → {channel}")
+        # note: the `--classic` flag will be ignored by snapd for strict snaps.
+        cp.exec(
+            ["snap", "refresh", config.SNAP_NAME, "--channel", channel, "--classic"]
+        )
+        util.wait_until_k8s_ready(cp, instances, output=False)
+
+        last_channel = current_channel
+        current_channel = channel
+
+        LOG.info(f"Step 2. Roll back from {current_channel} → {last_channel}")
+        # note: the `--classic` flag will be ignored by snapd for strict snaps.
+        cp.exec(
+            [
+                "snap",
+                "refresh",
+                config.SNAP_NAME,
+                "--channel",
+                last_channel,
+                "--classic",
+            ]
+        )
+        util.wait_until_k8s_ready(cp, instances, output=False)
+
+        LOG.info(
+            f"Step 3. Final downgrade to channel from {last_channel} → {current_channel}"
+        )
+        cp.exec(
+            [
+                "snap",
+                "refresh",
+                config.SNAP_NAME,
+                "--channel",
+                current_channel,
+                "--classic",
+            ]
+        )
+        util.wait_until_k8s_ready(cp, instances, output=False)
+
+        if channel == channels[-1]:
+            LOG.info(">>> Rollback test complete. All downgrade segments verified.")
+        else:
+            LOG.info(
+                ">>> Rollback segment complete. Proceeding to next downgrade segment."
+            )