fix(e2e): use flash undeploy --all --force for cleanup

Author: deanq

tests/e2e/conftest.py

@@ -1,9 +1,8 @@
 """E2E test fixtures: provision real endpoints, configure SDK, clean up."""
 
-import json
 import logging
 import os
-import urllib.request
+import subprocess
 from pathlib import Path
 
 import pytest
@@ -17,48 +16,6 @@
 # Repo root: tests/e2e/conftest.py -> ../../
 _REPO_ROOT = Path(__file__).resolve().parents[2]
 
-_GRAPHQL_URL = "https://api.runpod.io/graphql"
-
-
-def _graphql(api_key: str, query: str, variables: dict | None = None) -> dict:
-    """Execute a Runpod GraphQL query."""
-    payload = json.dumps({"query": query, "variables": variables or {}}).encode()
-    req = urllib.request.Request(
-        f"{_GRAPHQL_URL}?api_key={api_key}",
-        data=payload,
-        headers={"Content-Type": "application/json"},
-    )
-    with urllib.request.urlopen(req, timeout=30) as resp:
-        return json.loads(resp.read())
-
-
-def _delete_endpoints_by_name(api_key: str, names: list[str]) -> None:
-    """Delete endpoints matching the given names via GraphQL API."""
-    result = _graphql(api_key, """
-        query { myself { endpoints { id name } } }
-    """)
-    all_endpoints = result.get("data", {}).get("myself", {}).get("endpoints", [])
-    name_set = set(names)
-    targets = [ep for ep in all_endpoints if ep.get("name") in name_set]
-
-    if not targets:
-        log.warning("No matching endpoints found for names: %s", names)
-        return
-
-    for ep in targets:
-        try:
-            resp = _graphql(
-                api_key,
-                "mutation($id: String!) { deleteEndpoint(id: $id) }",
-                {"id": ep["id"]},
-            )
-            if "errors" in resp:
-                log.warning("Failed to delete %s (%s): %s", ep["name"], ep["id"], resp["errors"])
-            else:
-                log.info("Deleted endpoint %s (%s)", ep["name"], ep["id"])
-        except Exception:
-            log.exception("Error deleting endpoint %s (%s)", ep["name"], ep["id"])
-
 
 @pytest.fixture(scope="session", autouse=True)
 def verify_local_runpod():
@@ -100,13 +57,17 @@ def endpoints(require_api_key, test_cases):
         log.info("Endpoint ready: name=%s image=%s template.dockerArgs=%s", ep.name, ep.image, ep.template.dockerArgs if ep.template else "N/A")
     yield eps
 
-    # Delete provisioned endpoints via GraphQL API directly.
-    # flash undeploy relies on .runpod/resources.pkl which doesn't exist in CI.
-    api_key = os.environ.get("RUNPOD_API_KEY", "")
-    endpoint_names = [ep.name for ep in eps.values()]
-    log.info("Cleaning up %d provisioned endpoints: %s", len(endpoint_names), endpoint_names)
-    if api_key and endpoint_names:
-        try:
-            _delete_endpoints_by_name(api_key, endpoint_names)
-        except Exception:
-            log.exception("Endpoint cleanup failed (API key may lack GraphQL access)")
+    log.info("Cleaning up all provisioned endpoints")
+    try:
+        result = subprocess.run(
+            ["flash", "undeploy", "--all", "--force"],
+            capture_output=True,
+            text=True,
+            timeout=120,
+        )
+        if result.returncode == 0:
+            log.info("Undeployed all endpoints")
+        else:
+            log.warning("flash undeploy --all --force failed (rc=%d): %s", result.returncode, result.stderr)
+    except Exception:
+        log.exception("Failed to undeploy endpoints")