feat(k8s): verify the tenant can override the GPU Operator driver (K8S25-02)

docs/test-plan.adoc

@@ -4021,7 +4021,7 @@ a|
 | [[K8S25-02]]K8S25-02
 | K8S25
 | https://github.com/NVIDIA/ai-cloud-validation/issues/220[#220]
-| min_req
+| kubernetes, min_req
 | both
 |
 | P0

docs/test-plan.yaml

@@ -3626,6 +3626,7 @@ domains:
                 milestone: ""
               - summary: Verify provider-default accelerator operators and drivers can be replaced or overridden with tenant-required versions
                 labels:
+                  - kubernetes
                   - min_req
                 notes: ""
                 req_id: K8S25

isvctl/configs/suites/k8s.yaml

@@ -197,6 +197,11 @@ tests:
           test_id: "K8S25-01"
           labels: ["kubernetes", "min_req"]
           namespace: "{{ steps.setup.kubernetes.gpu_operator_namespace | default('nvidia-gpu-operator') }}"
+        K8sGpuOperatorOverrideCheck:
+          test_id: "K8S25-02"
+          labels: ["kubernetes", "min_req"]
+          namespace: "{{ steps.setup.kubernetes.gpu_operator_namespace | default('nvidia-gpu-operator') }}"
+          driver_version: "{{ steps.setup.kubernetes.driver_version | default('580.82.07') }}"
         K8sGpuLabelsCheck:
           test_id: "N/A"
           labels: ["kubernetes", "gpu"]

isvtest/src/isvtest/core/k8s.py

@@ -362,6 +362,28 @@ def kubectl_items_or_empty(
         return []
 
 
+def command_detail(result: KubectlJsonResult) -> str:
+    """Return the most informative detail from a failed kubectl invocation."""
+    exit_code = getattr(result, "exit_code", None)
+    if exit_code is None:
+        exit_code = getattr(result, "returncode", None)
+    return (result.stderr or "").strip() or (result.stdout or "").strip() or f"exit {exit_code}"
+
+
+def is_resource_absent(stderr: str) -> bool:
+    """Return True when kubectl reports the resource type or object is simply not there.
+
+    Covers both spellings: an API group the cluster does not serve (no CRD
+    installed) and a NotFound for an object within a served group.
+    """
+    lowered = (stderr or "").lower()
+    return (
+        "doesn't have a resource type" in lowered
+        or "could not find the requested resource" in lowered
+        or "notfound" in lowered.replace(" ", "")
+    )
+
+
 def names_from_items(items: list[dict[str, Any]]) -> list[str]:
     """Extract ``.metadata.name`` values from a list of Kubernetes API objects."""
     names: list[str] = []

isvtest/src/isvtest/validations/k8s_api_network_acl.py

@@ -23,7 +23,7 @@
 
 import pytest
 
-from isvtest.core.k8s import KubectlParseError, get_kubectl_base_shell, parse_kubectl_json
+from isvtest.core.k8s import KubectlParseError, command_detail, get_kubectl_base_shell, parse_kubectl_json
 from isvtest.core.validation import BaseValidation
 from isvtest.utils.checks import truncate
 
@@ -304,7 +304,7 @@ def _run_authorized_probe(self, authorized_probe_cmd: str, probe_timeout_s: int)
         """
         result = self.run_command(authorized_probe_cmd, timeout=probe_timeout_s)
         if result.exit_code != 0:
-            detail = result.stderr.strip() or result.stdout.strip() or f"exit code {result.exit_code}"
+            detail = command_detail(result)
             snippet = truncate(authorized_probe_cmd)
             self.set_failed(
                 f"Authorized probe failed (cmd: {snippet}): {detail}. A failing "
@@ -337,7 +337,7 @@ def _run_unauthorized_probe(
         # found". Treating them as an ACL-enforced pass would hide a broken
         # probe and yield false assurance.
         if result.exit_code in (126, 127):
-            detail = result.stderr.strip() or result.stdout.strip() or f"exit code {result.exit_code}"
+            detail = command_detail(result)
             self.set_failed(
                 f"Unauthorized probe could not execute (cmd: {snippet}): "
                 f"{detail}. Fix the probe tooling/command and re-run.{targets}"

isvtest/src/isvtest/validations/k8s_autoscaler.py

@@ -25,6 +25,7 @@
 
 from isvtest.core.k8s import (
     KubectlParseError,
+    command_detail,
     get_kubectl_base_shell,
     parse_kubectl_json,
     parse_kubectl_json_items,
@@ -164,8 +165,7 @@ def _run_provider_managed(self) -> None:
 
         result = self.run_command(command)
         if result.exit_code != 0:
-            detail = (result.stderr or result.stdout or f"exit {result.exit_code}").strip()
-            self.set_failed(f"Provider-managed autoscaler command failed: {detail}")
+            self.set_failed(f"Provider-managed autoscaler command failed: {command_detail(result)}")
             return
 
         try:
@@ -367,5 +367,4 @@ def _is_not_found(stderr: str) -> bool:
 
 def _format_error(scope: str, result: CommandResult) -> str:
     """Format a concise kubectl error."""
-    detail = (result.stderr or result.stdout or "").strip()
-    return f"Failed to get {scope}: {detail or f'exit {result.exit_code}'}"
+    return f"Failed to get {scope}: {command_detail(result)}"

isvtest/src/isvtest/validations/k8s_control_plane_logs.py

@@ -18,7 +18,12 @@
 import shlex
 from typing import Any, ClassVar
 
-from isvtest.core.k8s import KubectlParseError, get_kubectl_base_shell, parse_kubectl_json_items
+from isvtest.core.k8s import (
+    KubectlParseError,
+    command_detail,
+    get_kubectl_base_shell,
+    parse_kubectl_json_items,
+)
 from isvtest.core.validation import BaseValidation
 from isvtest.utils.checks import truncate
 
@@ -300,7 +305,7 @@ def _execute_plan(
 
             result = self.run_command(cmd)
             if result.exit_code != 0:
-                detail = result.stderr.strip() or result.stdout.strip() or f"exit code {result.exit_code}"
+                detail = command_detail(result)
                 if path == "kubectl":
                     failures.append(f"{label}: kubectl logs failed: {detail}")
                 else:
@@ -338,7 +343,7 @@ def _find_component_pods(self, namespace: str, components: list[str]) -> tuple[d
         cmd = f"{kubectl_base} get pods -n {shlex.quote(namespace)} -o json"
         result = self.run_command(cmd)
         if result.exit_code != 0:
-            probe_error = result.stderr.strip() or result.stdout.strip() or f"exit code {result.exit_code}"
+            probe_error = command_detail(result)
             return {}, probe_error
 
         try:

isvtest/src/isvtest/validations/k8s_crd_webhook.py

@@ -33,7 +33,13 @@
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.x509.oid import NameOID
 
-from isvtest.core.k8s import KubectlParseError, get_kubectl_base_shell, get_kubectl_command, parse_kubectl_json
+from isvtest.core.k8s import (
+    KubectlParseError,
+    command_detail,
+    get_kubectl_base_shell,
+    get_kubectl_command,
+    parse_kubectl_json,
+)
 from isvtest.core.validation import BaseValidation
 
 _MANIFEST_PATH = Path(__file__).parent / "manifests" / "k8s" / "crd_webhook.yaml"
@@ -159,13 +165,13 @@ def _create_namespace(self) -> bool:
         namespace = shlex.quote(self._names.namespace)
         result = self.run_command(f"{self._kubectl_base} create namespace {namespace}")
         if result.exit_code != 0:
-            self.set_failed(f"Failed to create namespace {self._names.namespace}: {_command_detail(result)}")
+            self.set_failed(f"Failed to create namespace {self._names.namespace}: {command_detail(result)}")
             return False
 
         label = shlex.quote(f"{_LABEL_KEY}={self._names.suffix}")
         label_result = self.run_command(f"{self._kubectl_base} label namespace {namespace} {label} --overwrite")
         if label_result.exit_code != 0:
-            self.set_failed(f"Failed to label namespace {self._names.namespace}: {_command_detail(label_result)}")
+            self.set_failed(f"Failed to label namespace {self._names.namespace}: {command_detail(label_result)}")
             return False
         return True
 
@@ -178,7 +184,7 @@ def _wait_for_webhook_deployment(self) -> bool:
         )
         result = self.run_command(cmd, timeout=self._wait_timeout + 30)
         if result.exit_code != 0:
-            self.set_failed(f"Webhook deployment did not become Available: {_command_detail(result)}")
+            self.set_failed(f"Webhook deployment did not become Available: {command_detail(result)}")
             return False
         return True
 
@@ -190,7 +196,7 @@ def _wait_for_crd(self) -> bool:
         )
         result = self.run_command(cmd, timeout=self._wait_timeout + 30)
         if result.exit_code != 0:
-            self.set_failed(f"CustomResourceDefinition did not become Established: {_command_detail(result)}")
+            self.set_failed(f"CustomResourceDefinition did not become Established: {command_detail(result)}")
             return False
         return True
 
@@ -203,14 +209,14 @@ def _verify_custom_resource_mutation(self) -> bool:
         for attempt in range(attempts):
             proc = self._run_apply(manifest, timeout=self._apply_timeout(deadline))
             if proc.returncode != 0:
-                last_detail = _process_detail(proc)
+                last_detail = command_detail(proc)
             else:
                 result = self.run_command(
                     f"{self._kubectl_base} get {shlex.quote(self._names.resource)} mutated "
                     f"-n {shlex.quote(self._names.namespace)} -o json"
                 )
                 if result.exit_code != 0:
-                    last_detail = _command_detail(result)
+                    last_detail = command_detail(result)
                 else:
                     try:
                         payload = parse_kubectl_json(result, "mutated custom resource")
@@ -245,7 +251,7 @@ def _expect_apply_rejected(
         deadline = time.monotonic() + self._wait_timeout
         for attempt in range(attempts):
             proc = self._run_apply(manifest, timeout=self._apply_timeout(deadline))
-            detail = _process_detail(proc)
+            detail = command_detail(proc)
             if proc.returncode != 0:
                 if expected in detail:
                     return True
@@ -278,7 +284,7 @@ def _apply_or_fail(self, manifest: str, label: str) -> bool:
         """Apply a manifest and mark the validation failed on kubectl errors."""
         proc = self._run_apply(manifest)
         if proc.returncode != 0:
-            self.set_failed(f"kubectl apply failed for {label}: {_process_detail(proc)}")
+            self.set_failed(f"kubectl apply failed for {label}: {command_detail(proc)}")
             return False
         return True
 
@@ -484,7 +490,7 @@ def _cleanup_command(self, cmd: str, label: str) -> None:
             self.log.warning("%s cleanup raised: %s", label, exc)
             return
         if result.exit_code != 0:
-            self.log.warning("%s cleanup failed: %s", label, _command_detail(result))
+            self.log.warning("%s cleanup failed: %s", label, command_detail(result))
 
 
 def _load_template_docs() -> list[dict[str, Any]]:
@@ -587,13 +593,3 @@ def _generate_cert_bundle(dns_names: list[str]) -> _CertBundle:
 def _b64(data: bytes) -> str:
     """Return base64-encoded text for Kubernetes byte fields."""
     return base64.b64encode(data).decode("ascii")
-
-
-def _process_detail(proc: subprocess.CompletedProcess[str]) -> str:
-    """Return the most useful stderr/stdout detail from a completed process."""
-    return (proc.stderr or proc.stdout or f"exit code {proc.returncode}").strip()
-
-
-def _command_detail(result: Any) -> str:
-    """Return the most useful stderr/stdout detail from a command result."""
-    return (result.stderr or result.stdout or f"exit code {result.exit_code}").strip()