test(iris): local K8s e2e with auto-provisioned kind cluster (#3097)

## Summary

- Add `k8s_cluster` (session-scoped) and `k8s_runtime` (per-test)
fixtures to `conftest.py` — reusable by any e2e test
- `k8s_cluster` auto-creates a kind cluster if no K8s cluster is
reachable, tears it down after the session
- `k8s_runtime` creates an ephemeral namespace per test for isolation
- Add `test_kubernetes_runtime_lifecycle` — validates the full
`KubernetesRuntime` lifecycle (create pod, run, succeed, read logs)
against real K8s
- Marked `slow` + `e2e` so default CI is unaffected
- Update `TESTING.md` with the run command

### Motivation

Relates to #2944, #3062.

The CW canary ferry is unreliable for validating K8s runtime behavior
(RBAC issues, Connection refused — see #3091). This test validates
`KubernetesRuntime` against any local cluster in seconds.

### How to use

```bash
# Just run it — kind cluster is auto-created if needed (requires: brew install kind)
uv run pytest lib/iris/tests/e2e/test_coreweave_live_kubernetes_runtime.py \
  -m slow -k kubernetes -v -o "addopts="

# Or bring your own cluster (kind, k3d, minikube, remote)
kind create cluster --name iris-test
uv run pytest lib/iris/tests/e2e/test_coreweave_live_kubernetes_runtime.py \
  -m slow -k kubernetes -v -o "addopts="
```

## Test plan

- [x] Test passes with auto-created kind cluster (~49s including cluster
create/teardown)
- [x] Test passes with pre-existing kind cluster (~14s)
- [x] Default `uv run pytest` skips the test (marked `slow`)
- [x] Fixture auto-skips when no cluster reachable and kind not
installed
- [x] Pre-commit passes

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: yonromai

lib/iris/TESTING.md

@@ -161,4 +161,8 @@ IRIS_SCREENSHOT_DIR=/tmp/shots uv run pytest lib/iris/tests/e2e/test_dashboard.p
 
 # When modifying the dashboard
 uv run pytest lib/iris/tests/e2e/test_dashboard.py -x -o "addopts="
+
+# K8s runtime tests (requires a running cluster — kind, k3d, minikube, etc.)
+uv run pytest lib/iris/tests/e2e/test_coreweave_live_kubernetes_runtime.py \
+  -m slow -k lifecycle -v
 ```

lib/iris/tests/e2e/conftest.py

@@ -13,12 +13,16 @@
 """
 
 import os
+import shutil
+import subprocess
 import time
+import uuid
 from dataclasses import dataclass
 from pathlib import Path
 
 import pytest
 from iris.chaos import reset_chaos
+from iris.cluster.runtime.kubernetes import KubernetesRuntime
 from iris.client.client import IrisClient, Job
 from iris.cluster.config import load_config, make_local_config
 from iris.cluster.manager import connect_cluster
@@ -420,3 +424,83 @@ def capture(label: str) -> Path:
         return path
 
     return capture
+
+
+# ---------------------------------------------------------------------------
+# Kubernetes fixtures (for tests against real K8s: kind, k3d, minikube, etc.)
+# ---------------------------------------------------------------------------
+
+KIND_CLUSTER_NAME = "iris-test"
+
+
+def _cluster_reachable() -> bool:
+    try:
+        result = subprocess.run(["kubectl", "cluster-info"], capture_output=True, timeout=10)
+        return result.returncode == 0
+    except (subprocess.TimeoutExpired, FileNotFoundError):
+        return False
+
+
+@pytest.fixture(scope="session")
+def k8s_cluster():
+    """Ensure a K8s cluster is available for the test session.
+
+    If a cluster is already reachable, uses it as-is. Otherwise, creates a
+    kind cluster and tears it down at the end of the session.
+    """
+    if shutil.which("kubectl") is None:
+        pytest.skip("kubectl not in PATH (install: brew install kubectl)")
+
+    if _cluster_reachable():
+        yield
+        return
+
+    if shutil.which("kind") is None:
+        pytest.skip("no reachable K8s cluster and kind not in PATH (install: brew install kind)")
+
+    subprocess.run(
+        ["kind", "create", "cluster", "--name", KIND_CLUSTER_NAME],
+        check=True,
+        timeout=120,
+    )
+    try:
+        yield
+    finally:
+        subprocess.run(
+            ["kind", "delete", "cluster", "--name", KIND_CLUSTER_NAME],
+            capture_output=True,
+            timeout=60,
+        )
+
+
+@pytest.fixture
+def k8s_runtime(k8s_cluster):
+    """KubernetesRuntime with an ephemeral namespace, torn down after each test."""
+    namespace = f"iris-test-{uuid.uuid4().hex[:8]}"
+    subprocess.run(
+        ["kubectl", "create", "namespace", namespace],
+        check=True,
+        capture_output=True,
+    )
+    # Wait for K8s to provision the default ServiceAccount in the new namespace.
+    # Without this, pod creation fails with "serviceaccount default not found".
+    deadline = time.monotonic() + 30
+    while time.monotonic() < deadline:
+        result = subprocess.run(
+            ["kubectl", "-n", namespace, "get", "serviceaccount", "default"],
+            capture_output=True,
+        )
+        if result.returncode == 0:
+            break
+        time.sleep(0.5)
+    else:
+        pytest.skip(f"default ServiceAccount not ready in namespace {namespace} after 30s")
+    runtime = KubernetesRuntime(namespace=namespace)
+    try:
+        yield runtime
+    finally:
+        runtime.cleanup()
+        subprocess.run(
+            ["kubectl", "delete", "namespace", namespace, "--ignore-not-found"],
+            capture_output=True,
+        )

lib/iris/tests/e2e/test_coreweave_live_kubernetes_runtime.py

@@ -8,7 +8,6 @@
 import io
 import os
 import posixpath
-import shutil
 import time
 import uuid
 from contextlib import contextmanager
@@ -162,7 +161,29 @@ def _coreweave_upload_env(config) -> object:
             fsspec.config.conf.pop("s3", None)
 
 
-@pytest.mark.skipif(shutil.which("kubectl") is None, reason="kubectl is not available")
+@pytest.mark.timeout(120)
+def test_kubernetes_runtime_lifecycle(k8s_runtime: KubernetesRuntime):
+    """Full KubernetesRuntime lifecycle: create pod, run, succeed, read logs."""
+    run_id = uuid.uuid4().hex[:8]
+    config = ContainerConfig(
+        image="python:3.11-slim",
+        entrypoint=_entrypoint(["bash", "-c", "echo lifecycle-test-ok && sleep 2"]),
+        env={},
+        workdir="/app",
+        task_id=f"lifecycle-{run_id}",
+        resources=cluster_pb2.ResourceSpecProto(cpu_millicores=100, memory_bytes=64 * 1024**2),
+    )
+
+    handle = k8s_runtime.create_container(config)
+    handle.run()
+
+    state = _wait_finished(handle, timeout_seconds=60)
+    assert state == cluster_pb2.TASK_STATE_SUCCEEDED
+
+    logs = handle.log_reader().read_all()
+    assert any("lifecycle-test-ok" in line.data for line in logs)
+
+
 @pytest.mark.timeout(1800)
 def test_coreweave_kubernetes_runtime_cpu_job_live(coreweave_runtime: KubernetesRuntime):
     """CPU pod should extract bundle and complete successfully via KubernetesRuntime."""
@@ -225,7 +246,6 @@ def test_coreweave_kubernetes_runtime_cpu_job_live(coreweave_runtime: Kubernetes
                 pass
 
 
-@pytest.mark.skipif(shutil.which("kubectl") is None, reason="kubectl is not available")
 @pytest.mark.timeout(1800)
 def test_incremental_log_reader_no_duplicates(coreweave_runtime: KubernetesRuntime):
     """Incremental log reads via byte-offset cursor must not produce duplicate lines.
@@ -286,7 +306,6 @@ def test_incremental_log_reader_no_duplicates(coreweave_runtime: KubernetesRunti
     assert numbered == expected
 
 
-@pytest.mark.skipif(shutil.which("kubectl") is None, reason="kubectl is not available")
 @pytest.mark.timeout(3600)
 def test_coreweave_kubernetes_runtime_gpu_job_live(coreweave_runtime: KubernetesRuntime):
     """GPU pod should request GPU and prove device access via nvidia-smi."""
@@ -327,7 +346,6 @@ def test_coreweave_kubernetes_runtime_gpu_job_live(coreweave_runtime: Kubernetes
     assert gpu_state == cluster_pb2.TASK_STATE_SUCCEEDED, f"gpu pod failed logs={gpu_logs}"
 
 
-@pytest.mark.skipif(shutil.which("kubectl") is None, reason="kubectl is not available")
 @pytest.mark.timeout(600)
 def test_tensorstore_s3_roundtrip():
     """Verify tensorstore can write and read zarr3 data via S3-compatible storage.