diff --git a/docs/test-plan.adoc b/docs/test-plan.adoc index fabd5257..4335941d 100644 --- a/docs/test-plan.adoc +++ b/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 diff --git a/docs/test-plan.yaml b/docs/test-plan.yaml index bf209c91..966ea207 100644 --- a/docs/test-plan.yaml +++ b/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 diff --git a/isvctl/configs/suites/k8s.yaml b/isvctl/configs/suites/k8s.yaml index 21000f17..f8919271 100644 --- a/isvctl/configs/suites/k8s.yaml +++ b/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"] diff --git a/isvtest/src/isvtest/core/k8s.py b/isvtest/src/isvtest/core/k8s.py index f1d18fab..ffecfdac 100644 --- a/isvtest/src/isvtest/core/k8s.py +++ b/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] = [] diff --git a/isvtest/src/isvtest/validations/k8s_api_network_acl.py b/isvtest/src/isvtest/validations/k8s_api_network_acl.py index 4a5a2275..45804cd1 100644 --- a/isvtest/src/isvtest/validations/k8s_api_network_acl.py +++ b/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}" diff --git a/isvtest/src/isvtest/validations/k8s_autoscaler.py b/isvtest/src/isvtest/validations/k8s_autoscaler.py index d3c497ec..ded13c88 100644 --- a/isvtest/src/isvtest/validations/k8s_autoscaler.py +++ b/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)}" diff --git a/isvtest/src/isvtest/validations/k8s_control_plane_logs.py b/isvtest/src/isvtest/validations/k8s_control_plane_logs.py index 33f340c0..e65bf01a 100644 --- a/isvtest/src/isvtest/validations/k8s_control_plane_logs.py +++ b/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: diff --git a/isvtest/src/isvtest/validations/k8s_crd_webhook.py b/isvtest/src/isvtest/validations/k8s_crd_webhook.py index b5420dde..4dc3001f 100644 --- a/isvtest/src/isvtest/validations/k8s_crd_webhook.py +++ b/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() diff --git a/isvtest/src/isvtest/validations/k8s_gpu_operator.py b/isvtest/src/isvtest/validations/k8s_gpu_operator.py index 8d91b7d0..1e02ed46 100644 --- a/isvtest/src/isvtest/validations/k8s_gpu_operator.py +++ b/isvtest/src/isvtest/validations/k8s_gpu_operator.py @@ -13,10 +13,24 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json import shlex +from typing import Any, NamedTuple + +import pytest from isvtest.config.settings import get_k8s_gpu_operator_namespace -from isvtest.core.k8s import get_kubectl_base_shell, kubectl_items_or_fail, pod_status_reason +from isvtest.core.k8s import ( + KubectlParseError, + command_detail, + get_kubectl_base_shell, + is_resource_absent, + kubectl_items_or_fail, + names_from_items, + parse_kubectl_json, + parse_kubectl_json_items, + pod_status_reason, +) from isvtest.core.validation import BaseValidation @@ -62,3 +76,222 @@ def run(self) -> None: return self.set_passed(f"Found {len(running_pods)} running pods in '{namespace}'") + + +class _DriverConfigKind(NamedTuple): + """A GPU Operator resource carrying the driver version a tenant sets.""" + + resource: str + version_path: tuple[str, ...] + + +# Both spellings of the operator's driver contract: ClusterPolicy is the +# long-standing single object, NVIDIADriver the per-node-pool CR that GPU +# Operator 24.6+ adds alongside it. +DRIVER_CONFIG_KINDS: tuple[_DriverConfigKind, ...] = ( + _DriverConfigKind("clusterpolicies.nvidia.com", ("spec", "driver", "version")), + _DriverConfigKind("nvidiadrivers.nvidia.com", ("spec", "version")), +) + +# Overriding the driver version in place is patch/update; swapping the whole +# configuration out is delete/create. +CONFIG_VERBS: tuple[str, ...] = ("patch", "update", "delete", "create") + +# Installing a different operator version rewrites its own workloads, and the +# driver runs as a DaemonSet the tenant has to be able to replace with it. +WORKLOAD_RESOURCES: tuple[str, ...] = ("deployments.apps", "daemonsets.apps") +WORKLOAD_VERBS: tuple[str, ...] = ("patch", "update") + + +class K8sGpuOperatorOverrideCheck(BaseValidation): + """Verify a tenant can override the provider-default GPU Operator and driver. + + A provider may ship the GPU Operator as a managed add-on; the requirement is + that a tenant can still install the operator and driver versions its + workloads need. Three things have to hold, none of which mutate the cluster: + + * The operator's driver configuration exists, so there is something to + override. A cluster with neither ``ClusterPolicy`` nor ``NVIDIADriver`` + exposes no operator-managed driver version and fails rather than passing + on the absence of evidence. + * The tenant is authorized to rewrite that configuration and the operator's + own workloads, asked server-side via ``kubectl auth can-i``. + * A write of the tenant-required driver version survives admission. The + patch runs with ``--dry-run=server``, so RBAC and every mutating and + validating webhook evaluate it but nothing is persisted. Reading the + version back off the returned object is what catches a provider webhook + that accepts the write and pins the version anyway. When the cluster + already runs the tenant-required version the patch would change nothing, + so a neighbouring version is requested to keep admission under test. + + Config: + driver_version: Tenant-required driver version to attempt. Skips when + empty - there is no override to prove. + namespace: Namespace holding the GPU Operator workloads. Falls back to + the GPU Operator namespace setting. + """ + + description = "Verify the provider-default GPU Operator driver version can be overridden by the tenant." + + def run(self) -> None: + """Probe driver configuration, tenant authorization, and admission.""" + driver_version = str(self.config.get("driver_version") or "").strip() + if not driver_version: + pytest.skip("driver_version is not configured; no tenant-required version to override to") + + namespace = self.config.get("namespace") or get_k8s_gpu_operator_namespace() + kubectl_base = get_kubectl_base_shell() + + discovered = self._discover_driver_config(kubectl_base) + if discovered is None: + return + kind, name, current_version = discovered + + if not self._tenant_can_replace_operator(kubectl_base, kind, namespace): + return + + self._verify_override_admitted(kubectl_base, kind, name, driver_version, current_version) + + def _discover_driver_config(self, kubectl_base: str) -> tuple[_DriverConfigKind, str, str] | None: + """Return the kind, object name, and configured version of the driver config. + + Returns ``None`` after marking the check failed when no kind is present + or a query failed for a reason other than the kind being absent. + """ + query_errors: list[str] = [] + + for kind in DRIVER_CONFIG_KINDS: + result = self.run_command(f"{kubectl_base} get {shlex.quote(kind.resource)} -o json") + if result.exit_code != 0: + if not is_resource_absent(result.stderr): + query_errors.append(f"{kind.resource}: {command_detail(result)}") + continue + try: + items = parse_kubectl_json_items(result, kind.resource) + except KubectlParseError as exc: + self.set_failed(str(exc)) + return None + if items: + return kind, names_from_items(items)[0], _dig(items[0], kind.version_path) + + if query_errors: + self.set_failed("Unable to query the GPU Operator driver configuration: " + "; ".join(query_errors)) + return None + + resources = ", ".join(kind.resource for kind in DRIVER_CONFIG_KINDS) + self.set_failed( + f"No GPU Operator driver configuration found ({resources}); the driver version is not " + "operator-managed, so a tenant override cannot be proven" + ) + return None + + def _tenant_can_replace_operator( + self, + kubectl_base: str, + kind: _DriverConfigKind, + namespace: str, + ) -> bool: + """Return True when every override route is authorized, marking failures itself.""" + probes = [(verb, kind.resource, "") for verb in CONFIG_VERBS] + probes += [(verb, resource, namespace) for resource in WORKLOAD_RESOURCES for verb in WORKLOAD_VERBS] + + denials: list[str] = [] + for verb, resource, scope in probes: + command = f"{kubectl_base} auth can-i {shlex.quote(verb)} {shlex.quote(resource)}" + if scope: + command += f" -n {shlex.quote(scope)}" + result = self.run_command(command) + + # `auth can-i` answers "no" with a non-zero exit, so the exit code + # alone cannot separate a denial from a broken probe. + answer = result.stdout.strip().lower() + if answer.startswith("yes"): + continue + if answer.startswith("no"): + denials.append(f"cannot {verb} {resource}" + (f" in {scope}" if scope else "")) + continue + + self.set_failed(f"Authorization probe for '{verb} {resource}' was inconclusive: {command_detail(result)}") + return False + + if denials: + self.set_failed( + f"Tenant is not authorized to replace the provider-default GPU Operator: {'; '.join(denials)}" + ) + return False + return True + + def _verify_override_admitted( + self, + kubectl_base: str, + kind: _DriverConfigKind, + name: str, + driver_version: str, + current_version: str, + ) -> None: + """Dry-run the version override server-side and confirm admission keeps it.""" + # Requesting the version already configured writes nothing, so admission + # would return it untouched and the check would pass without a version + # ever having been overridden. Probe with a neighbouring version instead. + requested = driver_version if driver_version != current_version else _next_version(current_version) + + patch = json.dumps(_nested(kind.version_path, requested)) + result = self.run_command( + f"{kubectl_base} patch {shlex.quote(kind.resource)} {shlex.quote(name)} " + f"--type=merge --patch {shlex.quote(patch)} --dry-run=server -o json" + ) + if result.exit_code != 0: + self.set_failed( + f"Admission rejected driver version '{requested}' on {kind.resource}/{name}: {command_detail(result)}" + ) + return + + try: + admitted = parse_kubectl_json(result, f"{kind.resource}/{name} dry-run response") + except KubectlParseError as exc: + self.set_failed(str(exc)) + return + + version_path = ".".join(kind.version_path) + admitted_version = _dig(admitted, kind.version_path) + if admitted_version != requested: + self.set_failed( + f"Admission kept the provider-default driver version: requested '{requested}' at " + f"{version_path} on {kind.resource}/{name}, admitted object reports " + f"'{admitted_version or 'unset'}'" + ) + return + + # A probed neighbouring version is one nobody asked for, so name the + # tenant-required version rather than leave it out of the report. + probed = "" if requested == driver_version else f"; tenant-required '{driver_version}' is already installed" + self.set_passed( + f"Tenant can override the provider-default driver: {kind.resource}/{name} {version_path} " + f"accepts a write of '{requested}' (currently '{current_version or 'unset'}'){probed}" + ) + + +def _dig(obj: dict[str, Any], path: tuple[str, ...]) -> str: + """Return the string at ``path`` in a nested mapping, or ``""`` when absent.""" + current: Any = obj + for key in path: + if not isinstance(current, dict): + return "" + current = current.get(key) + return current if isinstance(current, str) else "" + + +def _next_version(version: str) -> str: + """Return ``version`` with its trailing number incremented.""" + head, separator, tail = version.rpartition(".") + if tail.isdigit(): + return f"{head}{separator}{int(tail) + 1}" + return f"{version}.1" + + +def _nested(path: tuple[str, ...], value: str) -> dict[str, Any]: + """Build the nested mapping that places ``value`` at ``path``.""" + body: Any = value + for key in reversed(path): + body = {key: body} + return body diff --git a/isvtest/src/isvtest/validations/k8s_storage.py b/isvtest/src/isvtest/validations/k8s_storage.py index 94feb1e3..244a975a 100644 --- a/isvtest/src/isvtest/validations/k8s_storage.py +++ b/isvtest/src/isvtest/validations/k8s_storage.py @@ -56,6 +56,7 @@ ) from isvtest.core.k8s import ( KubectlParseError, + command_detail, get_kubectl_base_shell, get_kubectl_command, parse_kubectl_json, @@ -118,7 +119,7 @@ def _get_pvc_json(run_command, kubectl_base: str, namespace: str, pvc_name: str) cmd = f"{kubectl_base} get pvc {shlex.quote(pvc_name)} -n {shlex.quote(namespace)} -o json" result = run_command(cmd) if result.exit_code != 0: - return None, result.stderr.strip() or result.stdout.strip() or f"exit code {result.exit_code}" + return None, command_detail(result) try: return parse_kubectl_json(result, f"PVC {pvc_name!r}"), "" except KubectlParseError as exc: @@ -217,11 +218,7 @@ def _collect_storage_diagnostics( break try: result = run_command(command, timeout=min(_DIAGNOSTIC_COMMAND_TIMEOUT, max(1, int(remaining)))) - output = ( - result.stdout - if result.exit_code == 0 - else (result.stderr or result.stdout or f"command exited {result.exit_code}") - ) + output = result.stdout if result.exit_code == 0 else command_detail(result) except Exception as exc: output = f"diagnostic command failed: {type(exc).__name__}: {exc}" sections.append(_bounded_diagnostic_section(section_label, output)) diff --git a/isvtest/tests/test_k8s_gpu_operator.py b/isvtest/tests/test_k8s_gpu_operator.py index e171011c..358e6004 100644 --- a/isvtest/tests/test_k8s_gpu_operator.py +++ b/isvtest/tests/test_k8s_gpu_operator.py @@ -13,15 +13,20 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Tests for the GPU Operator pod-status validation.""" +"""Tests for the GPU Operator pod-status and tenant-override validations.""" from __future__ import annotations import json from unittest.mock import patch +import pytest + from isvtest.core.runners import CommandResult -from isvtest.validations.k8s_gpu_operator import K8sGpuOperatorPodsCheck +from isvtest.validations.k8s_gpu_operator import ( + K8sGpuOperatorOverrideCheck, + K8sGpuOperatorPodsCheck, +) def _ok(stdout: str = "", stderr: str = "") -> CommandResult: @@ -29,6 +34,11 @@ def _ok(stdout: str = "", stderr: str = "") -> CommandResult: return CommandResult(exit_code=0, stdout=stdout, stderr=stderr, duration=0.0) +def _fail(stdout: str = "", stderr: str = "", exit_code: int = 1) -> CommandResult: + """Return a failed ``CommandResult``.""" + return CommandResult(exit_code=exit_code, stdout=stdout, stderr=stderr, duration=0.0) + + def test_gpu_operator_pods_use_json_phase() -> None: """Verify GPU Operator pod status is parsed from JSON.""" check = K8sGpuOperatorPodsCheck(config={"namespace": "gpu-operator"}) @@ -69,3 +79,227 @@ def test_gpu_operator_pods_reject_crashlooping_running_phase() -> None: assert not check.passed assert "No GPU Operator pods are running" in check.message + + +CLUSTER_POLICY = { + "metadata": {"name": "cluster-policy"}, + "spec": {"driver": {"version": "550.54.15"}}, +} +NVIDIA_DRIVER = {"metadata": {"name": "gpu-driver"}, "spec": {"version": "550.54.15"}} + +NO_SUCH_RESOURCE = 'error: the server doesn\'t have a resource type "nvidiadrivers"' + +GET_CLUSTER_POLICY = "get clusterpolicies.nvidia.com" +GET_NVIDIA_DRIVER = "get nvidiadrivers.nvidia.com" +DRY_RUN = "--dry-run=server" + + +def _found(*objects: dict[str, object]) -> CommandResult: + """Return a ``kubectl get -o json`` list response.""" + return _ok(json.dumps({"items": list(objects)})) + + +def _override_check(**config: str) -> K8sGpuOperatorOverrideCheck: + """Build the override check with the suite's default wiring.""" + return K8sGpuOperatorOverrideCheck(config={"namespace": "gpu-operator", "driver_version": "580.82.07", **config}) + + +def _run(check: K8sGpuOperatorOverrideCheck, responses: dict[str, CommandResult]) -> list[str]: + """Run the check, answering each kubectl command from the first matching fragment. + + Commands no fragment matches answer ``yes``, so a test states only the + responses it is about - every other authorization probe is allowed. + """ + + def respond(command: str, *_args: object, **_kwargs: object) -> CommandResult: + for fragment, result in responses.items(): + if fragment in command: + return result + return _ok("yes") + + with ( + patch("isvtest.validations.k8s_gpu_operator.get_kubectl_base_shell", return_value="kubectl"), + patch.object(check, "run_command", side_effect=respond) as mock_run, + ): + check.run() + + return [call[0][0] for call in mock_run.call_args_list] + + +def test_override_passes_when_admission_keeps_the_tenant_driver_version() -> None: + """A writable ClusterPolicy whose dry-run keeps the requested version proves the override.""" + check = _override_check() + admitted = {"metadata": {"name": "cluster-policy"}, "spec": {"driver": {"version": "580.82.07"}}} + + commands = _run( + check, + {GET_CLUSTER_POLICY: _found(CLUSTER_POLICY), DRY_RUN: _ok(json.dumps(admitted))}, + ) + + assert check.passed, check.message + assert "accepts a write of '580.82.07'" in check.message + assert "currently '550.54.15'" in check.message + assert commands[0] == "kubectl get clusterpolicies.nvidia.com -o json" + assert "kubectl auth can-i patch clusterpolicies.nvidia.com" in commands + assert "kubectl auth can-i patch deployments.apps -n gpu-operator" in commands + assert "kubectl auth can-i patch daemonsets.apps -n gpu-operator" in commands + # The override must never be persisted: admission runs it, the cluster keeps + # the provider default. + assert commands[-1] == ( + "kubectl patch clusterpolicies.nvidia.com cluster-policy --type=merge " + '--patch \'{"spec": {"driver": {"version": "580.82.07"}}}\' --dry-run=server -o json' + ) + + +def test_override_falls_back_to_nvidiadriver_when_clusterpolicy_is_absent() -> None: + """Newer installs express the driver version on NVIDIADriver instead.""" + check = _override_check() + admitted = {"metadata": {"name": "gpu-driver"}, "spec": {"version": "580.82.07"}} + + commands = _run( + check, + { + GET_CLUSTER_POLICY: _found(), + GET_NVIDIA_DRIVER: _found(NVIDIA_DRIVER), + DRY_RUN: _ok(json.dumps(admitted)), + }, + ) + + assert check.passed, check.message + assert commands[1] == "kubectl get nvidiadrivers.nvidia.com -o json" + assert commands[-1] == ( + "kubectl patch nvidiadrivers.nvidia.com gpu-driver --type=merge " + '--patch \'{"spec": {"version": "580.82.07"}}\' --dry-run=server -o json' + ) + + +def test_override_fails_when_no_driver_configuration_exists() -> None: + """With no operator-managed driver version there is no override to prove.""" + check = _override_check() + + _run(check, {GET_CLUSTER_POLICY: _found(), GET_NVIDIA_DRIVER: _fail(stderr=NO_SUCH_RESOURCE)}) + + assert not check.passed + assert "No GPU Operator driver configuration found" in check.message + + +def test_override_fails_when_the_driver_configuration_query_errors() -> None: + """An unreachable API is reported as a query failure, not as a missing driver config.""" + check = _override_check() + + _run( + check, + { + GET_CLUSTER_POLICY: _fail(stderr="The connection to the server 10.0.0.1:6443 was refused"), + GET_NVIDIA_DRIVER: _fail(stderr=NO_SUCH_RESOURCE), + }, + ) + + assert not check.passed + assert "Unable to query the GPU Operator driver configuration" in check.message + assert "was refused" in check.message + + +def test_override_fails_when_a_required_verb_is_denied() -> None: + """A provider that locks the driver configuration down via RBAC fails.""" + check = _override_check() + + _run( + check, + {GET_CLUSTER_POLICY: _found(CLUSTER_POLICY), "can-i delete clusterpolicies.nvidia.com": _fail("no")}, + ) + + assert not check.passed + assert "cannot delete clusterpolicies.nvidia.com" in check.message + + +def test_override_fails_when_the_operator_workloads_are_read_only() -> None: + """Replacing the operator means rewriting its workloads in its own namespace.""" + check = _override_check() + + _run(check, {GET_CLUSTER_POLICY: _found(CLUSTER_POLICY), "can-i patch daemonsets.apps": _fail("no")}) + + assert not check.passed + assert "cannot patch daemonsets.apps in gpu-operator" in check.message + + +def test_override_fails_when_an_authorization_probe_is_inconclusive() -> None: + """A probe that answers neither yes nor no is an error, not a silent pass.""" + check = _override_check() + + _run( + check, + { + GET_CLUSTER_POLICY: _found(CLUSTER_POLICY), + "can-i patch clusterpolicies.nvidia.com": _fail(stderr="error: unknown flag: --subresource"), + }, + ) + + assert not check.passed + assert "was inconclusive" in check.message + + +def test_override_fails_when_admission_rejects_the_version() -> None: + """A validating webhook that refuses tenant driver versions fails the check.""" + check = _override_check() + + _run( + check, + { + GET_CLUSTER_POLICY: _found(CLUSTER_POLICY), + DRY_RUN: _fail(stderr='admission webhook "gpu-policy.provider.example" denied the request'), + }, + ) + + assert not check.passed + assert "Admission rejected driver version '580.82.07'" in check.message + assert "denied the request" in check.message + + +def test_override_fails_when_admission_pins_the_provider_default_version() -> None: + """A mutating webhook may accept the write and quietly restore its own version.""" + check = _override_check() + + _run(check, {GET_CLUSTER_POLICY: _found(CLUSTER_POLICY), DRY_RUN: _ok(json.dumps(CLUSTER_POLICY))}) + + assert not check.passed + assert "Admission kept the provider-default driver version" in check.message + assert "admitted object reports '550.54.15'" in check.message + + +def test_override_probes_a_neighbouring_version_when_the_required_one_is_installed() -> None: + """Requesting the version already set would prove nothing, so the probe moves off it.""" + check = _override_check(driver_version="550.54.15") + admitted = {"metadata": {"name": "cluster-policy"}, "spec": {"driver": {"version": "550.54.16"}}} + + commands = _run( + check, + {GET_CLUSTER_POLICY: _found(CLUSTER_POLICY), DRY_RUN: _ok(json.dumps(admitted))}, + ) + + assert check.passed, check.message + assert "accepts a write of '550.54.16'" in check.message + assert "tenant-required '550.54.15' is already installed" in check.message + assert commands[-1] == ( + "kubectl patch clusterpolicies.nvidia.com cluster-policy --type=merge " + '--patch \'{"spec": {"driver": {"version": "550.54.16"}}}\' --dry-run=server -o json' + ) + + +def test_override_fails_when_the_installed_version_is_pinned_against_any_change() -> None: + """A no-op write must not pass: admission returning the current version is a failure.""" + check = _override_check(driver_version="550.54.15") + + _run(check, {GET_CLUSTER_POLICY: _found(CLUSTER_POLICY), DRY_RUN: _ok(json.dumps(CLUSTER_POLICY))}) + + assert not check.passed + assert "Admission kept the provider-default driver version" in check.message + assert "requested '550.54.16'" in check.message + + +def test_override_skips_without_a_tenant_required_version() -> None: + """With no target version configured there is nothing to override to.""" + check = _override_check(driver_version="") + + with pytest.raises(pytest.skip.Exception, match="driver_version is not configured"): + check.run()