diff --git a/docs/guides/remote-deployment.md b/docs/guides/remote-deployment.md index 1bd553258..b6879265e 100644 --- a/docs/guides/remote-deployment.md +++ b/docs/guides/remote-deployment.md @@ -30,6 +30,8 @@ Set these on the local machine. The upload variables are read here, since | `ISV_CLIENT_SECRET` | Required for result upload to ISV Lab Service | locally | | `NGC_API_KEY` | Required for NIM model benchmarks | forwarded | | `ISVTEST_INCLUDE_UNRELEASED` | Include checks not yet in `released_tests.json` | forwarded | +| `ISVTEST_BREAKFIX_ALLOW_MUTATION` | Explicitly allow a mutating breakfix validation | forwarded | +| `ISVTEST_BREAKFIX_NODE` | Dedicated Kubernetes node selected for breakfix validation | forwarded | Anything else the tests need has to reach the target another way - a config file under `isvctl/` travels in the deployment archive, so `-f` overrides are the @@ -85,6 +87,23 @@ Pass extra pytest arguments after `--`: uv run isvctl deploy run -f isvctl/configs/suites/slurm.yaml -- -v -s -k "test_name" ``` +### Running the Kubernetes Cordon Breakfix Check + +The cordon reference requires explicit mutation consent and an exact dedicated +node on multi-node clusters. It restores the node's schedulability before the +run exits. + +```bash +ISVTEST_INCLUDE_UNRELEASED=1 \ +ISVTEST_BREAKFIX_ALLOW_MUTATION=1 \ +ISVTEST_BREAKFIX_NODE= \ +uv run isvctl deploy run \ + -j -u ubuntu \ + -f isvctl/configs/providers/kubernetes-breakfix.yaml \ + --phase test --no-upload \ + -- -v -s -k CordonNodeCheck +``` + ### With ISV Lab Service Integration Upload results to the ISV Lab Service: diff --git a/isvctl/configs/providers/kubernetes-breakfix.yaml b/isvctl/configs/providers/kubernetes-breakfix.yaml new file mode 100644 index 000000000..bed116d29 --- /dev/null +++ b/isvctl/configs/providers/kubernetes-breakfix.yaml @@ -0,0 +1,37 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Explicitly mutating Kubernetes BFX01-04 reference configuration. +# +# This uses the current kubectl context and deliberately has no cluster setup +# or teardown lifecycle. A single-node Minikube context can be selected +# automatically. Multi-node contexts, including DSX, require an explicit node: +# +# export ISVTEST_INCLUDE_UNRELEASED=1 +# export ISVTEST_BREAKFIX_ALLOW_MUTATION=1 +# export ISVTEST_BREAKFIX_NODE= +# isvctl test run -f isvctl/configs/providers/kubernetes-breakfix.yaml \ +# --label breakfix -- -v -s -k CordonNodeCheck + +import: ../suites/k8s.yaml + +version: "1.0" + +commands: + kubernetes: + phases: ["test"] + steps: + - name: cordon_node + phase: test + command: "python shared/breakfix/cordon_node.py" + args: + - "--node={{ env.ISVTEST_BREAKFIX_NODE | default('', true) }}" + timeout: 1200 + requires_available_validations: + - CordonNodeCheck + +tests: + description: "Explicit Kubernetes BFX01-04 cordon validation" + + settings: + show_skipped_tests: true diff --git a/isvctl/configs/providers/shared/breakfix/cordon_node.py b/isvctl/configs/providers/shared/breakfix/cordon_node.py new file mode 100644 index 000000000..70686c996 --- /dev/null +++ b/isvctl/configs/providers/shared/breakfix/cordon_node.py @@ -0,0 +1,547 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Exercise Kubernetes node cordoning semantics for BFX01-04.""" + +from __future__ import annotations + +import argparse +import json +import os +import shlex +import subprocess +import sys +import time +import uuid +from dataclasses import dataclass +from typing import Any + +DEFAULT_IMAGE = "registry.k8s.io/pause:3.10" +DEFAULT_COMMAND_TIMEOUT_SECONDS = 30.0 +DEFAULT_REQUEST_TIMEOUT_SECONDS = 15.0 +OWNER_ANNOTATION = "isvtest.nvidia.com/bfx01-04-owner" +OWNER_ANNOTATION_PATH = "/metadata/annotations/isvtest.nvidia.com~1bfx01-04-owner" +MUTATION_OPT_IN_ENV = "ISVTEST_BREAKFIX_ALLOW_MUTATION" +UNCORDON_ATTEMPTS = 3 + + +class CordonTestError(RuntimeError): + """Raised when the cordon workflow cannot prove the required behavior.""" + + +class KubectlTimeoutError(CordonTestError): + """Raised when a kubectl process exceeds its finite deadline.""" + + +@dataclass(frozen=True) +class NodeSelection: + """A Ready node snapshot used for the atomic cordon claim.""" + + name: str + hostname: str + resource_version: str + spec: dict[str, Any] + annotations_present: bool + tolerations: list[dict[str, str]] + + +@dataclass(frozen=True) +class CordonOwnership: + """The unique marker authorizing this run to restore a node.""" + + node_name: str + token: str + + +def _kubectl_command() -> list[str]: + """Return the configured kubectl-compatible command prefix.""" + configured = os.environ.get("KUBECTL", "kubectl") + try: + command = shlex.split(configured) + except ValueError as exc: + raise CordonTestError(f"Invalid KUBECTL value: {exc}") from exc + if not command: + raise CordonTestError("KUBECTL must not be blank") + return command + + +def _command_detail(completed: subprocess.CompletedProcess[str]) -> str: + """Return bounded stderr/stdout detail for a failed command.""" + detail = (completed.stderr or completed.stdout).strip() + return detail[-500:] if detail else "command failed without output" + + +def _run( + kubectl: list[str], + *args: str, + input_text: str | None = None, + check: bool = True, + command_timeout_seconds: float = DEFAULT_COMMAND_TIMEOUT_SECONDS, + request_timeout_seconds: float = DEFAULT_REQUEST_TIMEOUT_SECONDS, +) -> subprocess.CompletedProcess[str]: + """Run one bounded kubectl command and translate process failures.""" + if command_timeout_seconds <= 0 or request_timeout_seconds <= 0: + raise CordonTestError("Kubectl command and request timeouts must be greater than zero") + command = [*kubectl, *args, f"--request-timeout={request_timeout_seconds:g}s"] + try: + completed = subprocess.run( + command, + input=input_text, + capture_output=True, + text=True, + check=False, + timeout=command_timeout_seconds, + ) + except subprocess.TimeoutExpired as exc: + raise KubectlTimeoutError( + f"kubectl {' '.join(args)} timed out after {command_timeout_seconds:g} seconds" + ) from exc + except OSError as exc: + raise CordonTestError(f"Unable to run {' '.join(kubectl)}: {exc}") from exc + if check and completed.returncode != 0: + raise CordonTestError(f"kubectl {' '.join(args)} failed: {_command_detail(completed)}") + return completed + + +def _json_output(completed: subprocess.CompletedProcess[str], resource: str) -> dict[str, Any]: + """Parse one kubectl JSON object.""" + try: + payload = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise CordonTestError(f"kubectl returned invalid JSON for {resource}") from exc + if not isinstance(payload, dict): + raise CordonTestError(f"kubectl returned a non-object for {resource}") + return payload + + +def _node_is_available(node: dict[str, Any]) -> bool: + """Return whether a node is Ready, schedulable, and unclaimed.""" + conditions = node.get("status", {}).get("conditions", []) + ready = any(item.get("type") == "Ready" and item.get("status") == "True" for item in conditions) + metadata = node.get("metadata", {}) + annotations = metadata.get("annotations") or {} + unclaimed = isinstance(annotations, dict) and OWNER_ANNOTATION not in annotations + return ready and not node.get("spec", {}).get("unschedulable", False) and unclaimed + + +def _select_node(kubectl: list[str], requested_node: str | None) -> NodeSelection: + """Select a schedulable Ready node and retain its concurrency snapshot.""" + payload = _json_output(_run(kubectl, "get", "nodes", "-o", "json"), "node list") + items = payload.get("items") + if not isinstance(items, list): + raise CordonTestError("kubectl node list is missing items") + + candidates = [node for node in items if isinstance(node, dict) and _node_is_available(node)] + if requested_node: + candidates = [node for node in candidates if node.get("metadata", {}).get("name") == requested_node] + if not candidates: + raise CordonTestError(f"Requested node {requested_node!r} is not Ready, schedulable, and unclaimed") + if not candidates: + raise CordonTestError("No Ready, schedulable, unclaimed node is available for the cordon test") + if not requested_node and len(candidates) != 1: + raise CordonTestError("Multiple Ready, schedulable nodes are available; pass --node with a dedicated test node") + + node = candidates[0] + metadata = node.get("metadata", {}) + spec = node.get("spec", {}) + name = metadata.get("name") + hostname = metadata.get("labels", {}).get("kubernetes.io/hostname") + resource_version = metadata.get("resourceVersion") + if not isinstance(name, str) or not name: + raise CordonTestError("Selected node is missing metadata.name") + if not isinstance(hostname, str) or not hostname: + raise CordonTestError(f"Node {name!r} is missing the kubernetes.io/hostname label") + if not isinstance(resource_version, str) or not resource_version: + raise CordonTestError(f"Node {name!r} is missing metadata.resourceVersion") + if not isinstance(spec, dict): + raise CordonTestError(f"Node {name!r} has invalid spec data") + annotations = metadata.get("annotations") + if annotations is not None and not isinstance(annotations, dict): + raise CordonTestError(f"Node {name!r} has invalid metadata.annotations data") + + tolerations: list[dict[str, str]] = [] + for taint in spec.get("taints", []): + key = taint.get("key") + effect = taint.get("effect") + if not isinstance(key, str) or effect not in {"NoSchedule", "NoExecute"}: + continue + if key == "node.kubernetes.io/unschedulable": + continue + tolerations.append( + { + "key": key, + "operator": "Equal", + "value": str(taint.get("value", "")), + "effect": effect, + } + ) + return NodeSelection( + name=name, + hostname=hostname, + resource_version=resource_version, + spec=spec, + annotations_present=annotations is not None, + tolerations=tolerations, + ) + + +def _claim_node(kubectl: list[str], selection: NodeSelection, token: str) -> CordonOwnership: + """Atomically mark a selected node unschedulable and record ownership.""" + ownership = CordonOwnership(node_name=selection.name, token=token) + patch: list[dict[str, Any]] = [ + { + "op": "test", + "path": "/metadata/resourceVersion", + "value": selection.resource_version, + } + ] + if "unschedulable" in selection.spec: + patch.append({"op": "test", "path": "/spec/unschedulable", "value": False}) + else: + # The field is optional; testing the full snapshot proves it is still + # absent (and therefore false) when the conditional patch is applied. + patch.append({"op": "test", "path": "/spec", "value": selection.spec}) + if selection.annotations_present: + patch.append({"op": "add", "path": OWNER_ANNOTATION_PATH, "value": token}) + else: + patch.append({"op": "add", "path": "/metadata/annotations", "value": {OWNER_ANNOTATION: token}}) + patch.append({"op": "add", "path": "/spec/unschedulable", "value": True}) + + try: + completed = _run( + kubectl, + "patch", + "node", + selection.name, + "--type=json", + "-p", + json.dumps(patch, separators=(",", ":")), + check=False, + ) + except KubectlTimeoutError: + # The API may have committed the atomic patch before the client lost + # its response. Only claim cleanup ownership after observing our unique + # marker and the expected state. + if _claim_is_observed(kubectl, ownership): + return ownership + raise + if completed.returncode != 0: + # kubectl's own --request-timeout exits nonzero instead of raising a + # process timeout. Its request can still have reached the API server, + # so apply the same unique-marker verification before giving up cleanup + # ownership. + if _claim_is_observed(kubectl, ownership): + return ownership + raise CordonTestError(f"Could not atomically claim node {selection.name!r}: {_command_detail(completed)}") + return ownership + + +def _get_node(kubectl: list[str], node_name: str) -> dict[str, Any]: + """Return one node as a JSON object.""" + completed = _run(kubectl, "get", "node", node_name, "-o", "json") + return _json_output(completed, f"node {node_name}") + + +def _owned_by(node: dict[str, Any], token: str) -> bool: + """Return whether a node still carries this run's ownership marker.""" + annotations = node.get("metadata", {}).get("annotations") or {} + return isinstance(annotations, dict) and annotations.get(OWNER_ANNOTATION) == token + + +def _claim_is_observed(kubectl: list[str], ownership: CordonOwnership) -> bool: + """Confirm an ambiguous claim from its unique marker and cordoned state.""" + try: + node = _get_node(kubectl, ownership.node_name) + except CordonTestError: + return False + return _owned_by(node, ownership.token) and node.get("spec", {}).get("unschedulable") is True + + +def _release_node(kubectl: list[str], ownership: CordonOwnership) -> None: + """Conditionally uncordon a node only while this run still owns it.""" + last_error = "conditional patch did not succeed" + for _ in range(UNCORDON_ATTEMPTS): + try: + node = _get_node(kubectl, ownership.node_name) + except CordonTestError as exc: + last_error = str(exc) + continue + metadata = node.get("metadata", {}) + spec = node.get("spec", {}) + annotations = metadata.get("annotations") or {} + owner = annotations.get(OWNER_ANNOTATION) if isinstance(annotations, dict) else None + unschedulable = spec.get("unschedulable", False) if isinstance(spec, dict) else False + if owner is None and not unschedulable: + return + if owner != ownership.token: + raise CordonTestError(f"Node {ownership.node_name!r} ownership changed; leaving schedulability unchanged") + resource_version = metadata.get("resourceVersion") + if not isinstance(resource_version, str) or not resource_version: + raise CordonTestError(f"Node {ownership.node_name!r} is missing metadata.resourceVersion") + + patch: list[dict[str, Any]] = [ + {"op": "test", "path": "/metadata/resourceVersion", "value": resource_version}, + {"op": "test", "path": OWNER_ANNOTATION_PATH, "value": ownership.token}, + ] + if unschedulable: + patch.extend( + [ + {"op": "test", "path": "/spec/unschedulable", "value": True}, + {"op": "replace", "path": "/spec/unschedulable", "value": False}, + ] + ) + patch.append({"op": "remove", "path": OWNER_ANNOTATION_PATH}) + try: + completed = _run( + kubectl, + "patch", + "node", + ownership.node_name, + "--type=json", + "-p", + json.dumps(patch, separators=(",", ":")), + check=False, + ) + except KubectlTimeoutError as exc: + # Re-read on the next attempt. If the patch committed, the missing + # marker plus schedulable state is recognized as successful. + last_error = str(exc) + continue + if completed.returncode == 0: + return + last_error = _command_detail(completed) + raise CordonTestError(f"Could not safely uncordon node {ownership.node_name!r}: {last_error}") + + +def _pod_manifest( + name: str, + namespace: str, + hostname: str, + image: str, + tolerations: list[dict[str, str]], +) -> str: + """Build a minimal long-running pod constrained to one node hostname.""" + return json.dumps( + { + "apiVersion": "v1", + "kind": "Pod", + "metadata": { + "name": name, + "namespace": namespace, + "labels": { + "app.kubernetes.io/managed-by": "isvtest", + "isvtest.nvidia.com/purpose": "bfx01-04", + }, + }, + "spec": { + "restartPolicy": "Never", + "nodeSelector": {"kubernetes.io/hostname": hostname}, + "tolerations": tolerations, + "containers": [{"name": "probe", "image": image}], + }, + } + ) + + +def _get_pod(kubectl: list[str], namespace: str, name: str) -> dict[str, Any]: + """Return one pod as a JSON object.""" + completed = _run(kubectl, "get", "pod", name, "-n", namespace, "-o", "json") + return _json_output(completed, f"pod {namespace}/{name}") + + +def _pod_is_ready_on_node(pod: dict[str, Any], node_name: str) -> bool: + """Return whether a pod is still Ready and bound to the expected node.""" + if pod.get("spec", {}).get("nodeName") != node_name or pod.get("status", {}).get("phase") != "Running": + return False + conditions = pod.get("status", {}).get("conditions", []) + return any(item.get("type") == "Ready" and item.get("status") == "True" for item in conditions) + + +def _pod_is_unschedulable(pod: dict[str, Any]) -> bool: + """Return whether the scheduler explicitly reported the pod unschedulable.""" + if pod.get("spec", {}).get("nodeName"): + return False + conditions = pod.get("status", {}).get("conditions", []) + return any( + item.get("type") == "PodScheduled" and item.get("status") == "False" and item.get("reason") == "Unschedulable" + for item in conditions + ) + + +def _wait_for_unschedulable( + kubectl: list[str], + namespace: str, + name: str, + timeout_seconds: float, + poll_interval_seconds: float, +) -> bool: + """Poll until Kubernetes reports that the new probe cannot be scheduled.""" + deadline = time.monotonic() + timeout_seconds + while True: + if _pod_is_unschedulable(_get_pod(kubectl, namespace, name)): + return True + if time.monotonic() >= deadline: + return False + time.sleep(poll_interval_seconds) + + +def _cleanup( + kubectl: list[str], + namespace: str, + pod_names: list[str], + ownership: CordonOwnership | None, +) -> list[str]: + """Delete probe pods and conditionally restore the owned node.""" + errors: list[str] = [] + for pod_name in pod_names: + try: + completed = _run( + kubectl, + "delete", + "pod", + pod_name, + "-n", + namespace, + "--ignore-not-found=true", + "--wait=false", + check=False, + ) + except CordonTestError as exc: + errors.append(f"delete pod {namespace}/{pod_name}: {exc}") + continue + if completed.returncode != 0: + errors.append(f"delete pod {namespace}/{pod_name}: {_command_detail(completed)}") + if ownership: + try: + _release_node(kubectl, ownership) + except CordonTestError as exc: + errors.append(f"uncordon node {ownership.node_name}: {exc}") + return errors + + +def _parser() -> argparse.ArgumentParser: + """Build the command-line parser.""" + parser = argparse.ArgumentParser(description="Cordon a node and verify Kubernetes scheduling behavior") + parser.add_argument("--node", help="Specific Ready, schedulable node to test") + parser.add_argument("--namespace", default="default", help="Namespace for temporary probe pods") + parser.add_argument("--image", default=DEFAULT_IMAGE, help="Container image for temporary probe pods") + parser.add_argument("--timeout-seconds", type=float, default=120, help="Timeout for each scheduling assertion") + parser.add_argument("--poll-interval-seconds", type=float, default=2, help="Pending-pod polling interval") + return parser + + +def main() -> int: + """Run the reversible cordon test and emit its provider-neutral JSON result.""" + args = _parser().parse_args() + operation: dict[str, Any] = { + "cordoned": False, + "new_workloads_blocked": False, + "existing_workloads_running": False, + } + result: dict[str, Any] = {"success": False, "platform": "kubernetes", "test_name": "cordon_node"} + kubectl: list[str] = [] + created_pods: list[str] = [] + ownership: CordonOwnership | None = None + + try: + if args.timeout_seconds <= 0 or args.poll_interval_seconds <= 0: + raise CordonTestError("Timeout and poll interval must be greater than zero") + if os.environ.get(MUTATION_OPT_IN_ENV) != "1": + raise CordonTestError( + f"Refusing to mutate cluster state; explicitly set {MUTATION_OPT_IN_ENV}=1 for BFX01-04" + ) + kubectl = _kubectl_command() + selection = _select_node(kubectl, args.node) + operation["node_id"] = selection.name + run_id = uuid.uuid4().hex + suffix = run_id + existing_pod = f"isvtest-bfx-existing-{suffix}" + blocked_pod = f"isvtest-bfx-blocked-{suffix}" + + created_pods.append(existing_pod) + _run( + kubectl, + "create", + "-f", + "-", + input_text=_pod_manifest( + existing_pod, + args.namespace, + selection.hostname, + args.image, + selection.tolerations, + ), + ) + _run( + kubectl, + "wait", + "--for=condition=Ready", + f"pod/{existing_pod}", + "-n", + args.namespace, + f"--timeout={args.timeout_seconds:g}s", + command_timeout_seconds=args.timeout_seconds + DEFAULT_REQUEST_TIMEOUT_SECONDS + 5, + request_timeout_seconds=args.timeout_seconds + 5, + ) + + # Pod startup can take long enough for kubelet status updates to advance + # resourceVersion. Refresh immediately before the conditional claim. + claim_selection = _select_node(kubectl, selection.name) + # Retain a conditional cleanup candidate before issuing the PATCH. If + # the API commits the claim but both the PATCH response and immediate + # verification GET are lost, cleanup can later release the node only + # after observing this exact unique annotation. + ownership = CordonOwnership(selection.name, f"isvtest-bfx01-04-{run_id}") + _claim_node(kubectl, claim_selection, ownership.token) + node = _get_node(kubectl, selection.name) + operation["cordoned"] = node.get("spec", {}).get("unschedulable") is True and _owned_by(node, ownership.token) + if not operation["cordoned"]: + raise CordonTestError(f"Node {selection.name!r} was not marked unschedulable by this run") + + operation["existing_workloads_running"] = _pod_is_ready_on_node( + _get_pod(kubectl, args.namespace, existing_pod), selection.name + ) + if not operation["existing_workloads_running"]: + raise CordonTestError("Existing probe pod did not remain Ready on the cordoned node") + + created_pods.append(blocked_pod) + _run( + kubectl, + "create", + "-f", + "-", + input_text=_pod_manifest( + blocked_pod, + args.namespace, + selection.hostname, + args.image, + selection.tolerations, + ), + ) + operation["new_workloads_blocked"] = _wait_for_unschedulable( + kubectl, + args.namespace, + blocked_pod, + args.timeout_seconds, + args.poll_interval_seconds, + ) + if not operation["new_workloads_blocked"]: + raise CordonTestError("New probe pod was not confirmed unschedulable on the cordoned node") + result["success"] = True + except CordonTestError as exc: + result["error"] = str(exc) + finally: + cleanup_errors = _cleanup(kubectl, args.namespace, created_pods, ownership) if kubectl else [] + if cleanup_errors: + result["success"] = False + result["cleanup_errors"] = cleanup_errors + result.setdefault("error", "Cordon test cleanup failed") + + result["operation"] = operation + print(json.dumps(result, indent=2)) + return 0 if result["success"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/isvctl/configs/suites/README.md b/isvctl/configs/suites/README.md index 6399a2691..49c676f70 100644 --- a/isvctl/configs/suites/README.md +++ b/isvctl/configs/suites/README.md @@ -295,9 +295,10 @@ volume. The three test-phase steps all reuse that fixture. | `setup` | setup | `providers/my-isv/scripts/k8s/setup.sh` | | `teardown` | teardown | `providers/my-isv/scripts/k8s/teardown.sh` | | `reset_gpus` | test | `providers/my-isv/scripts/breakfix/reset_gpus.py` (BFX01-01) | -| `cordon_node` | test | `providers/my-isv/scripts/breakfix/cordon_node.py` (BFX01-04) | +| `cordon_node` | test | `providers/my-isv/scripts/breakfix/cordon_node.py` template; `providers/shared/breakfix/cordon_node.py` reference used by `providers/kubernetes-breakfix.yaml` (BFX01-04) | Validations use `kubectl` directly (or a custom CLI via the `KUBECTL` env var): node counts, GPU operator, pod health, NCCL/NIM workloads. Break-fix cordon and GPU reset are optional provider steps. +The shared cordon reference requires `ISVTEST_BREAKFIX_ALLOW_MUTATION=1`; multi-node clusters also require an explicit `ISVTEST_BREAKFIX_NODE` dedicated to the test. ### Slurm (`slurm.yaml`) diff --git a/isvctl/src/isvctl/cli/deploy.py b/isvctl/src/isvctl/cli/deploy.py index ad5090fd3..ccf09365d 100644 --- a/isvctl/src/isvctl/cli/deploy.py +++ b/isvctl/src/isvctl/cli/deploy.py @@ -52,6 +52,16 @@ logger = logging.getLogger(__name__) +# Explicit per-run controls that the remote validation process needs from the +# caller. These are intentionally allow-listed instead of forwarding the whole +# local environment. +REMOTE_TEST_ENV_VARS = ( + INCLUDE_UNRELEASED_ENV, + "ISVTEST_BREAKFIX_ALLOW_MUTATION", + "ISVTEST_BREAKFIX_NODE", +) + + # Default paths to include in the deployment archive DEFAULT_ARCHIVE_PATHS = [ "isvtest/", @@ -86,18 +96,20 @@ def _capability_option(capability: str | None) -> str: def _remote_env_assignments() -> str: """Render the environment the remote ``test run`` needs from this process. - Only values the target cannot obtain on its own: a credential and the - release gate, both set per invocation by whoever runs the deploy. Quoted - because they end up on a shell command line. Path-valued variables are - deliberately not forwarded, since they name files that exist only here. + Only explicit per-invocation values the target cannot obtain on its own are + forwarded: a credential, the release gate, and breakfix mutation controls. + Values are quoted because they end up on a shell command line. Path-valued + variables are deliberately not forwarded, since they name files that exist + only here. """ forwarded: dict[str, str] = {} ngc_api_key = get_ngc_api_key() if ngc_api_key: forwarded["NGC_API_KEY"] = ngc_api_key - include_unreleased = os.environ.get(INCLUDE_UNRELEASED_ENV, "") - if include_unreleased: - forwarded[INCLUDE_UNRELEASED_ENV] = include_unreleased + for name in REMOTE_TEST_ENV_VARS: + value = os.environ.get(name, "") + if value: + forwarded[name] = value return " ".join(f"{name}={shlex.quote(value)}" for name, value in forwarded.items()) diff --git a/isvctl/tests/test_deploy_passthrough.py b/isvctl/tests/test_deploy_passthrough.py index 45d69f7b4..ee54b4d32 100644 --- a/isvctl/tests/test_deploy_passthrough.py +++ b/isvctl/tests/test_deploy_passthrough.py @@ -8,6 +8,19 @@ from isvctl.cli.deploy import _pytest_passthrough, _remote_env_assignments +REMOTE_TEST_ENV_VARS = ( + "NGC_API_KEY", + "NGC_NIM_API_KEY", + INCLUDE_UNRELEASED_ENV, + "ISVTEST_BREAKFIX_ALLOW_MUTATION", + "ISVTEST_BREAKFIX_NODE", +) + + +def _clear_remote_test_env(monkeypatch: pytest.MonkeyPatch) -> None: + for name in REMOTE_TEST_ENV_VARS: + monkeypatch.delenv(name, raising=False) + def test_passthrough_carries_the_separator() -> None: """Without `--`, `test run` reads a bare pytest flag as an unknown isvctl option.""" @@ -26,7 +39,7 @@ def test_passthrough_quotes_a_multi_word_expression() -> None: def test_release_gate_reaches_the_remote_run(monkeypatch: pytest.MonkeyPatch) -> None: """Without this the remote run silently skips every unreleased check.""" - monkeypatch.delenv("NGC_NIM_API_KEY", raising=False) + _clear_remote_test_env(monkeypatch) monkeypatch.setenv("NGC_API_KEY", "secret key") monkeypatch.setenv(INCLUDE_UNRELEASED_ENV, "1") @@ -35,16 +48,23 @@ def test_release_gate_reaches_the_remote_run(monkeypatch: pytest.MonkeyPatch) -> def test_ngc_key_alias_is_forwarded_under_the_canonical_name(monkeypatch: pytest.MonkeyPatch) -> None: """The target reads NGC_API_KEY, whichever name it was supplied under here.""" - monkeypatch.delenv("NGC_API_KEY", raising=False) - monkeypatch.delenv(INCLUDE_UNRELEASED_ENV, raising=False) + _clear_remote_test_env(monkeypatch) monkeypatch.setenv("NGC_NIM_API_KEY", "nim-key") assert _remote_env_assignments() == "NGC_API_KEY=nim-key" +def test_breakfix_mutation_controls_reach_the_remote_run(monkeypatch: pytest.MonkeyPatch) -> None: + """A deploy must preserve both explicit mutation consent and its exact target.""" + _clear_remote_test_env(monkeypatch) + monkeypatch.setenv("ISVTEST_BREAKFIX_ALLOW_MUTATION", "1") + monkeypatch.setenv("ISVTEST_BREAKFIX_NODE", "dedicated node") + + assert _remote_env_assignments() == "ISVTEST_BREAKFIX_ALLOW_MUTATION=1 ISVTEST_BREAKFIX_NODE='dedicated node'" + + def test_nothing_is_forwarded_when_nothing_is_set(monkeypatch: pytest.MonkeyPatch) -> None: """An empty assignment list must not leave a stray token on the command line.""" - for name in ("NGC_API_KEY", "NGC_NIM_API_KEY", INCLUDE_UNRELEASED_ENV): - monkeypatch.delenv(name, raising=False) + _clear_remote_test_env(monkeypatch) assert _remote_env_assignments() == "" diff --git a/isvctl/tests/test_shared_cordon_node.py b/isvctl/tests/test_shared_cordon_node.py new file mode 100644 index 000000000..5e3e08ed1 --- /dev/null +++ b/isvctl/tests/test_shared_cordon_node.py @@ -0,0 +1,641 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the shared BFX01-04 Kubernetes cordon reference.""" + +from __future__ import annotations + +import importlib.util +import json +import subprocess +import sys +from pathlib import Path +from types import ModuleType +from typing import Any + +import pytest +import yaml + +from isvctl.config.merger import merge_yaml_files +from isvctl.config.schema import RunConfig +from isvctl.orchestrator.context import Context +from isvctl.orchestrator.step_executor import StepExecutor + +ISVCTL_ROOT = Path(__file__).resolve().parents[1] +CORDON_SCRIPT = ISVCTL_ROOT / "configs" / "providers" / "shared" / "breakfix" / "cordon_node.py" +MINIKUBE_CONFIG = ISVCTL_ROOT / "configs" / "providers" / "minikube.yaml" +BREAKFIX_CONFIG = ISVCTL_ROOT / "configs" / "providers" / "kubernetes-breakfix.yaml" + + +@pytest.fixture(autouse=True) +def _allow_explicit_test_mutation(monkeypatch: pytest.MonkeyPatch) -> None: + """Unit workflows opt in explicitly; production callers must do the same.""" + monkeypatch.setenv("ISVTEST_BREAKFIX_ALLOW_MUTATION", "1") + + +def _load_script() -> ModuleType: + """Load the shared cordon script as a module for direct testing.""" + spec = importlib.util.spec_from_file_location("test_shared_cordon_node_script", CORDON_SCRIPT) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _completed( + args: tuple[str, ...], + *, + payload: dict[str, Any] | None = None, + error: str = "", +) -> subprocess.CompletedProcess[str]: + """Build a completed kubectl command with optional JSON output.""" + return subprocess.CompletedProcess( + args=["kubectl", *args], + returncode=1 if error else 0, + stdout=json.dumps(payload) if payload is not None else "", + stderr=error, + ) + + +def _node( + *, + unschedulable: bool = False, + resource_version: str = "10", + owner: str | None = None, +) -> dict[str, Any]: + """Return one Ready fake node with concurrency metadata.""" + annotations = {} if owner is None else {"isvtest.nvidia.com/bfx01-04-owner": owner} + return { + "metadata": { + "name": "worker-1", + "resourceVersion": resource_version, + "annotations": annotations, + "labels": {"kubernetes.io/hostname": "worker-1-host"}, + }, + "spec": {"unschedulable": unschedulable}, + "status": {"conditions": [{"type": "Ready", "status": "True"}]}, + } + + +def _running_pod() -> dict[str, Any]: + """Return a Ready pod bound to the selected node.""" + return { + "spec": {"nodeName": "worker-1"}, + "status": {"phase": "Running", "conditions": [{"type": "Ready", "status": "True"}]}, + } + + +def _unschedulable_pod() -> dict[str, Any]: + """Return an unbound pod rejected by the scheduler.""" + return { + "spec": {}, + "status": { + "phase": "Pending", + "conditions": [{"type": "PodScheduled", "status": "False", "reason": "Unschedulable"}], + }, + } + + +def _patch_from_args(args: tuple[str, ...]) -> list[dict[str, Any]]: + """Decode the JSON Patch argument from a fake kubectl call.""" + return json.loads(args[args.index("-p") + 1]) + + +def test_explicit_breakfix_config_wires_the_shared_reference() -> None: + """Keep the mutating reference behind an explicitly selected config.""" + config = yaml.safe_load(BREAKFIX_CONFIG.read_text()) + steps = config["commands"]["kubernetes"]["steps"] + cordon_step = next(step for step in steps if step["name"] == "cordon_node") + + assert cordon_step["command"] == "python shared/breakfix/cordon_node.py" + assert cordon_step["phase"] == "test" + assert cordon_step["timeout"] == 1200 + assert cordon_step["requires_available_validations"] == ["CordonNodeCheck"] + assert cordon_step["args"] == ["--node={{ env.ISVTEST_BREAKFIX_NODE | default('', true) }}"] + + +def test_empty_node_selection_renders_as_one_safe_argument(monkeypatch: pytest.MonkeyPatch) -> None: + """Single-node runs must render ``--node=`` instead of a dangling flag.""" + monkeypatch.delenv("ISVTEST_BREAKFIX_NODE", raising=False) + config = RunConfig.model_validate(merge_yaml_files([BREAKFIX_CONFIG])) + cordon_step = next(step for step in config.commands["kubernetes"].steps if step.name == "cordon_node") + + rendered = StepExecutor()._render_args(cordon_step.args, Context(config)) + + assert rendered == ["--node="] + + +def test_normal_minikube_config_never_runs_the_mutating_step() -> None: + """An ordinary Minikube validation must not cordon a node implicitly.""" + config = yaml.safe_load(MINIKUBE_CONFIG.read_text()) + steps = config["commands"]["kubernetes"]["steps"] + + assert all(step["name"] != "cordon_node" for step in steps) + + +def test_missing_mutation_opt_in_fails_before_kubectl( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Selecting the explicit config alone must not authorize a cluster mutation.""" + module = _load_script() + monkeypatch.delenv(module.MUTATION_OPT_IN_ENV) + monkeypatch.setattr(sys, "argv", ["cordon_node.py"]) + monkeypatch.setattr( + module, + "_kubectl_command", + lambda: pytest.fail("kubectl must not run without explicit mutation opt-in"), + ) + + assert module.main() == 1 + result = json.loads(capsys.readouterr().out) + + assert result["operation"] == { + "cordoned": False, + "new_workloads_blocked": False, + "existing_workloads_running": False, + } + assert "ISVTEST_BREAKFIX_ALLOW_MUTATION=1" in result["error"] + + +def test_node_taints_are_tolerated_without_bypassing_cordon(monkeypatch: pytest.MonkeyPatch) -> None: + """Probe pods tolerate the selected GPU taint but never the cordon taint.""" + module = _load_script() + node = _node() + node["spec"]["taints"] = [ + {"key": "nvidia.com/gpu", "value": "present", "effect": "NoSchedule"}, + {"key": "node.kubernetes.io/unschedulable", "effect": "NoSchedule"}, + ] + monkeypatch.setattr( + module, + "_run", + lambda kubectl, *args, **kwargs: _completed(args, payload={"items": [node]}), + ) + + selection = module._select_node(["kubectl"], None) + + assert (selection.name, selection.hostname, selection.resource_version) == ("worker-1", "worker-1-host", "10") + assert selection.tolerations == [ + {"key": "nvidia.com/gpu", "operator": "Equal", "value": "present", "effect": "NoSchedule"} + ] + + +def test_multiple_nodes_require_an_explicit_target(monkeypatch: pytest.MonkeyPatch) -> None: + """Never choose a control-plane or shared-cluster node on the user's behalf.""" + module = _load_script() + second = _node() + second["metadata"]["name"] = "worker-2" + second["metadata"]["labels"]["kubernetes.io/hostname"] = "worker-2-host" + monkeypatch.setattr( + module, + "_run", + lambda kubectl, *args, **kwargs: _completed(args, payload={"items": [_node(), second]}), + ) + + with pytest.raises(module.CordonTestError, match="pass --node with a dedicated test node"): + module._select_node(["kubectl"], None) + + +def test_run_bounds_the_process_and_api_request(monkeypatch: pytest.MonkeyPatch) -> None: + """Every kubectl process has both subprocess and API request timeouts.""" + module = _load_script() + observed: dict[str, Any] = {} + + def fake_subprocess_run(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Capture the bounded command invocation.""" + observed["command"] = command + observed.update(kwargs) + return subprocess.CompletedProcess(command, 0, stdout="{}", stderr="") + + monkeypatch.setattr(module.subprocess, "run", fake_subprocess_run) + + module._run( + ["kubectl"], + "get", + "nodes", + command_timeout_seconds=7, + request_timeout_seconds=3, + ) + + assert observed["command"] == ["kubectl", "get", "nodes", "--request-timeout=3s"] + assert observed["timeout"] == 7 + + +def test_run_translates_subprocess_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + """A hung kubectl process becomes a structured workflow error.""" + module = _load_script() + + def fake_subprocess_run(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Model a kubectl process that exceeds its finite deadline.""" + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + + monkeypatch.setattr(module.subprocess, "run", fake_subprocess_run) + + with pytest.raises(module.CordonTestError, match="timed out after 4 seconds"): + module._run(["kubectl"], "get", "nodes", command_timeout_seconds=4) + + +def test_atomic_claim_uses_resource_version_and_unschedulable_snapshot(monkeypatch: pytest.MonkeyPatch) -> None: + """The node claim is one conditional JSON Patch, not idempotent kubectl cordon.""" + module = _load_script() + selection = module.NodeSelection( + name="worker-1", + hostname="worker-1-host", + resource_version="10", + spec={"unschedulable": False}, + annotations_present=True, + tolerations=[], + ) + calls: list[tuple[str, ...]] = [] + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Capture the atomic claim patch.""" + calls.append(args) + return _completed(args) + + monkeypatch.setattr(module, "_run", fake_run) + + ownership = module._claim_node(["kubectl"], selection, "owner-token") + patch = _patch_from_args(calls[0]) + + assert ownership == module.CordonOwnership("worker-1", "owner-token") + assert patch[:2] == [ + {"op": "test", "path": "/metadata/resourceVersion", "value": "10"}, + {"op": "test", "path": "/spec/unschedulable", "value": False}, + ] + assert {"op": "add", "path": module.OWNER_ANNOTATION_PATH, "value": "owner-token"} in patch + assert {"op": "add", "path": "/spec/unschedulable", "value": True} in patch + + +def test_claim_timeout_after_apply_recovers_verified_ownership(monkeypatch: pytest.MonkeyPatch) -> None: + """A lost PATCH response still yields cleanup ownership after a confirming GET.""" + module = _load_script() + selection = module.NodeSelection( + name="worker-1", + hostname="worker-1-host", + resource_version="10", + spec={"unschedulable": False}, + annotations_present=True, + tolerations=[], + ) + calls: list[tuple[str, ...]] = [] + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Model the API committing a claim before the client times out.""" + calls.append(args) + if args[:3] == ("patch", "node", "worker-1"): + raise module.KubectlTimeoutError("claim response timed out") + return _completed(args, payload=_node(unschedulable=True, resource_version="11", owner="owner-token")) + + monkeypatch.setattr(module, "_run", fake_run) + + ownership = module._claim_node(["kubectl"], selection, "owner-token") + + assert ownership == module.CordonOwnership("worker-1", "owner-token") + assert [call[:3] for call in calls] == [("patch", "node", "worker-1"), ("get", "node", "worker-1")] + + +def test_claim_request_timeout_after_apply_recovers_verified_ownership(monkeypatch: pytest.MonkeyPatch) -> None: + """A kubectl request deadline also verifies whether the atomic claim landed.""" + module = _load_script() + selection = module.NodeSelection( + name="worker-1", + hostname="worker-1-host", + resource_version="10", + spec={"unschedulable": False}, + annotations_present=True, + tolerations=[], + ) + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Return kubectl's nonzero form of an ambiguous request timeout.""" + if args[:3] == ("patch", "node", "worker-1"): + return _completed(args, error="context deadline exceeded") + return _completed(args, payload=_node(unschedulable=True, resource_version="11", owner="owner-token")) + + monkeypatch.setattr(module, "_run", fake_run) + + assert module._claim_node(["kubectl"], selection, "owner-token") == module.CordonOwnership( + "worker-1", "owner-token" + ) + + +def test_unverified_claim_timeout_still_runs_conditional_cleanup( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """A lost claim and verification response still retains a safe cleanup candidate.""" + module = _load_script() + calls: list[tuple[str, ...]] = [] + owner_token = "" + node_gets = 0 + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Lose both claim responses, then let cleanup observe and release the claim.""" + nonlocal node_gets, owner_token + calls.append(args) + if args[:4] == ("get", "nodes", "-o", "json"): + return _completed(args, payload={"items": [_node()]}) + if args[:3] == ("patch", "node", "worker-1"): + patch = _patch_from_args(args) + owner_operation = next( + (operation for operation in patch if operation.get("path") == module.OWNER_ANNOTATION_PATH), + None, + ) + if owner_operation and owner_operation["op"] == "add": + owner_token = owner_operation["value"] + raise module.KubectlTimeoutError("claim response timed out") + return _completed(args) + if args[:3] == ("get", "node", "worker-1"): + node_gets += 1 + if node_gets == 1: + raise module.KubectlTimeoutError("claim verification timed out") + return _completed(args, payload=_node(unschedulable=True, resource_version="11", owner=owner_token)) + return _completed(args) + + monkeypatch.setattr(module, "_run", fake_run) + monkeypatch.setattr(module.uuid, "uuid4", lambda: type("Uuid", (), {"hex": "deadbeefcafebabe"})()) + monkeypatch.setattr(sys, "argv", ["cordon_node.py", "--timeout-seconds", "1"]) + + assert module.main() == 1 + result = json.loads(capsys.readouterr().out) + + assert result["error"] == "claim response timed out" + patches = [_patch_from_args(call) for call in calls if call[:3] == ("patch", "node", "worker-1")] + assert len(patches) == 2 + assert {"op": "replace", "path": "/spec/unschedulable", "value": False} in patches[1] + + +def test_concurrent_claim_failure_never_uncordons( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """A stale resourceVersion conflict acquires no cleanup ownership.""" + module = _load_script() + calls: list[tuple[str, ...]] = [] + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Let another actor win between selection and the conditional patch.""" + calls.append(args) + if args[:4] == ("get", "nodes", "-o", "json"): + return _completed(args, payload={"items": [_node()]}) + if args[:3] == ("patch", "node", "worker-1"): + return _completed(args, error="Conflict: object has been modified") + return _completed(args) + + monkeypatch.setattr(module, "_run", fake_run) + monkeypatch.setattr(sys, "argv", ["cordon_node.py", "--timeout-seconds", "1"]) + + assert module.main() == 1 + result = json.loads(capsys.readouterr().out) + + assert "Could not atomically claim" in result["error"] + patches = [_patch_from_args(call) for call in calls if call[:3] == ("patch", "node", "worker-1")] + assert len(patches) == 1 + assert not any( + operation.get("path") == "/spec/unschedulable" + and operation.get("op") in {"add", "replace"} + and operation.get("value") is False + for operation in patches[0] + ) + + +def test_cordon_workflow_proves_requirements_and_conditionally_restores_node( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """A successful run proves all requirements and releases only its own claim.""" + module = _load_script() + calls: list[tuple[str, ...]] = [] + pod_gets = iter([_running_pod(), _unschedulable_pod()]) + owner_token = "" + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Return deterministic Kubernetes state for the happy path.""" + nonlocal owner_token + calls.append(args) + if args[:4] == ("get", "nodes", "-o", "json"): + return _completed(args, payload={"items": [_node()]}) + if args[:3] == ("patch", "node", "worker-1"): + patch = _patch_from_args(args) + owner_operation = next( + (operation for operation in patch if operation.get("path") == module.OWNER_ANNOTATION_PATH), + None, + ) + if owner_operation and owner_operation["op"] == "add": + owner_token = owner_operation["value"] + return _completed(args) + if args[:3] == ("get", "node", "worker-1"): + return _completed(args, payload=_node(unschedulable=True, resource_version="11", owner=owner_token)) + if args[:2] == ("get", "pod"): + return _completed(args, payload=next(pod_gets)) + return _completed(args) + + monkeypatch.setattr(module, "_run", fake_run) + monkeypatch.setattr(module.uuid, "uuid4", lambda: type("Uuid", (), {"hex": "deadbeefcafebabe"})()) + monkeypatch.setattr(sys, "argv", ["cordon_node.py", "--timeout-seconds", "1"]) + + assert module.main() == 0 + result = json.loads(capsys.readouterr().out) + + assert result["success"] is True + assert result["platform"] == "kubernetes" + assert result["operation"] == { + "cordoned": True, + "new_workloads_blocked": True, + "existing_workloads_running": True, + "node_id": "worker-1", + } + patches = [_patch_from_args(call) for call in calls if call[:3] == ("patch", "node", "worker-1")] + assert len(patches) == 2 + assert {"op": "add", "path": "/spec/unschedulable", "value": True} in patches[0] + assert {"op": "test", "path": module.OWNER_ANNOTATION_PATH, "value": owner_token} in patches[1] + assert {"op": "replace", "path": "/spec/unschedulable", "value": False} in patches[1] + assert {"op": "remove", "path": module.OWNER_ANNOTATION_PATH} in patches[1] + assert [call[:3] for call in calls].count(("delete", "pod", "isvtest-bfx-existing-deadbeefcafebabe")) == 1 + assert [call[:3] for call in calls].count(("delete", "pod", "isvtest-bfx-blocked-deadbeefcafebabe")) == 1 + + +def test_changed_ownership_preserves_a_later_cordon(monkeypatch: pytest.MonkeyPatch) -> None: + """Cleanup refuses to uncordon after another actor replaces the ownership marker.""" + module = _load_script() + ownership = module.CordonOwnership("worker-1", "our-token") + monkeypatch.setattr( + module, + "_get_node", + lambda kubectl, node_name: _node(unschedulable=True, owner="later-token"), + ) + monkeypatch.setattr( + module, + "_run", + lambda *args, **kwargs: pytest.fail("ownership change must not issue an uncordon patch"), + ) + + with pytest.raises(module.CordonTestError, match="ownership changed; leaving schedulability unchanged"): + module._release_node(["kubectl"], ownership) + + +def test_release_timeout_after_apply_is_confirmed_by_reread(monkeypatch: pytest.MonkeyPatch) -> None: + """A lost uncordon response is accepted only after observing released state.""" + module = _load_script() + ownership = module.CordonOwnership("worker-1", "our-token") + observed_nodes = iter( + [ + _node(unschedulable=True, resource_version="10", owner="our-token"), + _node(unschedulable=False, resource_version="11"), + ] + ) + patch_calls: list[tuple[str, ...]] = [] + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Model the API applying release before its response is lost.""" + patch_calls.append(args) + raise module.KubectlTimeoutError("release response timed out") + + monkeypatch.setattr(module, "_get_node", lambda kubectl, node_name: next(observed_nodes)) + monkeypatch.setattr(module, "_run", fake_run) + + module._release_node(["kubectl"], ownership) + + assert [call[:3] for call in patch_calls] == [("patch", "node", "worker-1")] + + +def test_release_retries_a_timed_out_state_read(monkeypatch: pytest.MonkeyPatch) -> None: + """A transient cleanup GET timeout must not prevent conditional release.""" + module = _load_script() + ownership = module.CordonOwnership("worker-1", "our-token") + get_calls = 0 + patch_calls: list[tuple[str, ...]] = [] + + def fake_get_node(kubectl: list[str], node_name: str) -> dict[str, Any]: + """Fail the first read and return the owned state on retry.""" + nonlocal get_calls + get_calls += 1 + if get_calls == 1: + raise module.KubectlTimeoutError("cleanup state read timed out") + return _node(unschedulable=True, resource_version="10", owner="our-token") + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Accept the conditional release patch.""" + patch_calls.append(args) + return _completed(args) + + monkeypatch.setattr(module, "_get_node", fake_get_node) + monkeypatch.setattr(module, "_run", fake_run) + + module._release_node(["kubectl"], ownership) + + assert get_calls == 2 + assert [call[:3] for call in patch_calls] == [("patch", "node", "worker-1")] + + +def test_cleanup_records_timeout_and_still_releases_node(monkeypatch: pytest.MonkeyPatch) -> None: + """A timed-out pod delete is reported without skipping node restoration.""" + module = _load_script() + ownership = module.CordonOwnership("worker-1", "our-token") + released: list[Any] = [] + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Model a bounded delete timeout.""" + raise module.CordonTestError("kubectl delete timed out after 30 seconds") + + monkeypatch.setattr(module, "_run", fake_run) + monkeypatch.setattr(module, "_release_node", lambda kubectl, claim: released.append(claim)) + + errors = module._cleanup(["kubectl"], "default", ["probe"], ownership) + + assert errors == ["delete pod default/probe: kubectl delete timed out after 30 seconds"] + assert released == [ownership] + + +def test_create_timeout_preregisters_probe_for_cleanup( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """An ambiguous create timeout still schedules ignore-not-found cleanup.""" + module = _load_script() + calls: list[tuple[str, ...]] = [] + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Time out after the API may have created the first probe pod.""" + calls.append(args) + if args[:4] == ("get", "nodes", "-o", "json"): + return _completed(args, payload={"items": [_node()]}) + if args[:3] == ("create", "-f", "-"): + raise module.KubectlTimeoutError("create response timed out") + return _completed(args) + + monkeypatch.setattr(module, "_run", fake_run) + monkeypatch.setattr(module.uuid, "uuid4", lambda: type("Uuid", (), {"hex": "deadbeefcafebabe"})()) + monkeypatch.setattr(sys, "argv", ["cordon_node.py", "--timeout-seconds", "1"]) + + assert module.main() == 1 + result = json.loads(capsys.readouterr().out) + + assert result["error"] == "create response timed out" + assert ("delete", "pod", "isvtest-bfx-existing-deadbeefcafebabe") in [call[:3] for call in calls] + + +def test_blocked_probe_create_timeout_preregisters_both_pods( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """A timeout creating the post-cordon probe cleans both possible pods.""" + module = _load_script() + calls: list[tuple[str, ...]] = [] + create_count = 0 + owner_token = "" + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + """Reach the second create, then lose its response.""" + nonlocal create_count, owner_token + calls.append(args) + if args[:4] == ("get", "nodes", "-o", "json"): + return _completed(args, payload={"items": [_node()]}) + if args[:3] == ("create", "-f", "-"): + create_count += 1 + if create_count == 2: + raise module.KubectlTimeoutError("blocked create response timed out") + return _completed(args) + if args[:3] == ("patch", "node", "worker-1"): + patch = _patch_from_args(args) + owner_operation = next( + (operation for operation in patch if operation.get("path") == module.OWNER_ANNOTATION_PATH), + None, + ) + if owner_operation and owner_operation["op"] == "add": + owner_token = owner_operation["value"] + return _completed(args) + if args[:3] == ("get", "node", "worker-1"): + return _completed(args, payload=_node(unschedulable=True, resource_version="11", owner=owner_token)) + if args[:2] == ("get", "pod"): + return _completed(args, payload=_running_pod()) + return _completed(args) + + monkeypatch.setattr(module, "_run", fake_run) + monkeypatch.setattr(module.uuid, "uuid4", lambda: type("Uuid", (), {"hex": "deadbeefcafebabe"})()) + monkeypatch.setattr(sys, "argv", ["cordon_node.py", "--timeout-seconds", "1"]) + + assert module.main() == 1 + result = json.loads(capsys.readouterr().out) + + assert result["error"] == "blocked create response timed out" + deleted_pods = [call[:3] for call in calls if call[:2] == ("delete", "pod")] + assert deleted_pods == [ + ("delete", "pod", "isvtest-bfx-existing-deadbeefcafebabe"), + ("delete", "pod", "isvtest-bfx-blocked-deadbeefcafebabe"), + ] + + +def test_requested_precordoned_node_is_rejected(monkeypatch: pytest.MonkeyPatch) -> None: + """The workflow never claims a node already cordoned by someone else.""" + module = _load_script() + monkeypatch.setattr( + module, + "_run", + lambda kubectl, *args, **kwargs: _completed(args, payload={"items": [_node(unschedulable=True)]}), + ) + + with pytest.raises(module.CordonTestError, match="not Ready, schedulable, and unclaimed"): + module._select_node(["kubectl"], "worker-1")