feat: TF-specific assertions

terraform/cos/locals.tf

@@ -1,3 +1,4 @@
+# TODO: Remove this when done testing
 locals {
   clouds                = ["aws", "self-managed"] # list of k8s clouds where this COS module can be deployed.
   tls_termination       = var.external_certificates_offer_url != null ? true : false

tests/integration/cos/tls_external/test_upgrade_cos_tls_external.py

@@ -38,10 +38,17 @@ def test_deploy_from_track_2(
 def test_deploy_to_track_dev(
     tmp_path, tf_manager, ca_model: jubilant.Juju, cos_model: jubilant.Juju
 ):
-    # WHEN upgraded to track dev
+    jubilant_args = {
+        "cos_model": cos_model.model,
+        "ca_model": ca_model.model,
+        **S3_ENDPOINT,
+    }
+    # WHEN planning to upgrade to track dev
     tf_manager.init(TRACK_DEV_TF_FILE)
-    tf_manager.apply(ca_model=ca_model.model, cos_model=cos_model.model, **S3_ENDPOINT)
-
+    # THEN only expected apps are replaced
+    tf_manager.apps_to_replace(["grafana"], **jubilant_args)
+    # AND WHEN upgraded to track dev
+    tf_manager.apply(**jubilant_args)
     # THEN the model is upgraded and is healthy
     generic_assertions(cos_model, ca_model, tmp_path)
     no_errors_in_otelcol_logs(cos_model)

tests/integration/cos/tls_full/test_upgrade_cos_tls_full.py

@@ -38,10 +38,17 @@ def test_deploy_from_track_2(
 def test_deploy_to_track_dev(
     tmp_path, tf_manager, ca_model: jubilant.Juju, cos_model: jubilant.Juju
 ):
-    # WHEN upgraded to track dev
+    jubilant_args = {
+        "cos_model": cos_model.model,
+        "ca_model": ca_model.model,
+        **S3_ENDPOINT,
+    }
+    # WHEN planning to upgrade to track dev
     tf_manager.init(TRACK_DEV_TF_FILE)
-    tf_manager.apply(ca_model=ca_model.model, cos_model=cos_model.model, **S3_ENDPOINT)
-
+    # THEN only expected apps are replaced
+    tf_manager.apps_to_replace(["grafana"], **jubilant_args)
+    # AND WHEN upgraded to track dev
+    tf_manager.apply(**jubilant_args)
     # THEN the model is upgraded and is healthy
     generic_assertions(cos_model, ca_model, tmp_path)
     no_errors_in_otelcol_logs(cos_model)

tests/integration/cos/tls_internal/test_upgrade_cos_tls_internal.py

@@ -34,10 +34,13 @@ def test_deploy_from_track_2(tf_manager, cos_model: jubilant.Juju):
 
 
 def test_deploy_to_track_dev(tf_manager, cos_model: jubilant.Juju):
-    # WHEN upgraded to track dev
+    jubilant_args = {"cos_model": cos_model.model}
+    # WHEN planning to upgrade to track dev
     tf_manager.init(TRACK_DEV_TF_FILE)
-    tf_manager.apply(model=cos_model.model, **S3_ENDPOINT)
-
+    # THEN only expected apps are replaced
+    tf_manager.apps_to_replace(["grafana"], **jubilant_args)
+    # AND WHEN upgraded to track dev
+    tf_manager.apply(**jubilant_args)
     # THEN the model is upgraded and is healthy
     generic_assertions(cos_model)
     no_errors_in_otelcol_logs(cos_model)

tests/integration/cos/tls_none/test_upgrade_cos_tls_none.py

@@ -34,10 +34,13 @@ def test_deploy_from_track_2(tf_manager, cos_model: jubilant.Juju):
 
 
 def test_deploy_to_track_dev(tf_manager, cos_model: jubilant.Juju):
-    # WHEN upgraded to track dev
+    jubilant_args = {"cos_model": cos_model.model}
+    # WHEN planning to upgrade to track dev
     tf_manager.init(TRACK_DEV_TF_FILE)
-    tf_manager.apply(model=cos_model.model, **S3_ENDPOINT)
-
+    # THEN only expected apps are replaced
+    tf_manager.apps_to_replace(["grafana"], **jubilant_args)
+    # AND WHEN upgraded to track dev
+    tf_manager.apply(**jubilant_args)
     # THEN the model is upgraded and is healthy
     generic_assertions(cos_model)
     no_errors_in_otelcol_logs(cos_model)

tests/integration/cos_dev/tls_external/test_cos_dev_tls_external.py

@@ -7,20 +7,15 @@
 from pathlib import Path
 
 import jubilant
-from helpers import (
-    catalogue_apps_are_reachable,
-    get_tls_context,
-    no_errors_in_otelcol_logs,
-    wait_for_active_idle_without_error,
-)
+from helpers import generic_assertions, no_errors_in_otelcol_logs
 
 TRACK_DEV_TF_FILE = Path(__file__).parent.resolve() / "track-dev.tf"
 
 
-def test_deploy(tmp_path, tf_manager, ca_model: jubilant.Juju, cos_model: jubilant.Juju):
+def test_deploy(
+    tmp_path, tf_manager, ca_model: jubilant.Juju, cos_model: jubilant.Juju
+):
     tf_manager.init(TRACK_DEV_TF_FILE)
     tf_manager.apply(ca_model=ca_model.model, cos_model=cos_model.model)
-    wait_for_active_idle_without_error([ca_model, cos_model])
-    tls_ctx = get_tls_context(tmp_path, ca_model, "self-signed-certificates")
-    catalogue_apps_are_reachable(cos_model, tls_ctx)
+    generic_assertions(cos_model, ca_model, tmp_path)
     no_errors_in_otelcol_logs(cos_model)

tests/integration/cos_dev/tls_full/test_cos_dev_tls_full.py

@@ -7,20 +7,15 @@
 from pathlib import Path
 
 import jubilant
-from helpers import (
-    catalogue_apps_are_reachable,
-    get_tls_context,
-    no_errors_in_otelcol_logs,
-    wait_for_active_idle_without_error,
-)
+from helpers import generic_assertions, no_errors_in_otelcol_logs
 
 TRACK_DEV_TF_FILE = Path(__file__).parent.resolve() / "track-dev.tf"
 
 
-def test_deploy(tmp_path, tf_manager, ca_model: jubilant.Juju, cos_model: jubilant.Juju):
+def test_deploy(
+    tmp_path, tf_manager, ca_model: jubilant.Juju, cos_model: jubilant.Juju
+):
     tf_manager.init(TRACK_DEV_TF_FILE)
     tf_manager.apply(ca_model=ca_model.model, cos_model=cos_model.model)
-    wait_for_active_idle_without_error([ca_model, cos_model])
-    tls_ctx = get_tls_context(tmp_path, ca_model, "self-signed-certificates")
-    catalogue_apps_are_reachable(cos_model, tls_ctx)
+    generic_assertions(cos_model, ca_model, tmp_path)
     no_errors_in_otelcol_logs(cos_model)

tests/integration/cos_dev/tls_internal/test_cos_dev_tls_internal.py

@@ -6,18 +6,13 @@
 from pathlib import Path
 
 import jubilant
-from helpers import (
-    catalogue_apps_are_reachable,
-    no_errors_in_otelcol_logs,
-    wait_for_active_idle_without_error,
-)
+from helpers import generic_assertions, no_errors_in_otelcol_logs
 
 TRACK_DEV_TF_FILE = Path(__file__).parent.resolve() / "track-dev.tf"
 
 
 def test_deploy(tf_manager, cos_model: jubilant.Juju):
     tf_manager.init(TRACK_DEV_TF_FILE)
     tf_manager.apply(model=cos_model.model)
-    wait_for_active_idle_without_error([cos_model])
-    catalogue_apps_are_reachable(cos_model)
+    generic_assertions(cos_model)
     no_errors_in_otelcol_logs(cos_model)

tests/integration/cos_dev/tls_none/test_cos_dev_tls_none.py

@@ -6,18 +6,13 @@
 from pathlib import Path
 
 import jubilant
-from helpers import (
-    catalogue_apps_are_reachable,
-    no_errors_in_otelcol_logs,
-    wait_for_active_idle_without_error,
-)
+from helpers import generic_assertions, no_errors_in_otelcol_logs
 
 TRACK_DEV_TF_FILE = Path(__file__).parent.resolve() / "track-dev.tf"
 
 
 def test_deploy(tf_manager, cos_model: jubilant.Juju):
     tf_manager.init(TRACK_DEV_TF_FILE)
     tf_manager.apply(model=cos_model.model)
-    wait_for_active_idle_without_error([cos_model])
-    catalogue_apps_are_reachable(cos_model)
+    generic_assertions(cos_model)
     no_errors_in_otelcol_logs(cos_model)

tests/integration/cos_lite/tls_external/test_upgrade_cos_lite_tls_external.py

@@ -25,9 +25,12 @@ def test_deploy_from_track_2(
 def test_deploy_to_track_dev(
     tmp_path, tf_manager, ca_model: jubilant.Juju, cos_model: jubilant.Juju
 ):
-    # WHEN upgraded to track dev
+    jubilant_args = {"cos_model": cos_model.model, "ca_model": ca_model.model}
+    # WHEN planning to upgrade to track dev
     tf_manager.init(TRACK_DEV_TF_FILE)
-    tf_manager.apply(ca_model=ca_model.model, cos_model=cos_model.model)
-
+    # THEN only expected apps are replaced
+    tf_manager.apps_to_replace(["grafana"], **jubilant_args)
+    # AND WHEN upgraded to track dev
+    tf_manager.apply(**jubilant_args)
     # THEN the model is upgraded and is healthy
     generic_assertions(cos_model, ca_model, tmp_path)

tests/integration/cos_lite/tls_full/test_upgrade_cos_lite_tls_full.py

@@ -25,9 +25,12 @@ def test_deploy_from_track_2(
 def test_deploy_to_track_dev(
     tmp_path, tf_manager, ca_model: jubilant.Juju, cos_model: jubilant.Juju
 ):
-    # WHEN upgraded to track dev
+    jubilant_args = {"cos_model": cos_model.model, "ca_model": ca_model.model}
+    # WHEN planning to upgrade to track dev
     tf_manager.init(TRACK_DEV_TF_FILE)
-    tf_manager.apply(ca_model=ca_model.model, cos_model=cos_model.model)
-
+    # THEN only expected apps are replaced
+    tf_manager.apps_to_replace(["grafana"], **jubilant_args)
+    # AND WHEN upgraded to track dev
+    tf_manager.apply(**jubilant_args)
     # THEN the model is upgraded and is healthy
     generic_assertions(cos_model, ca_model, tmp_path)

tests/integration/cos_lite/tls_internal/test_upgrade_cos_lite_tls_internal.py

@@ -21,9 +21,12 @@ def test_deploy_from_track_2(tf_manager, cos_model: jubilant.Juju):
 
 
 def test_deploy_to_track_dev(tf_manager, cos_model: jubilant.Juju):
-    # WHEN upgraded to track dev
+    jubilant_args = {"cos_model": cos_model.model}
+    # WHEN planning to upgrade to track dev
     tf_manager.init(TRACK_DEV_TF_FILE)
-    tf_manager.apply(model=cos_model.model)
-
+    # THEN only expected apps are replaced
+    tf_manager.apps_to_replace(["grafana"], **jubilant_args)
+    # AND WHEN upgraded to track dev
+    tf_manager.apply(**jubilant_args)
     # THEN the model is upgraded and is healthy
     generic_assertions(cos_model)

tests/integration/cos_lite/tls_none/test_upgrade_cos_lite_tls_none.py

@@ -21,9 +21,12 @@ def test_deploy_from_track_2(tf_manager, cos_model: jubilant.Juju):
 
 
 def test_deploy_to_track_dev(tf_manager, cos_model: jubilant.Juju):
-    # WHEN upgraded to track dev
+    jubilant_args = {"cos_model": cos_model.model}
+    # WHEN planning to upgrade to track dev
     tf_manager.init(TRACK_DEV_TF_FILE)
-    tf_manager.apply(model=cos_model.model)
-
+    # THEN only expected apps are replaced
+    tf_manager.apps_to_replace(["grafana"], **jubilant_args)
+    # AND WHEN upgraded to track dev
+    tf_manager.apply(**jubilant_args)
     # THEN the model is upgraded and is healthy
     generic_assertions(cos_model)

tests/integration/helpers.py

@@ -32,14 +32,55 @@ def init(self, tf_file: str):
     def _args_str(target: Optional[str] = None, **kwargs) -> str:
         target_arg = f"-target module.{target}" if target else ""
         var_args = " ".join(f"-var {k}={v}" for k, v in kwargs.items())
-        return "-auto-approve " + f"{target_arg} " + var_args
+        return f"{target_arg} " + var_args
+
+    def plan_json(self, **kwargs) -> dict:
+        """Run terraform plan and return the JSON representation."""
+        plan_file = os.path.join(self.dir, "tfplan")
+        extras = self._args_str(**kwargs)
+        plan_cmd = f"{self.tf_cmd} plan -out={plan_file} " + extras
+        subprocess.run(shlex.split(plan_cmd), check=True)
+        show_cmd = f"{self.tf_cmd} show -json {plan_file}"
+        result = subprocess.run(
+            shlex.split(show_cmd), check=True, capture_output=True, text=True
+        )
+        return json.loads(result.stdout)
+
+    def plan_has_changes(self, **kwargs) -> bool:
+        """Return True if terraform plan detects pending changes.
+
+        Uses -detailed-exitcode: exit 0 means no changes, exit 2 means changes pending.
+        """
+        extras = self._args_str(**kwargs)
+        plan_cmd = f"{self.tf_cmd} plan -detailed-exitcode " + extras
+        result = subprocess.run(shlex.split(plan_cmd))
+        if result.returncode == 0:
+            return False
+        if result.returncode == 2:
+            return True
+        # Exit code 1 means error
+        raise subprocess.CalledProcessError(result.returncode, plan_cmd)
+
+    def apps_to_replace(self, destroys: List[str], **kwargs):
+        plan = self.plan_json(**kwargs)
+        destroy_plan_apps = [
+            rc["address"].split(".")[-1]
+            for rc in plan.get("resource_changes", [])
+            if "delete" in rc["change"]["actions"]
+            and rc.get("type", "") == "juju_application"
+        ]
+        assert set(destroys) == set(destroy_plan_apps), (
+            f"Expected destroys: {destroys}, got: {destroy_plan_apps}"
+        )
 
     def apply(self, target: Optional[str] = None, **kwargs):
-        cmd_str = f"{self.tf_cmd} apply " + self._args_str(target, **kwargs)
+        extras = self._args_str(target, **kwargs)
+        cmd_str = f"{self.tf_cmd} apply -auto-approve " + extras
         subprocess.run(shlex.split(cmd_str), check=True)
 
     def destroy(self, **kwargs):
-        cmd_str = f"{self.tf_cmd} destroy " + self._args_str(None, **kwargs)
+        extras = self._args_str(**kwargs)
+        cmd_str = f"{self.tf_cmd} destroy -auto-approve " + extras
         subprocess.run(shlex.split(cmd_str), check=True)