From 8a459b43cfd4cbefa3834f7b51c22ba78de62a2b Mon Sep 17 00:00:00 2001 From: Alexandra Bueno Date: Thu, 25 Jun 2026 22:24:53 -0600 Subject: [PATCH 1/2] report csi driver health; add base config and tests Signed-off-by: Alexandra Bueno --- isvctl/configs/suites/k8s.yaml | 23 ++ .../src/isvtest/validations/k8s_storage.py | 263 +++++++++++++++ isvtest/tests/test_k8s_storage.py | 309 ++++++++++++++++++ 3 files changed, 595 insertions(+) diff --git a/isvctl/configs/suites/k8s.yaml b/isvctl/configs/suites/k8s.yaml index 86a5f060..a62f6771 100644 --- a/isvctl/configs/suites/k8s.yaml +++ b/isvctl/configs/suites/k8s.yaml @@ -296,6 +296,29 @@ tests: access_mode: "ReadWriteOnce" bind_timeout_s: 180 namespace_prefix: "isvtest-csi-prov" + K8sCsiDriverHealthCheck: + test_id: "N/A" + labels: ["kubernetes", "min_req"] + # Verifies each configured CSI driver is installed and healthy. + # Controller/node health (Deployment + DaemonSet) runs only + # when a spec supplies a `workloads` block; otherwise those subtests are + # skipped (driver registration is still checked). + # - storage_classes: ["..."] + # workloads: + # namespace: kube-system + # controller: { deployment: } + # node: { daemonset: } + drivers: + - storage_classes: + - "{{ steps.setup.csi.block_storage_class | default('', true) }}" + workloads: {} + - storage_classes: + - "{{ steps.setup.csi.shared_fs_storage_class | default('', true) }}" + workloads: {} + - storage_classes: + - "{{ steps.setup.csi.nfs_storage_class | default('', true) }}" + workloads: {} + min_controller_replicas: 1 k8s_identity: checks: diff --git a/isvtest/src/isvtest/validations/k8s_storage.py b/isvtest/src/isvtest/validations/k8s_storage.py index befc73b7..9582509c 100644 --- a/isvtest/src/isvtest/validations/k8s_storage.py +++ b/isvtest/src/isvtest/validations/k8s_storage.py @@ -30,6 +30,9 @@ * :class:`K8sCsiProvisioningModesCheck` - verify CSI supports both dynamic and static provisioning, driving each through a real mount + canary-file write/read inside a BusyBox pod. +* :class:`K8sCsiDriverHealthCheck` - verify each configured StorageClass + resolves to a registered ``CSIDriver`` and that the driver's controller + Deployment and node DaemonSet are healthy. """ from __future__ import annotations @@ -1242,6 +1245,11 @@ def _parse_label_pairs(raw: list[Any]) -> list[tuple[str, str]]: return pairs +def _as_dict(value: Any) -> dict[str, Any]: + """Return ``value`` when it is a dict, else an empty dict (defensive config access).""" + return value if isinstance(value, dict) else {} + + def _load_items(stdout: str) -> list[dict[str, Any]]: """Parse a ``kubectl get ... -o json`` list payload into its ``items`` list.""" if not stdout: @@ -1942,3 +1950,258 @@ def _set_mount_pod_fields( volumes[0].pop("emptyDir", None) volumes[0]["persistentVolumeClaim"] = {"claimName": pvc_name} return doc + + +class K8sCsiDriverHealthCheck(BaseValidation): + """Verify CSI drivers are installed and healthy for the configured StorageClasses. + + Resolves each StorageClass to its ``spec.provisioner`` and, per unique + provisioner, checks the StorageClass exists, the ``CSIDriver`` is registered, + and (when ``workloads`` are given) the controller Deployment and node + DaemonSet are Ready. + + Config (``drivers`` is a list of specs; the whole check skips when no + StorageClasses resolve):: + + drivers: + - storage_classes: [, ...] + workloads: # optional; omit to skip controller/node checks + namespace: kube-system + controller: {deployment: } + node: {daemonset: } + + ``min_controller_replicas`` (default 1) sets the minimum Ready controller + replicas. + """ + + description: ClassVar[str] = "Verify the CSI driver is installed and healthy." + timeout: ClassVar[int] = 60 + + def run(self) -> None: + """Resolve each driver spec's StorageClasses to provisioners and run health subtests.""" + sc_to_workloads = self._collect_driver_specs() + if not sc_to_workloads: + self.set_passed("Skipped: no storage_classes configured") + return + + min_replicas = self._parse_positive_int("min_controller_replicas", default=1) + if min_replicas is None: + return + kubectl_base = get_kubectl_base_shell() + + # Single kubectl call to resolve all SC names → provisioners. + sc_result = self.run_command(f"{kubectl_base} get storageclass -o json") + if sc_result.exit_code != 0: + self.set_failed(f"kubectl get storageclass failed: {sc_result.stderr.strip()}") + return + all_sc_items = _load_items(sc_result.stdout) + sc_to_provisioner: dict[str, str] = { + (item.get("metadata") or {}).get("name", ""): str(item.get("provisioner") or "") + for item in all_sc_items + } + + any_failed = False + provisioner_to_scs: dict[str, list[str]] = {} + # Multiple StorageClasses can share a provisioner; track each distinct + # workloads block to detect conflicting configs. + provisioner_to_workloads: dict[str, list[dict[str, Any]]] = {} + for sc_name, workloads in sc_to_workloads.items(): + provisioner = sc_to_provisioner.get(sc_name) + if not provisioner: + self.report_subtest( + f"storageclass-found[{sc_name}]", + passed=False, + message=( + f"StorageClass {sc_name!r} not found in the cluster " + f"({len(all_sc_items)} StorageClass(es) present)" + ), + ) + any_failed = True + continue + provisioner_to_scs.setdefault(provisioner, []).append(sc_name) + if workloads: + distinct = provisioner_to_workloads.setdefault(provisioner, []) + if workloads not in distinct: + distinct.append(workloads) + + for provisioner in sorted(provisioner_to_scs): + if self._report_provisioner_health( + kubectl_base, + provisioner, + provisioner_to_scs[provisioner], + provisioner_to_workloads.get(provisioner) or [], + min_replicas, + ): + any_failed = True + + if any_failed: + self.set_failed("One or more CSI driver health subtests failed; see subtest details") + else: + self.set_passed(f"CSI driver health verified for: {', '.join(sorted(provisioner_to_scs))}") + + def _collect_driver_specs(self) -> dict[str, dict[str, Any]]: + """Map each configured StorageClass name to its (optional) ``workloads``""" + specs = self.config.get("drivers") + sc_to_workloads: dict[str, dict[str, Any]] = {} + for spec in specs if isinstance(specs, list) else []: + if not isinstance(spec, dict): + continue + workloads = _as_dict(spec.get("workloads")) + for raw_sc in spec.get("storage_classes") or []: + sc = str(raw_sc).strip() + if sc and sc not in sc_to_workloads: + sc_to_workloads[sc] = workloads + return sc_to_workloads + + def _report_provisioner_health( + self, + kubectl_base: str, + provisioner: str, + storage_classes: list[str], + workloads_list: list[dict[str, Any]], + min_replicas: int, + ) -> bool: + """Report the registration + controller/node subtests for one provisioner. + + Returns ``True`` if any non-skipped subtest failed. + """ + tag = f"[{provisioner}]" + + ok, msg = self._check_csidriver_registered(kubectl_base, provisioner) + self.report_subtest(f"csidriver-registered{tag}", passed=ok, message=msg) + failed = not ok + + if len(workloads_list) > 1: + conflict_msg = ( + f"Conflicting `workloads` blocks for provisioner {provisioner!r} " + f"across StorageClasses {storage_classes} ({len(workloads_list)} distinct); " + "use a single workloads block per provisioner" + ) + self.report_subtest(f"controller-deployment-healthy{tag}", passed=False, message=conflict_msg) + self.report_subtest(f"node-daemonset-healthy{tag}", passed=False, message=conflict_msg) + return True + + if not workloads_list: + skip_msg = ( + f"No workloads configured for provisioner {provisioner!r}; add a " + "`workloads` block (controller.deployment and node.daemonset) to its " + "drivers entry to health-check the controller Deployment and node DaemonSet" + ) + self.report_subtest(f"controller-deployment-healthy{tag}", passed=True, message=skip_msg, skipped=True) + self.report_subtest(f"node-daemonset-healthy{tag}", passed=True, message=skip_msg, skipped=True) + return failed + + workloads = workloads_list[0] + namespace = str(workloads.get("namespace") or "kube-system") + controller = _as_dict(workloads.get("controller")) + node = _as_dict(workloads.get("node")) + + ok, msg = self._check_controller_deployment( + kubectl_base, namespace, str(controller.get("deployment") or ""), min_replicas + ) + self.report_subtest(f"controller-deployment-healthy{tag}", passed=ok, message=msg) + failed = failed or not ok + + ok, msg = self._check_node_daemonset(kubectl_base, namespace, str(node.get("daemonset") or "")) + self.report_subtest(f"node-daemonset-healthy{tag}", passed=ok, message=msg) + return failed or not ok + + def _check_csidriver_registered(self, kubectl_base: str, provisioner: str) -> tuple[bool, str]: + """Return ``(ok, message)`` for the ``csidriver-registered`` subtest.""" + cmd = f"{kubectl_base} get csidriver {shlex.quote(provisioner)} --ignore-not-found=true -o json" + result = self.run_command(cmd) + if result.exit_code != 0: + return False, f"kubectl get csidriver {provisioner!r} failed: {result.stderr.strip()}" + if not result.stdout.strip(): + return False, f"CSIDriver/{provisioner} is not registered in the cluster" + try: + payload = parse_kubectl_json(result, f"CSIDriver {provisioner!r}") + except KubectlParseError as exc: + return False, str(exc) + name = (payload.get("metadata") or {}).get("name", "") + return True, f"CSIDriver/{name} is registered" + + def _check_controller_deployment( + self, + kubectl_base: str, + namespace: str, + deployment_name: str, + min_replicas: int, + ) -> tuple[bool, str]: + """Return ``(ok, message)`` for the ``controller-deployment-healthy`` subtest.""" + deployment, error = self._get_workload( + kubectl_base, "deployment", namespace, deployment_name, "workloads.controller.deployment" + ) + if error: + return False, error + + meta = deployment.get("metadata") or {} + spec = deployment.get("spec") or {} + status = deployment.get("status") or {} + name = meta.get("name", "") + desired = int(spec.get("replicas") or 0) + ready = int(status.get("readyReplicas") or 0) + if desired < min_replicas: + return False, ( + f"Deployment {namespace}/{name} has spec.replicas={desired} < min_controller_replicas={min_replicas}" + ) + if ready != desired: + return False, (f"Deployment {namespace}/{name}: readyReplicas={ready} != spec.replicas={desired}") + return True, f"Deployment {namespace}/{name}: {ready}/{desired} replicas Ready" + + def _check_node_daemonset( + self, + kubectl_base: str, + namespace: str, + daemonset_name: str, + ) -> tuple[bool, str]: + """Return ``(ok, message)`` for the ``node-daemonset-healthy`` subtest.""" + daemonset, error = self._get_workload( + kubectl_base, "daemonset", namespace, daemonset_name, "workloads.node.daemonset" + ) + if error: + return False, error + + meta = daemonset.get("metadata") or {} + status = daemonset.get("status") or {} + name = meta.get("name", "") + desired = int(status.get("desiredNumberScheduled") or 0) + available = int(status.get("numberAvailable") or 0) + ready = int(status.get("numberReady") or 0) + if desired == 0: + # An empty DaemonSet usually means its nodeSelector / tolerations + # don't match any node. + return False, ( + f"DaemonSet {namespace}/{name}: desiredNumberScheduled=0 (no nodes match the DaemonSet's node selector)" + ) + if available != desired or ready != desired: + return False, (f"DaemonSet {namespace}/{name}: desired={desired}, available={available}, ready={ready}") + return True, f"DaemonSet {namespace}/{name}: {ready}/{desired} pods Ready and Available" + + def _get_workload( + self, + kubectl_base: str, + kind: str, + namespace: str, + name: str, + name_config_key: str, + ) -> tuple[dict[str, Any], str]: + """Fetch a Deployment or DaemonSet by name. + + Returns ``(workload, "")`` on success or ``({}, error)`` on failure. + """ + if not name: + return {}, f"{name_config_key} must be set for the {kind} subtest" + cmd = ( + f"{kubectl_base} get {kind} {shlex.quote(name)} " + f"-n {shlex.quote(namespace)} --ignore-not-found=true -o json" + ) + result = self.run_command(cmd) + if result.exit_code != 0: + return {}, f"kubectl get {kind} failed: {result.stderr.strip()}" + if not result.stdout.strip(): + return {}, f"{kind.capitalize()} {namespace}/{name} not found" + try: + return parse_kubectl_json(result, f"{kind.capitalize()} {name!r}"), "" + except KubectlParseError as exc: + return {}, str(exc) diff --git a/isvtest/tests/test_k8s_storage.py b/isvtest/tests/test_k8s_storage.py index 035422ac..8ab05a42 100644 --- a/isvtest/tests/test_k8s_storage.py +++ b/isvtest/tests/test_k8s_storage.py @@ -19,6 +19,7 @@ import json import re +import shlex import subprocess from contextlib import contextmanager from typing import Any @@ -29,6 +30,7 @@ from isvtest.core.runners import CommandResult from isvtest.validations.k8s_storage import ( + K8sCsiDriverHealthCheck, K8sCsiProvisioningModesCheck, K8sCsiStorageQuotaApiCheck, K8sCsiStorageTypesCheck, @@ -2176,3 +2178,310 @@ def fake_run(cmd: str, timeout: int | None = None) -> CommandResult: assert not check.passed assert "Failed to create namespace" in check._error assert check._subtest_results == [] + + +def _deployment_json(name: str, *, replicas: int = 1, ready: int = 1) -> dict[str, Any]: + """Build a minimal Deployment payload for the health check.""" + return { + "metadata": {"name": name}, + "spec": {"replicas": replicas}, + "status": {"readyReplicas": ready}, + } + + +def _daemonset_json(name: str, *, desired: int = 3, available: int = 3, ready: int = 3) -> dict[str, Any]: + """Build a minimal DaemonSet payload for the health check.""" + return { + "metadata": {"name": name}, + "status": { + "desiredNumberScheduled": desired, + "numberAvailable": available, + "numberReady": ready, + }, + } + + +def _name_after(cmd: str, kind: str) -> str: + """Return the resource name token following ``get `` in a kubectl command.""" + tokens = shlex.split(cmd) + idx = tokens.index(kind) + return tokens[idx + 1] + + +class TestK8sCsiDriverHealthCheck: + """Tests for ``K8sCsiDriverHealthCheck``.""" + + def _make(self, config: dict[str, Any] | None = None) -> K8sCsiDriverHealthCheck: + return K8sCsiDriverHealthCheck(config=config or {}) + + def _run( + self, + check: K8sCsiDriverHealthCheck, + *, + sc_items: list[dict[str, Any]], + csidrivers: dict[str, dict[str, Any]] | None = None, + deployments: dict[str, dict[str, Any]] | None = None, + daemonsets: dict[str, dict[str, Any]] | None = None, + sc_result: CommandResult | None = None, + ) -> dict[str, dict[str, Any]]: + """Drive ``check.run()`` with stubbed kubectl responses; return subtests by name. + + Lookups that miss the provided dicts return empty stdout, mirroring + ``kubectl ... --ignore-not-found=true`` for a missing object. + """ + csidrivers = csidrivers or {} + deployments = deployments or {} + daemonsets = daemonsets or {} + + def fake_run(cmd: str, timeout: int | None = None) -> CommandResult: + if "get storageclass" in cmd: + return sc_result if sc_result is not None else _ok(stdout=json.dumps({"items": sc_items})) + if "get csidriver" in cmd: + obj = csidrivers.get(_name_after(cmd, "csidriver")) + return _ok(stdout=json.dumps(obj) if obj else "") + if "get deployment" in cmd: + obj = deployments.get(_name_after(cmd, "deployment")) + return _ok(stdout=json.dumps(obj) if obj else "") + if "get daemonset" in cmd: + obj = daemonsets.get(_name_after(cmd, "daemonset")) + return _ok(stdout=json.dumps(obj) if obj else "") + raise AssertionError(f"unexpected command: {cmd}") + + with patch.object(check, "run_command", side_effect=fake_run): + check.run() + return {r["name"]: r for r in check._subtest_results} + + @staticmethod + def _workloads( + deployment: str = "ebs-csi-controller", + daemonset: str = "ebs-csi-node", + namespace: str = "kube-system", + ) -> dict[str, Any]: + return { + "namespace": namespace, + "controller": {"deployment": deployment}, + "node": {"daemonset": daemonset}, + } + + def test_skips_workload_subtests_when_no_workloads_configured(self) -> None: + # A driver spec that lists StorageClasses but no `workloads`: the driver + # registration is still verified, but the controller/node subtests are + # reported as skipped (not failed) and no workload queries are issued. + check = self._make({"drivers": [{"storage_classes": ["ebs"]}]}) + + outcomes = self._run( + check, + sc_items=[{"metadata": {"name": "ebs"}, "provisioner": "ebs.csi.aws.com"}], + csidrivers={"ebs.csi.aws.com": {"metadata": {"name": "ebs.csi.aws.com"}}}, + ) + + assert check.passed, check._error + assert outcomes["csidriver-registered[ebs.csi.aws.com]"]["passed"] + assert outcomes["controller-deployment-healthy[ebs.csi.aws.com]"]["skipped"] + assert outcomes["node-daemonset-healthy[ebs.csi.aws.com]"]["skipped"] + + def test_skips_whole_check_when_no_storage_classes_resolve(self) -> None: + # Blank/whitespace StorageClasses (e.g. unrendered Jinja defaults) mean + # the whole check is a no-op pass and no kubectl call is issued. + check = self._make({"drivers": [{"storage_classes": ["", " "]}]}) + + with patch.object(check, "run_command", side_effect=AssertionError("should not query")): + check.run() + + assert check.passed + assert "no storage_classes" in check._output + assert check._subtest_results == [] + + def test_healthy_controller_and_daemonset_pass(self) -> None: + check = self._make( + {"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]} + ) + + outcomes = self._run( + check, + sc_items=[{"metadata": {"name": "ebs"}, "provisioner": "ebs.csi.aws.com"}], + csidrivers={"ebs.csi.aws.com": {"metadata": {"name": "ebs.csi.aws.com"}}}, + deployments={"ebs-csi-controller": _deployment_json("ebs-csi-controller", replicas=2, ready=2)}, + daemonsets={"ebs-csi-node": _daemonset_json("ebs-csi-node")}, + ) + + assert check.passed, check._error + assert outcomes["csidriver-registered[ebs.csi.aws.com]"]["passed"] + assert outcomes["controller-deployment-healthy[ebs.csi.aws.com]"]["passed"] + assert outcomes["node-daemonset-healthy[ebs.csi.aws.com]"]["passed"] + + def test_storageclass_not_found_fails(self) -> None: + check = self._make({"drivers": [{"storage_classes": ["missing"]}]}) + + outcomes = self._run(check, sc_items=[{"metadata": {"name": "other"}, "provisioner": "x"}]) + + assert not check.passed + assert outcomes["storageclass-found[missing]"]["passed"] is False + assert "not found" in outcomes["storageclass-found[missing]"]["message"] + + def test_csidriver_not_registered_fails(self) -> None: + check = self._make({"drivers": [{"storage_classes": ["ebs"]}]}) + + outcomes = self._run( + check, + sc_items=[{"metadata": {"name": "ebs"}, "provisioner": "ebs.csi.aws.com"}], + csidrivers={}, # no CSIDriver registered + ) + + assert not check.passed + reg = outcomes["csidriver-registered[ebs.csi.aws.com]"] + assert reg["passed"] is False + assert "not registered" in reg["message"] + + def test_storageclass_list_command_failure_sets_failed(self) -> None: + check = self._make({"drivers": [{"storage_classes": ["ebs"]}]}) + + self._run(check, sc_items=[], sc_result=_fail(stderr="boom")) + + assert not check.passed + assert "kubectl get storageclass failed" in check._error + assert check._subtest_results == [] + + def test_controller_deployment_not_ready_fails(self) -> None: + check = self._make( + {"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]} + ) + + outcomes = self._run( + check, + sc_items=[{"metadata": {"name": "ebs"}, "provisioner": "ebs.csi.aws.com"}], + csidrivers={"ebs.csi.aws.com": {"metadata": {"name": "ebs.csi.aws.com"}}}, + deployments={"ebs-csi-controller": _deployment_json("ebs-csi-controller", replicas=2, ready=1)}, + daemonsets={"ebs-csi-node": _daemonset_json("ebs-csi-node")}, + ) + + assert not check.passed + ctrl = outcomes["controller-deployment-healthy[ebs.csi.aws.com]"] + assert ctrl["passed"] is False + assert "readyReplicas=1" in ctrl["message"] + + def test_controller_replicas_below_min_fails(self) -> None: + check = self._make( + { + "drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}], + "min_controller_replicas": 2, + } + ) + + outcomes = self._run( + check, + sc_items=[{"metadata": {"name": "ebs"}, "provisioner": "ebs.csi.aws.com"}], + csidrivers={"ebs.csi.aws.com": {"metadata": {"name": "ebs.csi.aws.com"}}}, + deployments={"ebs-csi-controller": _deployment_json("ebs-csi-controller", replicas=1, ready=1)}, + daemonsets={"ebs-csi-node": _daemonset_json("ebs-csi-node")}, + ) + + assert not check.passed + ctrl = outcomes["controller-deployment-healthy[ebs.csi.aws.com]"] + assert ctrl["passed"] is False + assert "min_controller_replicas=2" in ctrl["message"] + + def test_daemonset_no_nodes_scheduled_fails(self) -> None: + check = self._make( + {"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]} + ) + + outcomes = self._run( + check, + sc_items=[{"metadata": {"name": "ebs"}, "provisioner": "ebs.csi.aws.com"}], + csidrivers={"ebs.csi.aws.com": {"metadata": {"name": "ebs.csi.aws.com"}}}, + deployments={"ebs-csi-controller": _deployment_json("ebs-csi-controller")}, + daemonsets={"ebs-csi-node": _daemonset_json("ebs-csi-node", desired=0, available=0, ready=0)}, + ) + + assert not check.passed + ds = outcomes["node-daemonset-healthy[ebs.csi.aws.com]"] + assert ds["passed"] is False + assert "desiredNumberScheduled=0" in ds["message"] + + def test_daemonset_unavailable_fails(self) -> None: + check = self._make( + {"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]} + ) + + outcomes = self._run( + check, + sc_items=[{"metadata": {"name": "ebs"}, "provisioner": "ebs.csi.aws.com"}], + csidrivers={"ebs.csi.aws.com": {"metadata": {"name": "ebs.csi.aws.com"}}}, + deployments={"ebs-csi-controller": _deployment_json("ebs-csi-controller")}, + daemonsets={"ebs-csi-node": _daemonset_json("ebs-csi-node", desired=3, available=2, ready=2)}, + ) + + assert not check.passed + ds = outcomes["node-daemonset-healthy[ebs.csi.aws.com]"] + assert ds["passed"] is False + assert "available=2" in ds["message"] + + def test_missing_deployment_name_fails(self) -> None: + # `workloads` present but controller.deployment unset: the controller + # subtest fails with a config-key hint and no deployment query runs. + check = self._make( + { + "drivers": [ + { + "storage_classes": ["ebs"], + "workloads": {"controller": {}, "node": {"daemonset": "ebs-csi-node"}}, + } + ] + } + ) + + outcomes = self._run( + check, + sc_items=[{"metadata": {"name": "ebs"}, "provisioner": "ebs.csi.aws.com"}], + csidrivers={"ebs.csi.aws.com": {"metadata": {"name": "ebs.csi.aws.com"}}}, + daemonsets={"ebs-csi-node": _daemonset_json("ebs-csi-node")}, + ) + + assert not check.passed + ctrl = outcomes["controller-deployment-healthy[ebs.csi.aws.com]"] + assert ctrl["passed"] is False + assert "workloads.controller.deployment must be set" in ctrl["message"] + + def test_duplicate_storage_classes_resolve_to_single_provisioner(self) -> None: + # Two StorageClasses sharing a provisioner (AWS maps shared_fs and nfs to + # the same EFS class) collapse to one set of per-provisioner subtests. + check = self._make( + {"drivers": [{"storage_classes": ["efs"]}, {"storage_classes": ["efs"]}]} + ) + + outcomes = self._run( + check, + sc_items=[{"metadata": {"name": "efs"}, "provisioner": "efs.csi.aws.com"}], + csidrivers={"efs.csi.aws.com": {"metadata": {"name": "efs.csi.aws.com"}}}, + ) + + assert check.passed, check._error + registered = [n for n in outcomes if n.startswith("csidriver-registered")] + assert registered == ["csidriver-registered[efs.csi.aws.com]"] + + def test_conflicting_workloads_for_same_provisioner_fails(self) -> None: + # Two StorageClasses resolve to the same provisioner but supply different + # workloads blocks: instead of silently using the first, the check fails. + check = self._make( + { + "drivers": [ + {"storage_classes": ["efs-a"], "workloads": self._workloads(namespace="ns-a")}, + {"storage_classes": ["efs-b"], "workloads": self._workloads(namespace="ns-b")}, + ] + } + ) + + outcomes = self._run( + check, + sc_items=[ + {"metadata": {"name": "efs-a"}, "provisioner": "efs.csi.aws.com"}, + {"metadata": {"name": "efs-b"}, "provisioner": "efs.csi.aws.com"}, + ], + csidrivers={"efs.csi.aws.com": {"metadata": {"name": "efs.csi.aws.com"}}}, + ) + + assert not check.passed + ctrl = outcomes["controller-deployment-healthy[efs.csi.aws.com]"] + assert ctrl["passed"] is False + assert "Conflicting" in ctrl["message"] From e4bcdfe04f6031dcf53039873e6abb0aa3f97a79 Mon Sep 17 00:00:00 2001 From: Alexandra Bueno Date: Sun, 5 Jul 2026 21:19:59 -0600 Subject: [PATCH 2/2] fix: ruff-format updated files Signed-off-by: Alexandra Bueno --- .../src/isvtest/validations/k8s_storage.py | 6 ++---- isvtest/tests/test_k8s_storage.py | 20 +++++-------------- 2 files changed, 7 insertions(+), 19 deletions(-) diff --git a/isvtest/src/isvtest/validations/k8s_storage.py b/isvtest/src/isvtest/validations/k8s_storage.py index 9582509c..fb9bf364 100644 --- a/isvtest/src/isvtest/validations/k8s_storage.py +++ b/isvtest/src/isvtest/validations/k8s_storage.py @@ -1996,8 +1996,7 @@ def run(self) -> None: return all_sc_items = _load_items(sc_result.stdout) sc_to_provisioner: dict[str, str] = { - (item.get("metadata") or {}).get("name", ""): str(item.get("provisioner") or "") - for item in all_sc_items + (item.get("metadata") or {}).get("name", ""): str(item.get("provisioner") or "") for item in all_sc_items } any_failed = False @@ -2193,8 +2192,7 @@ def _get_workload( if not name: return {}, f"{name_config_key} must be set for the {kind} subtest" cmd = ( - f"{kubectl_base} get {kind} {shlex.quote(name)} " - f"-n {shlex.quote(namespace)} --ignore-not-found=true -o json" + f"{kubectl_base} get {kind} {shlex.quote(name)} -n {shlex.quote(namespace)} --ignore-not-found=true -o json" ) result = self.run_command(cmd) if result.exit_code != 0: diff --git a/isvtest/tests/test_k8s_storage.py b/isvtest/tests/test_k8s_storage.py index 8ab05a42..19696edd 100644 --- a/isvtest/tests/test_k8s_storage.py +++ b/isvtest/tests/test_k8s_storage.py @@ -2293,9 +2293,7 @@ def test_skips_whole_check_when_no_storage_classes_resolve(self) -> None: assert check._subtest_results == [] def test_healthy_controller_and_daemonset_pass(self) -> None: - check = self._make( - {"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]} - ) + check = self._make({"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]}) outcomes = self._run( check, @@ -2343,9 +2341,7 @@ def test_storageclass_list_command_failure_sets_failed(self) -> None: assert check._subtest_results == [] def test_controller_deployment_not_ready_fails(self) -> None: - check = self._make( - {"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]} - ) + check = self._make({"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]}) outcomes = self._run( check, @@ -2382,9 +2378,7 @@ def test_controller_replicas_below_min_fails(self) -> None: assert "min_controller_replicas=2" in ctrl["message"] def test_daemonset_no_nodes_scheduled_fails(self) -> None: - check = self._make( - {"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]} - ) + check = self._make({"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]}) outcomes = self._run( check, @@ -2400,9 +2394,7 @@ def test_daemonset_no_nodes_scheduled_fails(self) -> None: assert "desiredNumberScheduled=0" in ds["message"] def test_daemonset_unavailable_fails(self) -> None: - check = self._make( - {"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]} - ) + check = self._make({"drivers": [{"storage_classes": ["ebs"], "workloads": self._workloads()}]}) outcomes = self._run( check, @@ -2446,9 +2438,7 @@ def test_missing_deployment_name_fails(self) -> None: def test_duplicate_storage_classes_resolve_to_single_provisioner(self) -> None: # Two StorageClasses sharing a provisioner (AWS maps shared_fs and nfs to # the same EFS class) collapse to one set of per-provisioner subtests. - check = self._make( - {"drivers": [{"storage_classes": ["efs"]}, {"storage_classes": ["efs"]}]} - ) + check = self._make({"drivers": [{"storage_classes": ["efs"]}, {"storage_classes": ["efs"]}]}) outcomes = self._run( check,