OSAC-1586: add E2E test for cluster deletion feedback

tests/caas/test_cluster_create.py

@@ -1,10 +1,14 @@
 from __future__ import annotations
 
+import contextlib
+import subprocess
 from pathlib import Path
 
 from tests.core.grpc_client import GRPCClient
 from tests.core.helpers import (
+    wait_for_cluster_deleting,
     wait_for_cluster_deletion,
+    wait_for_cluster_grpc_deleting_or_archived,
     wait_for_cluster_grpc_removal,
     wait_for_cluster_order_cr,
     wait_for_cluster_ready,
@@ -26,11 +30,20 @@ def test_cluster_create(
         template_parameter_files={"pull_secret": pull_secret_path},
         template_parameters={"ssh_public_key": Path(ssh_public_key_path).read_text().strip()},
     )
-    co_name = wait_for_cluster_order_cr(k8s=k8s_hub_client, uuid=uuid)
-    assert uuid in grpc.list_cluster_ids()
 
-    wait_for_cluster_ready(k8s=k8s_hub_client, name=co_name)
+    try:
+        co_name = wait_for_cluster_order_cr(k8s=k8s_hub_client, uuid=uuid)
+        assert uuid in grpc.list_cluster_ids()
 
-    cli.delete_cluster(uuid=uuid)
-    wait_for_cluster_deletion(k8s=k8s_hub_client, name=co_name)
-    wait_for_cluster_grpc_removal(grpc=grpc, uuid=uuid)
+        wait_for_cluster_ready(k8s=k8s_hub_client, name=co_name)
+
+        cli.delete_cluster(uuid=uuid)
+
+        wait_for_cluster_deleting(k8s=k8s_hub_client, name=co_name)
+        wait_for_cluster_grpc_deleting_or_archived(grpc=grpc, uuid=uuid)
+
+        wait_for_cluster_deletion(k8s=k8s_hub_client, name=co_name)
+        wait_for_cluster_grpc_removal(grpc=grpc, uuid=uuid)
+    finally:
+        with contextlib.suppress(subprocess.CalledProcessError):
+            cli.delete_cluster(uuid=uuid)

tests/caas/test_cluster_delete_feedback_light.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import contextlib
+import subprocess
+from pathlib import Path
+
+from tests.core.grpc_client import GRPCClient
+from tests.core.helpers import (
+    wait_for_cluster_deleting,
+    wait_for_cluster_deletion,
+    wait_for_cluster_grpc_deleting_or_archived,
+    wait_for_cluster_grpc_removal,
+    wait_for_cluster_order_cr,
+    wait_for_cluster_progressing,
+)
+from tests.core.k8s_client import K8sClient
+from tests.core.osac_cli import OsacCLI
+
+
+def test_cluster_delete_reports_deleting_state_without_provisioning(
+    cli: OsacCLI,
+    grpc: GRPCClient,
+    k8s_hub_client: K8sClient,
+    cluster_template: str,
+    pull_secret_path: str,
+    ssh_public_key_path: str,
+) -> None:
+    """Verify that cluster deletion transitions through DELETING state
+    without waiting for full provisioning. Runs on kind without HyperShift
+    (OSAC-1586)."""
+
+    uuid = cli.create_cluster(
+        template=cluster_template,
+        template_parameter_files={"pull_secret": pull_secret_path},
+        template_parameters={"ssh_public_key": Path(ssh_public_key_path).read_text().strip()},
+    )
+
+    try:
+        co_name = wait_for_cluster_order_cr(k8s=k8s_hub_client, uuid=uuid)
+        assert uuid in grpc.list_cluster_ids()
+
+        wait_for_cluster_progressing(k8s=k8s_hub_client, name=co_name)
+
+        cli.delete_cluster(uuid=uuid)
+
+        wait_for_cluster_deleting(k8s=k8s_hub_client, name=co_name)
+        wait_for_cluster_grpc_deleting_or_archived(grpc=grpc, uuid=uuid)
+        wait_for_cluster_deletion(k8s=k8s_hub_client, name=co_name)
+        wait_for_cluster_grpc_removal(grpc=grpc, uuid=uuid)
+    finally:
+        with contextlib.suppress(subprocess.CalledProcessError):
+            cli.delete_cluster(uuid=uuid)

tests/core/helpers.py

@@ -263,6 +263,16 @@ def wait_for_cluster_order_cr(*, k8s: K8sClient, uuid: str) -> str:
     )
 
 
+def wait_for_cluster_progressing(*, k8s: K8sClient, name: str) -> None:
+    poll_until(
+        fn=lambda: k8s.get_cluster_order_phase(name=name, checked=False),
+        until=lambda v: v == "Progressing",
+        retries=30,
+        delay=2,
+        description=f"{name} ClusterOrder Progressing phase",
+    )
+
+
 def wait_for_cluster_ready(*, k8s: K8sClient, name: str) -> None:
     # Must stay safely above osac-aap's own wait_for_clusteroperators_retries
     # budget (60 min) plus earlier steps in the same AAP job (create hosted
@@ -391,6 +401,41 @@ def _force_cleanup_machine_preterminate_hooks(*, k8s: K8sClient, name: str) -> N
         run_unchecked(*base_args, "annotate", f"machines.cluster.x-k8s.io/{machine_name}", "-n", cp_ns, f"{hook}-")
 
 
+def wait_for_cluster_deleting(*, k8s: K8sClient, name: str) -> None:
+    poll_until(
+        fn=lambda: k8s.get_cluster_order_phase(name=name, checked=False),
+        until=lambda v: v == "Deleting",
+        retries=30,
+        delay=5,
+        description=f"{name} ClusterOrder Deleting phase",
+    )
+
+
+def wait_for_cluster_grpc_deleting_or_archived(*, grpc: GRPCClient, uuid: str) -> None:
+    """Succeed if we catch CLUSTER_STATE_DELETING or if the cluster is already archived.
+
+    The DELETING window in the fulfillment-service is extremely short (one
+    Update + Signal round-trip).  Polling for the exact state is racey; accepting
+    either DELETING or 'already gone' makes the assertion reliable.
+    """
+
+    def _done() -> bool:
+        try:
+            cluster = grpc.get_cluster(cluster_id=uuid)
+            state = cluster.get("object", {}).get("status", {}).get("state", "")
+            return state == "CLUSTER_STATE_DELETING"
+        except subprocess.CalledProcessError:
+            return True
+
+    poll_until(
+        fn=_done,
+        until=lambda v: v is True,
+        retries=30,
+        delay=2,
+        description=f"{uuid} gRPC DELETING or already archived",
+    )
+
+
 def wait_for_cluster_grpc_removal(*, grpc: GRPCClient, uuid: str) -> None:
     # retry_on_error=True: a flaky grpcurl call hitting a momentarily-busy
     # route right after heavy cluster-deletion activity shouldn't fail the