diff --git a/docs/guides/remote-deployment.md b/docs/guides/remote-deployment.md index 1bd553258..521b26cfc 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 break-fix validation | forwarded | +| `ISVTEST_BREAKFIX_NODE` | Exact Kubernetes node selected for break-fix 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,16 @@ Pass extra pytest arguments after `--`: uv run isvctl deploy run -f isvctl/configs/suites/slurm.yaml -- -v -s -k "test_name" ``` +### Node Maintenance Validation + +The BFX01-02 reference uses the NVIDIA Maintenance Operator API on the +target's active Kubernetes context. It never selects a node implicitly and +drains only its uniquely labelled probe workload. + +```bash +ISVTEST_INCLUDE_UNRELEASED=1 ISVTEST_BREAKFIX_ALLOW_MUTATION=1 ISVTEST_BREAKFIX_NODE= uv run isvctl deploy run -f isvctl/configs/providers/kubernetes-node-maintenance.yaml -- -v -s -k ReturnNodeMaintenanceCheck +``` + ### With ISV Lab Service Integration Upload results to the ISV Lab Service: diff --git a/isvctl/configs/providers/kubernetes-node-maintenance.yaml b/isvctl/configs/providers/kubernetes-node-maintenance.yaml new file mode 100644 index 000000000..4a1114154 --- /dev/null +++ b/isvctl/configs/providers/kubernetes-node-maintenance.yaml @@ -0,0 +1,40 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Explicitly mutating Kubernetes BFX01-02 configuration. +# +# The active Kubernetes cluster must provide the NVIDIA Maintenance Operator +# NodeMaintenance CRD. This run creates a uniquely owned probe Deployment, +# requests maintenance for one explicit node, verifies the operator cordons the +# node and evicts only that probe, then deletes the request and verifies the +# node and workload recover. +# +# export ISVTEST_INCLUDE_UNRELEASED=1 +# export ISVTEST_BREAKFIX_ALLOW_MUTATION=1 +# export ISVTEST_BREAKFIX_NODE= +# uv run isvctl test run \ +# -f isvctl/configs/providers/kubernetes-node-maintenance.yaml \ +# --label breakfix -- -v -s -k ReturnNodeMaintenanceCheck + +import: ../suites/bare_metal.yaml + +version: "1.0" + +commands: + bare_metal: + phases: ["test"] + steps: + - name: return_node_maintenance + phase: test + command: "python shared/breakfix/return_node_maintenance.py" + args: + - "--node={{ env.ISVTEST_BREAKFIX_NODE | default('', true) }}" + timeout: 1200 + requires_available_validations: + - ReturnNodeMaintenanceCheck + +tests: + description: "Bare-metal BFX01-02 validation through the Kubernetes Maintenance Operator API" + + settings: + show_skipped_tests: true diff --git a/isvctl/configs/providers/my-isv/scripts/breakfix/return_node_maintenance.py b/isvctl/configs/providers/my-isv/scripts/breakfix/return_node_maintenance.py index cdad3fa63..bbab208a0 100644 --- a/isvctl/configs/providers/my-isv/scripts/breakfix/return_node_maintenance.py +++ b/isvctl/configs/providers/my-isv/scripts/breakfix/return_node_maintenance.py @@ -30,7 +30,8 @@ def main() -> int: "requested": True, "accepted": True, "machine_id": machine_id, - "maintenance_mode": True, + "maintenance_mode": "Maintenance", + "restored": True, }, ) diff --git a/isvctl/configs/providers/nico/config/bare_metal.yaml b/isvctl/configs/providers/nico/config/bare_metal.yaml index 91713234c..6f5db03b2 100644 --- a/isvctl/configs/providers/nico/config/bare_metal.yaml +++ b/isvctl/configs/providers/nico/config/bare_metal.yaml @@ -85,6 +85,9 @@ # - Expected machines pre-registered; machines ingested and DPUs initialized # - NICO_ORGANIZATION and NICO_SITE_ID environment variables set # - Optional NICO_INSTANCE_ID narrows inventory checks to a known instance +# - BFX01-02 is staging-only and mutating. It structured-skips unless both +# NICO_BREAKFIX_MACHINE_ID names a dedicated fixture and +# NICO_BREAKFIX_ALLOW_MUTATION=1 explicitly opts in. # - ssh-keygen on the host (used by query_key_access to mint the throwaway key) # # Usage: @@ -439,7 +442,7 @@ commands: - name: return_node_maintenance phase: test continue_on_failure: true - command: "python ../scripts/breakfix/gap_stub.py" + command: "python ../scripts/breakfix/return_node_maintenance.py" args: - "--org" - "{{org}}" @@ -447,9 +450,11 @@ commands: - "{{site_id}}" - "--api-base" - "{{nico_api_base}}" - - "--gap" - - "BFX01-02" - timeout: 600 + - "--machine-id={{breakfix_machine_id}}" + - "--allow-mutation={{breakfix_allow_mutation}}" + # The subprocess must outlive both bounded PATCH calls and restoration + # polling so its finally block cannot be killed before cleanup. + timeout: 1200 - name: return_rack_maintenance phase: test @@ -592,6 +597,11 @@ tests: site_id: "{{env.NICO_SITE_ID}}" nico_api_base: "{{env.NICO_API_BASE}}" instance_id: "{{env.NICO_INSTANCE_ID}}" + # BFX01-02 is mutating and never selects a Machine automatically. Set this + # only to a dedicated staging fixture that the test may enter and leave + # maintenance mode. + breakfix_machine_id: "{{env.NICO_BREAKFIX_MACHINE_ID}}" + breakfix_allow_mutation: "{{env.NICO_BREAKFIX_ALLOW_MUTATION}}" # The imported bare_metal suite defines these for the full instance # lifecycle, which NICo does not run yet. Blank them so the suite's # {{instance_type}} self-reference does not emit missing-variable warnings. diff --git a/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py b/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py index 55fdc447e..1bedda46c 100644 --- a/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py +++ b/isvctl/configs/providers/nico/scripts/breakfix/gap_stub.py @@ -5,7 +5,7 @@ """Emit the documented NICo gap payload for a break-fix requirement. Several break-fix requirements have no NICo tenant REST surface to exercise: -the mutating BFX01 workflows run through Maestro/repair fixtures, and the +the remaining mutating BFX01 workflows run through Maestro/repair fixtures, and the BFX02-02/BFX03-02/BFX04-01/BFX05/BFX06 signals are not exposed at all. Each of those steps emits a structured skip naming the gap rather than a hard failure, so the suite reports "not available on this platform" instead of "broken". @@ -28,10 +28,6 @@ # gap id -> (skip reason, contract fields the bound validation still expects) GAPS: dict[str, tuple[str, dict[str, Any]]] = { - "BFX01-02": ( - "Return-node-for-maintenance is a mutating NICo repair workflow requiring lab fixtures (BFX01-02 gap)", - {"operation": {"requested": False, "accepted": False}}, - ), "BFX01-03": ( "Rack-level maintenance return API is not exposed on NICo tenant REST (BFX01-03 gap)", {"operation": {"requested": False, "accepted": False}}, diff --git a/isvctl/configs/providers/nico/scripts/breakfix/return_node_maintenance.py b/isvctl/configs/providers/nico/scripts/breakfix/return_node_maintenance.py new file mode 100755 index 000000000..70ef58491 --- /dev/null +++ b/isvctl/configs/providers/nico/scripts/breakfix/return_node_maintenance.py @@ -0,0 +1,214 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Return one explicit NICo machine for maintenance and restore it (BFX01-02). + +The caller must name a dedicated fixture. The script never discovers or selects +a mutation target on its own. It verifies NICo reports the Machine in +``Maintenance`` after the request, then disables maintenance mode in ``finally`` +so failed assertions do not strand the fixture. +""" + +from __future__ import annotations + +import argparse +import sys +import time +from pathlib import Path +from typing import Any +from urllib.error import URLError +from urllib.parse import quote + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from breakfix._common import emit +from common.nico_client import NicoAuthError, forge_get, forge_patch, resolve_auth + +MUTATION_TIMEOUT_SECONDS = 300 +RESTORE_TIMEOUT_SECONDS = 120 +RESTORE_POLL_INTERVAL_SECONDS = 2.0 +MAINTENANCE_MESSAGE = "ISV BFX01-02 validation fixture; automatically restored" + + +def _operation(machine_id: str) -> dict[str, Any]: + """Build the provider-neutral operation result with safe defaults.""" + return { + "requested": False, + "accepted": False, + "machine_id": machine_id, + "maintenance_mode": "", + "restored": False, + } + + +def _api_error(exc: Exception) -> str: + """Return a concise API error without response payloads or credentials.""" + return f"{type(exc).__name__}: {exc}" + + +def _machine_status(machine: dict[str, Any]) -> str: + """Return a normalized Machine status string.""" + return str(machine.get("status") or "").strip() + + +def _has_binding(machine: dict[str, Any], key: str) -> bool: + """Return whether a Machine has a non-empty allocation identifier.""" + value = machine.get(key) + return value is not None and bool(str(value).strip()) + + +def _wait_for_status( + org: str, + machine_path: str, + token: str, + *, + base_url: str, + expected_status: str, +) -> dict[str, Any]: + """Poll NICo until the Machine returns to its exact initial status.""" + deadline = time.monotonic() + RESTORE_TIMEOUT_SECONDS + while True: + current = forge_get(org, machine_path, token, base_url=base_url) + if _machine_status(current) == expected_status or time.monotonic() >= deadline: + return current + time.sleep(RESTORE_POLL_INTERVAL_SECONDS) + + +def main() -> int: + """Request and verify maintenance mode, then restore the explicit fixture.""" + parser = argparse.ArgumentParser(description="Return an explicit NICo machine for maintenance") + parser.add_argument("--org", required=True) + parser.add_argument("--site-id", required=True) + parser.add_argument("--api-base", required=True) + parser.add_argument("--machine-id", default="", help="Dedicated staging Machine ID; no automatic selection") + parser.add_argument("--allow-mutation", default="", help="Must be exactly 1 to mutate the staging fixture") + args = parser.parse_args() + + machine_id = args.machine_id.strip() + result: dict[str, Any] = { + "success": False, + "platform": "nico", + "site_id": args.site_id, + "operation": _operation(machine_id), + } + operation = result["operation"] + + if not machine_id: + result.update( + { + "success": True, + "skipped": True, + "skip_reason": ( + "No dedicated maintenance fixture configured; set NICO_BREAKFIX_MACHINE_ID " + "to a staging Machine that may be mutated and restored" + ), + } + ) + return emit(result) + if args.allow_mutation != "1": + result.update( + { + "success": True, + "skipped": True, + "skip_reason": ( + "NICo maintenance mutation is disabled; set NICO_BREAKFIX_ALLOW_MUTATION=1 " + "only for an approved staging fixture" + ), + } + ) + return emit(result) + + machine_path = f"machine/{quote(machine_id, safe='')}" + try: + auth = resolve_auth() + initial = forge_get(args.org, machine_path, auth.token, base_url=args.api_base) + except (NicoAuthError, URLError, ValueError) as exc: + result["error"] = _api_error(exc) + return emit(result) + + if initial.get("id") != machine_id: + result["error"] = "NICo returned a different Machine than the configured maintenance fixture" + return emit(result) + if initial.get("siteId") != args.site_id: + result["error"] = "Configured maintenance fixture does not belong to the configured Site" + return emit(result) + + initial_status = _machine_status(initial) + if not initial_status: + result["error"] = "Configured maintenance fixture has no observable status" + return emit(result) + if initial_status == "Maintenance": + result["error"] = "Configured maintenance fixture is already in Maintenance; refusing to take ownership" + return emit(result) + if initial_status != "Ready": + result["error"] = "Configured maintenance fixture must be Ready before validation" + return emit(result) + if _has_binding(initial, "instanceId") or _has_binding(initial, "tenantId"): + result["error"] = "Configured maintenance fixture is allocated; refusing to mutate it" + return emit(result) + + maintenance_attempted = False + cleanup_errors: list[str] = [] + try: + operation["requested"] = True + maintenance_attempted = True + updated = forge_patch( + args.org, + machine_path, + auth.token, + base_url=args.api_base, + body={"setMaintenanceMode": True, "maintenanceMessage": MAINTENANCE_MESSAGE}, + timeout=MUTATION_TIMEOUT_SECONDS, + ) + current = forge_get(args.org, machine_path, auth.token, base_url=args.api_base) + updated_status = _machine_status(updated) + current_status = _machine_status(current) + operation["maintenance_mode"] = current_status + operation["accepted"] = updated_status == "Maintenance" and current_status == "Maintenance" + if not operation["accepted"]: + result["error"] = ( + "NICo did not confirm Maintenance state " + f"(response={updated_status or 'missing'}, current={current_status or 'missing'})" + ) + except (NicoAuthError, URLError, ValueError) as exc: + result["error"] = _api_error(exc) + finally: + if maintenance_attempted: + try: + forge_patch( + args.org, + machine_path, + auth.token, + base_url=args.api_base, + body={"setMaintenanceMode": False}, + timeout=MUTATION_TIMEOUT_SECONDS, + ) + restored_current = _wait_for_status( + args.org, + machine_path, + auth.token, + base_url=args.api_base, + expected_status=initial_status, + ) + current_status = _machine_status(restored_current) + operation["restored"] = current_status == initial_status + if not operation["restored"]: + cleanup_errors.append( + "NICo did not restore the fixture to its initial Ready state " + f"(current={current_status or 'missing'})" + ) + except (NicoAuthError, URLError, ValueError) as exc: + cleanup_errors.append(_api_error(exc)) + + if cleanup_errors: + result["cleanup_errors"] = cleanup_errors + result.setdefault("error", "Failed to restore the maintenance fixture") + + result["success"] = bool(operation["requested"] and operation["accepted"] and operation["restored"]) + if not result["success"]: + result.setdefault("error", "Node maintenance validation did not complete") + return emit(result) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/isvctl/configs/providers/shared/breakfix/return_node_maintenance.py b/isvctl/configs/providers/shared/breakfix/return_node_maintenance.py new file mode 100644 index 000000000..b315bfda6 --- /dev/null +++ b/isvctl/configs/providers/shared/breakfix/return_node_maintenance.py @@ -0,0 +1,919 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Exercise a reversible Kubernetes NodeMaintenance request for BFX01-02.""" + +from __future__ import annotations + +import argparse +import json +import os +import shlex +import subprocess +import sys +import time +import uuid +from typing import Any +from urllib.parse import quote + +DEFAULT_IMAGE = "registry.k8s.io/pause:3.10" +DEFAULT_COMMAND_TIMEOUT_SECONDS = 30.0 +DEFAULT_REQUEST_TIMEOUT_SECONDS = 15.0 +MUTATION_OPT_IN_ENV = "ISVTEST_BREAKFIX_ALLOW_MUTATION" +NODE_MAINTENANCE_RESOURCE = "nodemaintenances.maintenance.nvidia.com" +NODE_MAINTENANCE_CRD = f"{NODE_MAINTENANCE_RESOURCE}" +RUN_LABEL = "isvtest.nvidia.com/bfx01-02-run" +REQUESTOR_ID = "bfx01-02.isvtest.nvidia.com" + + +class MaintenanceTestError(RuntimeError): + """Raised when the maintenance workflow cannot prove the requirement.""" + + +class KubectlTimeoutError(MaintenanceTestError): + """Raised when a bounded kubectl process exceeds its deadline.""" + + +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 MaintenanceTestError(f"Invalid KUBECTL value: {exc}") from exc + if not command: + raise MaintenanceTestError("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 MaintenanceTestError("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 MaintenanceTestError(f"Unable to run {' '.join(kubectl)}: {exc}") from exc + if check and completed.returncode != 0: + raise MaintenanceTestError(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 MaintenanceTestError(f"kubectl returned invalid JSON for {resource}") from exc + if not isinstance(payload, dict): + raise MaintenanceTestError(f"kubectl returned a non-object for {resource}") + return payload + + +def _get_json(kubectl: list[str], *args: str, resource: str) -> dict[str, Any]: + """Get one Kubernetes object or list as JSON.""" + return _json_output(_run(kubectl, *args, "-o", "json"), resource) + + +def _condition_status(payload: dict[str, Any], condition_type: str) -> tuple[str, str, int | None]: + """Return status, reason, and observed generation for one condition.""" + conditions = payload.get("status", {}).get("conditions", []) + if not isinstance(conditions, list): + return "", "", None + for condition in conditions: + if not isinstance(condition, dict) or condition.get("type") != condition_type: + continue + observed = condition.get("observedGeneration") + return ( + str(condition.get("status") or ""), + str(condition.get("reason") or ""), + observed if isinstance(observed, int) else None, + ) + return "", "", None + + +def _node_ready_and_schedulable(node: dict[str, Any]) -> bool: + """Return whether a node is Ready and not already under maintenance.""" + conditions = node.get("status", {}).get("conditions", []) + ready = any( + isinstance(item, dict) and item.get("type") == "Ready" and item.get("status") == "True" for item in conditions + ) + return ready and node.get("spec", {}).get("unschedulable", False) is not True + + +def _require_permission( + kubectl: list[str], + verb: str, + resource: str, + *, + namespace: str | None = None, + all_namespaces: bool = False, +) -> None: + """Require one Kubernetes permission before creating test resources.""" + args = ["auth", "can-i", verb, resource] + if namespace: + args.extend(["-n", namespace]) + if all_namespaces: + args.append("--all-namespaces") + allowed = _run(kubectl, *args).stdout.strip() + if allowed != "yes": + scope = f" in {namespace}" if namespace else " cluster-wide" + raise MaintenanceTestError(f"Kubernetes RBAC does not allow {verb} on {resource}{scope}") + + +def _maintenance_requests(kubectl: list[str], node_name: str) -> list[dict[str, Any]]: + """Return NodeMaintenance objects targeting the explicit node.""" + existing = _get_json( + kubectl, + "get", + NODE_MAINTENANCE_RESOURCE, + "-A", + resource="NodeMaintenance list", + ) + items = existing.get("items") + if not isinstance(items, list): + raise MaintenanceTestError("NodeMaintenance list is missing items") + return [item for item in items if isinstance(item, dict) and item.get("spec", {}).get("nodeName") == node_name] + + +def _maintenance_owners(kubectl: list[str], node_name: str) -> list[str]: + """Return namespaced names of maintenance requests targeting a node.""" + return [ + f"{item.get('metadata', {}).get('namespace', '')}/{item.get('metadata', {}).get('name', '')}" + for item in _maintenance_requests(kubectl, node_name) + ] + + +def _require_unclaimed_node(kubectl: list[str], node_name: str) -> dict[str, Any]: + """Require a Ready, schedulable node without another maintenance request.""" + node = _get_json(kubectl, "get", "node", node_name, resource=f"node {node_name}") + if node.get("metadata", {}).get("name") != node_name: + raise MaintenanceTestError("Kubernetes returned a different node than the explicit target") + if not _node_ready_and_schedulable(node): + raise MaintenanceTestError(f"Target node {node_name!r} must be Ready and schedulable") + owners = _maintenance_owners(kubectl, node_name) + if owners: + raise MaintenanceTestError( + f"Target node {node_name!r} already has a NodeMaintenance request: {', '.join(owners)}" + ) + return node + + +def _preflight(kubectl: list[str], node_name: str, namespace: str) -> dict[str, Any]: + """Validate the API, permissions, target node, and exclusive ownership.""" + _run(kubectl, "get", "crd", NODE_MAINTENANCE_CRD) + for verb in ("create", "get", "delete"): + _require_permission(kubectl, verb, NODE_MAINTENANCE_RESOURCE, namespace=namespace) + _require_permission(kubectl, "list", NODE_MAINTENANCE_RESOURCE, all_namespaces=True) + _require_permission(kubectl, "get", "nodes") + for verb in ("create", "get", "delete"): + _require_permission(kubectl, verb, "deployments.apps", namespace=namespace) + _require_permission(kubectl, "list", "pods", namespace=namespace) + + return _require_unclaimed_node(kubectl, node_name) + + +def _node_tolerations(node: dict[str, Any]) -> list[dict[str, str]]: + """Mirror existing node taints without tolerating the maintenance cordon.""" + tolerations: list[dict[str, str]] = [] + for taint in node.get("spec", {}).get("taints", []): + if not isinstance(taint, dict): + continue + 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") or ""), + "effect": effect, + } + ) + return tolerations + + +def _deployment_manifest( + name: str, + namespace: str, + hostname: str, + run_id: str, + image: str, + tolerations: list[dict[str, str]], +) -> str: + """Build a uniquely labelled probe Deployment pinned to the target node.""" + labels = { + "app.kubernetes.io/managed-by": "isvtest", + "isvtest.nvidia.com/purpose": "bfx01-02", + RUN_LABEL: run_id, + } + return json.dumps( + { + "apiVersion": "apps/v1", + "kind": "Deployment", + "metadata": {"name": name, "namespace": namespace, "labels": labels}, + "spec": { + "replicas": 1, + "selector": {"matchLabels": {RUN_LABEL: run_id}}, + "template": { + "metadata": {"labels": labels}, + "spec": { + "nodeSelector": {"kubernetes.io/hostname": hostname}, + "tolerations": tolerations, + "containers": [{"name": "probe", "image": image}], + }, + }, + }, + }, + separators=(",", ":"), + ) + + +def _maintenance_manifest( + name: str, + namespace: str, + node_name: str, + run_id: str, + timeout_seconds: int, +) -> str: + """Build a NodeMaintenance request that drains only the owned probe.""" + return json.dumps( + { + "apiVersion": "maintenance.nvidia.com/v1alpha1", + "kind": "NodeMaintenance", + "metadata": { + "name": name, + "namespace": namespace, + "labels": { + "app.kubernetes.io/managed-by": "isvtest", + "isvtest.nvidia.com/purpose": "bfx01-02", + RUN_LABEL: run_id, + }, + }, + "spec": { + "requestorID": REQUESTOR_ID, + "nodeName": node_name, + "cordon": True, + "drainSpec": { + "force": False, + "deleteEmptyDir": False, + "podSelector": f"{RUN_LABEL}={run_id}", + "timeoutSeconds": timeout_seconds, + }, + }, + }, + separators=(",", ":"), + ) + + +def _list_probe_pods(kubectl: list[str], namespace: str, run_id: str) -> list[dict[str, Any]]: + """Return probe pods created by this run.""" + payload = _get_json( + kubectl, + "get", + "pods", + "-n", + namespace, + "-l", + f"{RUN_LABEL}={run_id}", + resource="probe pod list", + ) + items = payload.get("items") + if not isinstance(items, list): + raise MaintenanceTestError("Probe pod list is missing items") + return [item for item in items if isinstance(item, dict)] + + +def _validate_owned_resource( + payload: dict[str, Any], + *, + kind: str, + name: str, + namespace: str, + run_id: str, + node_name: str | None = None, +) -> dict[str, Any]: + """Require exact identity and ownership on a resource created by this run.""" + metadata = payload.get("metadata", {}) + if metadata.get("name") != name or metadata.get("namespace") != namespace: + raise MaintenanceTestError(f"Kubernetes returned a different {kind} than the owned resource") + if metadata.get("labels", {}).get(RUN_LABEL) != run_id: + raise MaintenanceTestError(f"Refusing to use {kind} {namespace}/{name} without this run's ownership label") + uid = metadata.get("uid") + if not isinstance(uid, str) or not uid: + raise MaintenanceTestError(f"Owned {kind} {namespace}/{name} has no Kubernetes UID") + if node_name is not None: + spec = payload.get("spec", {}) + if spec.get("nodeName") != node_name or spec.get("requestorID") != REQUESTOR_ID: + raise MaintenanceTestError("NodeMaintenance ownership fields do not match this run") + return payload + + +def _read_owned_resource( + kubectl: list[str], + kind: str, + name: str, + namespace: str, + run_id: str, + *, + node_name: str | None = None, +) -> dict[str, Any] | None: + """Read and validate an owned resource, returning None only when absent.""" + completed = _run( + kubectl, + "get", + kind, + name, + "-n", + namespace, + "-o", + "json", + check=False, + ) + if completed.returncode != 0: + detail = _command_detail(completed) + if "notfound" in detail.lower() or "not found" in detail.lower(): + return None + raise MaintenanceTestError(f"get {kind} {namespace}/{name} failed: {detail}") + payload = _json_output(completed, f"{kind} {namespace}/{name}") + return _validate_owned_resource( + payload, + kind=kind, + name=name, + namespace=namespace, + run_id=run_id, + node_name=node_name, + ) + + +def _create_owned_resource( + kubectl: list[str], + kind: str, + name: str, + namespace: str, + run_id: str, + manifest: str, + *, + node_name: str | None = None, +) -> dict[str, Any]: + """Create a unique owned resource and recover safely from a lost response.""" + try: + completed = _run( + kubectl, + "create", + "-f", + "-", + "-o", + "json", + input_text=manifest, + check=False, + ) + except KubectlTimeoutError as exc: + observed = _read_owned_resource( + kubectl, + kind, + name, + namespace, + run_id, + node_name=node_name, + ) + if observed is None: + raise exc + return observed + if completed.returncode == 0: + payload = _json_output(completed, f"created {kind} {namespace}/{name}") + return _validate_owned_resource( + payload, + kind=kind, + name=name, + namespace=namespace, + run_id=run_id, + node_name=node_name, + ) + observed = _read_owned_resource( + kubectl, + kind, + name, + namespace, + run_id, + node_name=node_name, + ) + if observed is None: + raise MaintenanceTestError(f"Could not create {kind}: {_command_detail(completed)}") + return observed + + +def _pod_ready_on_node(pod: dict[str, Any], node_name: str) -> bool: + """Return whether one probe is Ready on the target 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( + isinstance(item, dict) and item.get("type") == "Ready" and item.get("status") == "True" for item in conditions + ) + + +def _pod_unschedulable(pod: dict[str, Any]) -> bool: + """Return whether a replacement probe is blocked by node maintenance.""" + if pod.get("spec", {}).get("nodeName"): + return False + conditions = pod.get("status", {}).get("conditions", []) + return any( + isinstance(item, dict) + and item.get("type") == "PodScheduled" + and item.get("status") == "False" + and item.get("reason") == "Unschedulable" + for item in conditions + ) + + +def _wait_for_initial_probe( + kubectl: list[str], + namespace: str, + run_id: str, + node_name: str, + deadline: float, + poll_interval_seconds: float, +) -> str: + """Wait for the original probe and return its Kubernetes UID.""" + while True: + for pod in _list_probe_pods(kubectl, namespace, run_id): + if _pod_ready_on_node(pod, node_name): + uid = pod.get("metadata", {}).get("uid") + if isinstance(uid, str) and uid: + return uid + if time.monotonic() >= deadline: + raise MaintenanceTestError("Owned probe did not become Ready on the target node") + time.sleep(poll_interval_seconds) + + +def _wait_for_maintenance_ready( + kubectl: list[str], + namespace: str, + name: str, + deadline: float, + poll_interval_seconds: float, +) -> dict[str, Any]: + """Wait until the operator reports Ready or RequestorFailed.""" + while True: + payload = _get_json( + kubectl, + "get", + NODE_MAINTENANCE_RESOURCE, + name, + "-n", + namespace, + resource=f"NodeMaintenance {namespace}/{name}", + ) + generation = payload.get("metadata", {}).get("generation") + failed, failed_reason, failed_generation = _condition_status(payload, "RequestorFailed") + if failed == "True" and failed_generation == generation: + raise MaintenanceTestError(f"NodeMaintenance failed: {failed_reason or 'operator reported failure'}") + ready, ready_reason, ready_generation = _condition_status(payload, "Ready") + if ready_reason == "RequestorFailed" and ready_generation == generation: + raise MaintenanceTestError("NodeMaintenance entered RequestorFailed state") + if ready == "True" and ready_reason == "Ready" and ready_generation == generation: + return payload + if time.monotonic() >= deadline: + raise MaintenanceTestError("Timed out waiting for NodeMaintenance Ready=True") + time.sleep(poll_interval_seconds) + + +def _wait_for_replacement_blocked( + kubectl: list[str], + namespace: str, + run_id: str, + original_uid: str, + deadline: float, + poll_interval_seconds: float, +) -> tuple[bool, bool]: + """Wait for original eviction and a different unschedulable replacement.""" + evacuated = False + while True: + pods = _list_probe_pods(kubectl, namespace, run_id) + uids = {str(pod.get("metadata", {}).get("uid")) for pod in pods if pod.get("metadata", {}).get("uid")} + evacuated = original_uid not in uids + replacement_blocked = any( + str(pod.get("metadata", {}).get("uid") or "") != original_uid and _pod_unschedulable(pod) for pod in pods + ) + if evacuated and replacement_blocked: + return True, True + if time.monotonic() >= deadline: + return evacuated, replacement_blocked + time.sleep(poll_interval_seconds) + + +def _wait_for_recovery( + kubectl: list[str], + namespace: str, + run_id: str, + node_name: str, + deadline: float, + poll_interval_seconds: float, +) -> bool: + """Wait for a replacement probe to become Ready after maintenance.""" + while True: + if any(_pod_ready_on_node(pod, node_name) for pod in _list_probe_pods(kubectl, namespace, run_id)): + return True + if time.monotonic() >= deadline: + return False + time.sleep(poll_interval_seconds) + + +def _delete_resource( + kubectl: list[str], + kind: str, + name: str, + namespace: str, + *, + uid: str, +) -> None: + """Delete one resource atomically using its Kubernetes UID.""" + escaped_namespace = quote(namespace, safe="") + escaped_name = quote(name, safe="") + if kind == NODE_MAINTENANCE_RESOURCE: + uri = f"/apis/maintenance.nvidia.com/v1alpha1/namespaces/{escaped_namespace}/nodemaintenances/{escaped_name}" + elif kind == "deployment": + uri = f"/apis/apps/v1/namespaces/{escaped_namespace}/deployments/{escaped_name}" + else: + raise MaintenanceTestError(f"No atomic-delete API path is defined for {kind}") + delete_options = json.dumps( + { + "apiVersion": "v1", + "kind": "DeleteOptions", + "preconditions": {"uid": uid}, + "propagationPolicy": "Background", + }, + separators=(",", ":"), + ) + completed = _run( + kubectl, + "delete", + f"--raw={uri}", + "-f", + "-", + input_text=delete_options, + check=False, + ) + if completed.returncode != 0: + detail = _command_detail(completed) + if "notfound" in detail.lower() or "not found" in detail.lower(): + return + raise MaintenanceTestError(f"delete {kind} {namespace}/{name} failed: {detail}") + + +def _delete_owned_resource( + kubectl: list[str], + kind: str, + name: str, + namespace: str, + run_id: str, + *, + expected_uid: str, + timeout_seconds: float, + node_name: str | None = None, +) -> None: + """Delete and wait for only the exact resource UID created by this run.""" + payload = _read_owned_resource( + kubectl, + kind, + name, + namespace, + run_id, + node_name=node_name, + ) + if payload is None: + return + actual_uid = str(payload.get("metadata", {}).get("uid") or "") + if expected_uid and actual_uid != expected_uid: + raise MaintenanceTestError(f"Refusing to delete replaced {kind} {namespace}/{name}") + additional = payload.get("spec", {}).get("additionalRequestors") if node_name else None + try: + _delete_resource( + kubectl, + kind, + name, + namespace, + uid=actual_uid, + ) + deadline = time.monotonic() + timeout_seconds + while True: + observed = _read_owned_resource( + kubectl, + kind, + name, + namespace, + run_id, + node_name=node_name, + ) + if observed is None: + return + observed_uid = str(observed.get("metadata", {}).get("uid") or "") + if observed_uid != actual_uid: + raise MaintenanceTestError(f"A replacement {kind} appeared while waiting for deletion") + if time.monotonic() >= deadline: + raise MaintenanceTestError(f"Timed out waiting for {kind} {namespace}/{name} deletion") + time.sleep(1) + except MaintenanceTestError as exc: + suffix = "" + if isinstance(additional, list) and additional: + suffix = f"; {len(additional)} additional requestor(s) still hold maintenance" + raise MaintenanceTestError(f"{exc}{suffix}") from exc + + +def _wait_for_probe_absent( + kubectl: list[str], + namespace: str, + run_id: str, + deadline: float, + poll_interval_seconds: float, +) -> bool: + """Wait for cascading deletion of every pod owned by the probe run.""" + while True: + if not _list_probe_pods(kubectl, namespace, run_id): + return True + if time.monotonic() >= deadline: + return False + time.sleep(poll_interval_seconds) + + +def _wait_for_node_restored( + kubectl: list[str], + node_name: str, + deadline: float, + poll_interval_seconds: float, +) -> bool: + """Wait for the operator to return the target node to its initial state.""" + while True: + node = _get_json(kubectl, "get", "node", node_name, resource=f"node {node_name}") + if _node_ready_and_schedulable(node): + return True + if time.monotonic() >= deadline: + return False + time.sleep(poll_interval_seconds) + + +def _parser() -> argparse.ArgumentParser: + """Build the command-line parser.""" + parser = argparse.ArgumentParser(description="Request and restore Kubernetes node maintenance") + parser.add_argument("--node", default="", help="Explicit Ready node dedicated to this validation") + parser.add_argument("--namespace", default="default", help="Namespace for owned validation resources") + parser.add_argument("--image", default=DEFAULT_IMAGE, help="Container image for the owned probe") + parser.add_argument("--timeout-seconds", type=float, default=300, help="Timeout for each state transition") + parser.add_argument("--poll-interval-seconds", type=float, default=2, help="State polling interval") + return parser + + +def main() -> int: + """Run one reversible NodeMaintenance request and emit provider-neutral JSON.""" + args = _parser().parse_args() + operation: dict[str, Any] = { + "requested": False, + "accepted": False, + "maintenance_mode": "", + "workload_evacuated": False, + "replacement_blocked": False, + "workload_recovered": False, + "restored": False, + } + result: dict[str, Any] = { + "success": False, + "platform": "kubernetes", + "test_name": "return_node_maintenance", + } + kubectl: list[str] = [] + deployment_create_attempted = False + maintenance_create_attempted = False + deployment_uid = "" + maintenance_uid = "" + run_id = uuid.uuid4().hex[:16] + deployment_name = f"isvtest-bfx01-02-probe-{run_id}" + maintenance_name = f"isvtest-bfx01-02-{run_id}" + + try: + if args.timeout_seconds <= 0 or args.poll_interval_seconds <= 0: + raise MaintenanceTestError("Timeout and poll interval must be greater than zero") + if os.environ.get(MUTATION_OPT_IN_ENV) != "1": + raise MaintenanceTestError( + f"Refusing to mutate cluster state; explicitly set {MUTATION_OPT_IN_ENV}=1 for BFX01-02" + ) + node_name = args.node.strip() + if not node_name: + raise MaintenanceTestError("BFX01-02 requires an explicit --node dedicated to the validation") + operation["node_id"] = node_name + kubectl = _kubectl_command() + node = _preflight(kubectl, node_name, args.namespace) + hostname = node.get("metadata", {}).get("labels", {}).get("kubernetes.io/hostname") + if not isinstance(hostname, str) or not hostname: + raise MaintenanceTestError(f"Target node {node_name!r} is missing the kubernetes.io/hostname label") + + deployment_create_attempted = True + deployment = _create_owned_resource( + kubectl, + "deployment", + deployment_name, + args.namespace, + run_id, + _deployment_manifest( + deployment_name, + args.namespace, + hostname, + run_id, + args.image, + _node_tolerations(node), + ), + ) + deployment_uid = str(deployment["metadata"]["uid"]) + deadline = time.monotonic() + args.timeout_seconds + original_uid = _wait_for_initial_probe( + kubectl, + args.namespace, + run_id, + node_name, + deadline, + args.poll_interval_seconds, + ) + + # Close the preflight/create window immediately before asking the + # operator to mutate the node. The operator itself schedules at most + # one NodeMaintenance per node if another request races this check. + _require_unclaimed_node(kubectl, node_name) + maintenance_create_attempted = True + operation["requested"] = True + maintenance_created = _create_owned_resource( + kubectl, + NODE_MAINTENANCE_RESOURCE, + maintenance_name, + args.namespace, + run_id, + _maintenance_manifest( + maintenance_name, + args.namespace, + node_name, + run_id, + max(1, int(args.timeout_seconds)), + ), + node_name=node_name, + ) + maintenance_uid = str(maintenance_created["metadata"]["uid"]) + node_requests = _maintenance_requests(kubectl, node_name) + request_uids = {item.get("metadata", {}).get("uid") for item in node_requests} + if len(node_requests) != 1 or request_uids != {maintenance_uid}: + raise MaintenanceTestError("A concurrent NodeMaintenance request claimed the target node") + + deadline = time.monotonic() + args.timeout_seconds + maintenance = _wait_for_maintenance_ready( + kubectl, + args.namespace, + maintenance_name, + deadline, + args.poll_interval_seconds, + ) + node = _get_json(kubectl, "get", "node", node_name, resource=f"node {node_name}") + if node.get("spec", {}).get("unschedulable") is not True: + raise MaintenanceTestError("NodeMaintenance reported Ready but the node is not cordoned") + + drain = maintenance.get("status", {}).get("drain") or {} + eviction_pods = drain.get("evictionPods") + if eviction_pods != 1: + raise MaintenanceTestError("NodeMaintenance did not report exactly one owned probe for eviction") + if drain.get("drainProgress") != 100: + raise MaintenanceTestError("NodeMaintenance reported Ready without completing its drain") + if drain.get("waitForEviction") not in (None, []): + raise MaintenanceTestError("NodeMaintenance reported Ready with pending pod evictions") + + transition_deadline = time.monotonic() + args.timeout_seconds + evacuated, replacement_blocked = _wait_for_replacement_blocked( + kubectl, + args.namespace, + run_id, + original_uid, + transition_deadline, + args.poll_interval_seconds, + ) + operation["workload_evacuated"] = evacuated + operation["replacement_blocked"] = replacement_blocked + if not evacuated or not replacement_blocked: + raise MaintenanceTestError( + "NodeMaintenance did not prove owned workload evacuation and replacement blocking" + ) + operation["maintenance_mode"] = "Maintenance" + operation["accepted"] = True + except MaintenanceTestError as exc: + result["error"] = str(exc) + finally: + cleanup_errors: list[str] = [] + if kubectl and maintenance_create_attempted: + try: + _delete_owned_resource( + kubectl, + NODE_MAINTENANCE_RESOURCE, + maintenance_name, + args.namespace, + run_id, + expected_uid=maintenance_uid, + timeout_seconds=args.timeout_seconds, + node_name=str(operation.get("node_id") or ""), + ) + except MaintenanceTestError as exc: + cleanup_errors.append(f"delete NodeMaintenance: {exc}") + if kubectl and operation.get("node_id") and maintenance_create_attempted: + try: + deadline = time.monotonic() + args.timeout_seconds + node_restored = _wait_for_node_restored( + kubectl, + str(operation["node_id"]), + deadline, + args.poll_interval_seconds, + ) + if not node_restored: + cleanup_errors.append("NodeMaintenance deletion did not restore node schedulability") + elif deployment_create_attempted: + recovery_deadline = time.monotonic() + args.timeout_seconds + operation["workload_recovered"] = _wait_for_recovery( + kubectl, + args.namespace, + run_id, + str(operation["node_id"]), + recovery_deadline, + args.poll_interval_seconds, + ) + if not operation["workload_recovered"]: + cleanup_errors.append("Owned workload did not recover after maintenance") + operation["restored"] = bool(node_restored and operation["workload_recovered"]) + except MaintenanceTestError as exc: + cleanup_errors.append(f"verify restoration: {exc}") + if kubectl and deployment_create_attempted: + try: + cleanup_timeout = min(args.timeout_seconds, 120) + _delete_owned_resource( + kubectl, + "deployment", + deployment_name, + args.namespace, + run_id, + expected_uid=deployment_uid, + timeout_seconds=cleanup_timeout, + ) + cleanup_deadline = time.monotonic() + cleanup_timeout + if not _wait_for_probe_absent( + kubectl, + args.namespace, + run_id, + cleanup_deadline, + args.poll_interval_seconds, + ): + cleanup_errors.append("Probe pods remained after Deployment cleanup") + except MaintenanceTestError as exc: + cleanup_errors.append(f"delete probe Deployment: {exc}") + if cleanup_errors: + result["cleanup_errors"] = cleanup_errors + result.setdefault("error", "Node maintenance cleanup failed") + + result["success"] = bool( + operation["requested"] + and operation["accepted"] + and operation["workload_evacuated"] + and operation["replacement_blocked"] + and operation["workload_recovered"] + and operation["restored"] + and not result.get("cleanup_errors") + ) + if not result["success"]: + result.setdefault("error", "Node maintenance validation did not complete") + 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 4c71c3bba..b49189298 100644 --- a/isvctl/configs/suites/README.md +++ b/isvctl/configs/suites/README.md @@ -265,7 +265,7 @@ its plan item is not platform-scoped. | `query_repair_history` | test | `providers/nico/scripts/breakfix/query_repair_history.py` | `history_queryable`, `records[].{machine_id,entries}` -- a record needs non-empty `entries` to count (BFX02-03) | | `query_switch_firmware` | test | `providers/my-isv/scripts/breakfix/query_switch_firmware.py` | `trays[].{tray_id,firmware_version}` (BFX03-02) | | `query_bmc_kernel_logs` | test | `providers/nico/scripts/breakfix/query_bmc_kernel_logs.py` | `hosts[].{host_id,kernel_log_available}` (BFX03-03) | -| `return_node_maintenance` | test | `providers/my-isv/scripts/breakfix/return_node_maintenance.py` | `operation.{requested,accepted,machine_id,maintenance_mode}` (BFX01-02) | +| `return_node_maintenance` | test | `providers/my-isv/scripts/breakfix/return_node_maintenance.py` template; `providers/shared/breakfix/return_node_maintenance.py` Kubernetes Maintenance Operator reference | `operation.{requested,accepted,machine_id|node_id,maintenance_mode,restored}`; Kubernetes also requires `workload_evacuated`, `replacement_blocked`, and `workload_recovered` (BFX01-02) | | `return_rack_maintenance` | test | `providers/my-isv/scripts/breakfix/return_rack_maintenance.py` | `operation.{requested,accepted,rack_id}` (BFX01-03) | | `request_host_replacement` | test | `providers/my-isv/scripts/breakfix/request_host_replacement.py` | `operation.{requested,node_removed_from_pool,machine_id}` (BFX01-05) | | `query_node_health_agents` | test | `providers/my-isv/scripts/breakfix/query_node_health_agents.py` | `agents_observable`, `agents[].{node_id,agent_name,running}` (BFX04-01) | diff --git a/isvctl/src/isvctl/cli/deploy.py b/isvctl/src/isvctl/cli/deploy.py index ad5090fd3..d1f375352 100644 --- a/isvctl/src/isvctl/cli/deploy.py +++ b/isvctl/src/isvctl/cli/deploy.py @@ -86,10 +86,11 @@ 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 values the target cannot obtain on its own: a credential, the release + gate, and explicit break-fix mutation controls 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. """ forwarded: dict[str, str] = {} ngc_api_key = get_ngc_api_key() @@ -98,6 +99,10 @@ def _remote_env_assignments() -> str: include_unreleased = os.environ.get(INCLUDE_UNRELEASED_ENV, "") if include_unreleased: forwarded[INCLUDE_UNRELEASED_ENV] = include_unreleased + for name in ("ISVTEST_BREAKFIX_ALLOW_MUTATION", "ISVTEST_BREAKFIX_NODE"): + 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/providers/nico/test_nico_provider.py b/isvctl/tests/providers/nico/test_nico_provider.py index e89b3dfe9..ce78bd5ce 100644 --- a/isvctl/tests/providers/nico/test_nico_provider.py +++ b/isvctl/tests/providers/nico/test_nico_provider.py @@ -27,7 +27,7 @@ from pathlib import Path from types import ModuleType, SimpleNamespace from typing import Any -from urllib.error import HTTPError +from urllib.error import HTTPError, URLError from urllib.parse import parse_qs import pytest @@ -3921,3 +3921,308 @@ def test_query_key_access_no_provision_never_mutates( assert code == 0 assert out["skipped"] is True assert calls == [] + + +# --------------------------------------------------------------------------- +# return_node_maintenance (BFX01-02) script +# --------------------------------------------------------------------------- + + +def _load_return_node_maintenance_script() -> ModuleType: + """Load the reversible NICo maintenance script for direct unit testing.""" + return _load_nico_script("breakfix/return_node_maintenance.py", "test_return_node_maintenance") + + +def _maintenance_argv(machine_id: str = "fixture-1", *, allow_mutation: str = "1") -> list[str]: + """Build arguments targeting one explicit staging fixture.""" + return [ + "return_node_maintenance.py", + "--org", + "o", + "--site-id", + "site-1", + "--api-base", + "http://x", + f"--machine-id={machine_id}", + f"--allow-mutation={allow_mutation}", + ] + + +def test_return_node_maintenance_requires_explicit_fixture( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """An empty fixture ID skips without authenticating or calling NICo.""" + module = _load_return_node_maintenance_script() + calls: list[str] = [] + monkeypatch.setattr(module, "resolve_auth", lambda: calls.append("auth")) + + code, out = _run_script_main(module, monkeypatch, capsys, _maintenance_argv("")) + + assert code == 0 + assert out["skipped"] is True + assert "NICO_BREAKFIX_MACHINE_ID" in out["skip_reason"] + assert calls == [] + + +def test_return_node_maintenance_requires_mutation_opt_in( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A fixture ID alone remains read-only unless mutation is explicitly enabled.""" + module = _load_return_node_maintenance_script() + calls: list[str] = [] + monkeypatch.setattr(module, "resolve_auth", lambda: calls.append("auth")) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + _maintenance_argv(allow_mutation=""), + ) + + assert code == 0 + assert out["skipped"] is True + assert "NICO_BREAKFIX_ALLOW_MUTATION=1" in out["skip_reason"] + assert calls == [] + + +def test_return_node_maintenance_rejects_wrong_site_without_mutation( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """The configured Machine must belong to the configured Site before any PATCH.""" + module = _load_return_node_maintenance_script() + patches: list[dict[str, Any]] = [] + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr( + module, + "forge_get", + lambda *args, **kwargs: {"id": "fixture-1", "siteId": "site-2", "status": "Ready"}, + ) + monkeypatch.setattr(module, "forge_patch", lambda *args, **kwargs: patches.append(kwargs["body"])) + + code, out = _run_script_main(module, monkeypatch, capsys, _maintenance_argv()) + + assert code == 1 + assert out["success"] is False + assert "does not belong" in out["error"] + assert patches == [] + + +def test_return_node_maintenance_refuses_existing_maintenance_owner( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A pre-existing maintenance state is never claimed or restored by this run.""" + module = _load_return_node_maintenance_script() + patches: list[dict[str, Any]] = [] + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr( + module, + "forge_get", + lambda *args, **kwargs: {"id": "fixture-1", "siteId": "site-1", "status": "Maintenance"}, + ) + monkeypatch.setattr(module, "forge_patch", lambda *args, **kwargs: patches.append(kwargs["body"])) + + code, out = _run_script_main(module, monkeypatch, capsys, _maintenance_argv()) + + assert code == 1 + assert "already in Maintenance" in out["error"] + assert patches == [] + + +@pytest.mark.parametrize( + ("machine_fields", "error_fragment"), + [ + ({"status": "InUse"}, "must be Ready"), + ({"status": "Ready", "instanceId": "instance-1"}, "is allocated"), + ({"status": "Ready", "tenantId": "tenant-1"}, "is allocated"), + ], +) +def test_return_node_maintenance_requires_idle_ready_fixture( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + machine_fields: dict[str, Any], + error_fragment: str, +) -> None: + """Only a Ready Machine with no instance or tenant binding may be mutated.""" + module = _load_return_node_maintenance_script() + patches: list[dict[str, Any]] = [] + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr( + module, + "forge_get", + lambda *args, **kwargs: {"id": "fixture-1", "siteId": "site-1", **machine_fields}, + ) + monkeypatch.setattr(module, "forge_patch", lambda *args, **kwargs: patches.append(kwargs["body"])) + + code, out = _run_script_main(module, monkeypatch, capsys, _maintenance_argv()) + + assert code == 1 + assert error_fragment in out["error"] + assert patches == [] + + +def test_return_node_maintenance_verifies_and_restores( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Both the PATCH response and a fresh GET must show Maintenance before cleanup.""" + module = _load_return_node_maintenance_script() + gets = iter( + [ + {"id": "fixture-1", "siteId": "site-1", "status": "Ready"}, + {"id": "fixture-1", "siteId": "site-1", "status": "Maintenance"}, + {"id": "fixture-1", "siteId": "site-1", "status": "Ready"}, + ] + ) + patches: list[dict[str, Any]] = [] + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "forge_get", lambda *args, **kwargs: next(gets)) + + def patch_machine(*args: object, **kwargs: Any) -> dict[str, Any]: + """Record the request and return NICo's synchronous state.""" + body = kwargs["body"] + patches.append(body) + return {"status": "Maintenance" if body["setMaintenanceMode"] else "Initializing"} + + monkeypatch.setattr(module, "forge_patch", patch_machine) + + code, out = _run_script_main(module, monkeypatch, capsys, _maintenance_argv()) + + assert code == 0 + assert out["success"] is True + assert out["operation"] == { + "requested": True, + "accepted": True, + "machine_id": "fixture-1", + "maintenance_mode": "Maintenance", + "restored": True, + } + assert patches[0] == { + "setMaintenanceMode": True, + "maintenanceMessage": module.MAINTENANCE_MESSAGE, + } + assert patches[1] == {"setMaintenanceMode": False} + + +def test_return_node_maintenance_rejects_unconfirmed_state_but_restores( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """An accepted PATCH is insufficient when the current Machine is not in Maintenance.""" + module = _load_return_node_maintenance_script() + gets = iter( + [ + {"id": "fixture-1", "siteId": "site-1", "status": "Ready"}, + {"id": "fixture-1", "siteId": "site-1", "status": "Ready"}, + {"id": "fixture-1", "siteId": "site-1", "status": "Ready"}, + ] + ) + patches: list[dict[str, Any]] = [] + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "forge_get", lambda *args, **kwargs: next(gets)) + + def patch_machine(*args: object, **kwargs: Any) -> dict[str, Any]: + """Return Maintenance for the request and Initializing for restoration.""" + patches.append(kwargs["body"]) + return {"status": "Maintenance" if len(patches) == 1 else "Initializing"} + + monkeypatch.setattr(module, "forge_patch", patch_machine) + + code, out = _run_script_main(module, monkeypatch, capsys, _maintenance_argv()) + + assert code == 1 + assert out["operation"]["accepted"] is False + assert out["operation"]["restored"] is True + assert patches[-1] == {"setMaintenanceMode": False} + + +def test_return_node_maintenance_restores_after_request_error( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Once the enabling request starts, cleanup runs even when that request errors.""" + module = _load_return_node_maintenance_script() + gets = iter( + [ + {"id": "fixture-1", "siteId": "site-1", "status": "Ready"}, + {"id": "fixture-1", "siteId": "site-1", "status": "Ready"}, + ] + ) + patches: list[dict[str, Any]] = [] + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "forge_get", lambda *args, **kwargs: next(gets)) + + def patch_machine(*args: object, **kwargs: Any) -> dict[str, Any]: + """Fail the enabling request and accept the restoration request.""" + body = kwargs["body"] + patches.append(body) + if body["setMaintenanceMode"]: + raise URLError("maintenance request failed") + return {"status": "Initializing"} + + monkeypatch.setattr(module, "forge_patch", patch_machine) + + code, out = _run_script_main(module, monkeypatch, capsys, _maintenance_argv()) + + assert code == 1 + assert "maintenance request failed" in out["error"] + assert out["operation"]["restored"] is True + assert patches[-1] == {"setMaintenanceMode": False} + + +def test_return_node_maintenance_cleanup_failure_fails_result( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A successful maintenance observation cannot pass when restoration fails.""" + module = _load_return_node_maintenance_script() + gets = iter( + [ + {"id": "fixture-1", "siteId": "site-1", "status": "Ready"}, + {"id": "fixture-1", "siteId": "site-1", "status": "Maintenance"}, + ] + ) + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "forge_get", lambda *args, **kwargs: next(gets)) + calls = 0 + + def patch_machine(*args: object, **kwargs: Any) -> dict[str, Any]: + """Accept maintenance, then fail the restoration request.""" + nonlocal calls + calls += 1 + if calls == 2: + raise URLError("restore failed") + return {"status": "Maintenance"} + + monkeypatch.setattr(module, "forge_patch", patch_machine) + + code, out = _run_script_main(module, monkeypatch, capsys, _maintenance_argv()) + + assert code == 1 + assert out["operation"]["accepted"] is True + assert out["operation"]["restored"] is False + assert out["cleanup_errors"] == ["URLError: "] + + +def test_return_node_maintenance_requires_exact_ready_restoration( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Leaving Maintenance is insufficient when the fixture has not returned to Ready.""" + module = _load_return_node_maintenance_script() + gets = iter( + [ + {"id": "fixture-1", "siteId": "site-1", "status": "Ready"}, + {"id": "fixture-1", "siteId": "site-1", "status": "Maintenance"}, + {"id": "fixture-1", "siteId": "site-1", "status": "Error"}, + ] + ) + monkeypatch.setattr(module, "RESTORE_TIMEOUT_SECONDS", 0) + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "forge_get", lambda *args, **kwargs: next(gets)) + monkeypatch.setattr( + module, + "forge_patch", + lambda *args, **kwargs: {"status": "Maintenance" if kwargs["body"]["setMaintenanceMode"] else "Initializing"}, + ) + + code, out = _run_script_main(module, monkeypatch, capsys, _maintenance_argv()) + + assert code == 1 + assert out["operation"]["accepted"] is True + assert out["operation"]["restored"] is False + assert out["cleanup_errors"] == ["NICo did not restore the fixture to its initial Ready state (current=Error)"] diff --git a/isvctl/tests/test_deploy_passthrough.py b/isvctl/tests/test_deploy_passthrough.py index 45d69f7b4..86788366f 100644 --- a/isvctl/tests/test_deploy_passthrough.py +++ b/isvctl/tests/test_deploy_passthrough.py @@ -44,7 +44,26 @@ def test_ngc_key_alias_is_forwarded_under_the_canonical_name(monkeypatch: pytest 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): + for name in ( + "NGC_API_KEY", + "NGC_NIM_API_KEY", + INCLUDE_UNRELEASED_ENV, + "ISVTEST_BREAKFIX_ALLOW_MUTATION", + "ISVTEST_BREAKFIX_NODE", + ): monkeypatch.delenv(name, raising=False) assert _remote_env_assignments() == "" + + +def test_breakfix_mutation_controls_are_forwarded_and_quoted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Remote focused runs need the same explicit consent and node target.""" + monkeypatch.delenv("NGC_API_KEY", raising=False) + monkeypatch.delenv("NGC_NIM_API_KEY", raising=False) + monkeypatch.delenv(INCLUDE_UNRELEASED_ENV, raising=False) + 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'" diff --git a/isvctl/tests/test_shared_node_maintenance.py b/isvctl/tests/test_shared_node_maintenance.py new file mode 100644 index 000000000..c52091f25 --- /dev/null +++ b/isvctl/tests/test_shared_node_maintenance.py @@ -0,0 +1,544 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the shared BFX01-02 Kubernetes maintenance 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] +SCRIPT = ISVCTL_ROOT / "configs" / "providers" / "shared" / "breakfix" / "return_node_maintenance.py" +CONFIG = ISVCTL_ROOT / "configs" / "providers" / "kubernetes-node-maintenance.yaml" +MINIKUBE_CONFIG = ISVCTL_ROOT / "configs" / "providers" / "minikube.yaml" + + +def _load_script() -> ModuleType: + """Load the script as a module for direct testing.""" + spec = importlib.util.spec_from_file_location("test_shared_return_node_maintenance_script", 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 call.""" + return subprocess.CompletedProcess( + ["kubectl", *args], + 1 if error else 0, + stdout=json.dumps(payload) if payload is not None else "", + stderr=error, + ) + + +def _node(*, unschedulable: bool = False) -> dict[str, Any]: + """Return one Ready test node.""" + return { + "metadata": { + "name": "worker-1", + "labels": {"kubernetes.io/hostname": "worker-1-host"}, + }, + "spec": { + "unschedulable": unschedulable, + "taints": [{"key": "nvidia.com/gpu", "value": "present", "effect": "NoSchedule"}], + }, + "status": {"conditions": [{"type": "Ready", "status": "True"}]}, + } + + +def test_explicit_config_wires_only_the_maintenance_reference() -> None: + """Keep the mutating step behind its explicit provider config.""" + config = yaml.safe_load(CONFIG.read_text()) + steps = config["commands"]["bare_metal"]["steps"] + + assert steps == [ + { + "name": "return_node_maintenance", + "phase": "test", + "command": "python shared/breakfix/return_node_maintenance.py", + "args": ["--node={{ env.ISVTEST_BREAKFIX_NODE | default('', true) }}"], + "timeout": 1200, + "requires_available_validations": ["ReturnNodeMaintenanceCheck"], + } + ] + + +def test_empty_node_renders_as_one_safe_argument(monkeypatch: pytest.MonkeyPatch) -> None: + """An unset target must not become a dangling command-line flag.""" + monkeypatch.delenv("ISVTEST_BREAKFIX_NODE", raising=False) + config = RunConfig.model_validate(merge_yaml_files([CONFIG])) + step = next(item for item in config.commands["bare_metal"].steps if item.name == "return_node_maintenance") + + assert StepExecutor()._render_args(step.args, Context(config)) == ["--node="] + + +def test_normal_minikube_config_never_runs_maintenance() -> None: + """Ordinary Kubernetes validation must not request maintenance.""" + config = yaml.safe_load(MINIKUBE_CONFIG.read_text()) + steps = config["commands"]["kubernetes"]["steps"] + + assert all(step["name"] != "return_node_maintenance" for step in steps) + + +def test_missing_mutation_opt_in_fails_before_kubectl( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """The focused config alone is not mutation authorization.""" + module = _load_script() + monkeypatch.delenv(module.MUTATION_OPT_IN_ENV, raising=False) + monkeypatch.setattr(sys, "argv", ["return_node_maintenance.py", "--node=worker-1"]) + monkeypatch.setattr( + module, + "_kubectl_command", + lambda: pytest.fail("kubectl must not run without opt-in"), + ) + + assert module.main() == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is False + assert payload["operation"]["requested"] is False + assert "ISVTEST_BREAKFIX_ALLOW_MUTATION=1" in payload["error"] + + +def test_explicit_node_is_required_before_kubectl( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Never choose a maintenance target on the caller's behalf.""" + module = _load_script() + monkeypatch.setenv(module.MUTATION_OPT_IN_ENV, "1") + monkeypatch.setattr(sys, "argv", ["return_node_maintenance.py"]) + monkeypatch.setattr( + module, + "_kubectl_command", + lambda: pytest.fail("kubectl must not run without a target"), + ) + + assert module.main() == 1 + payload = json.loads(capsys.readouterr().out) + assert "requires an explicit --node" in payload["error"] + + +def test_manifests_limit_drain_to_the_owned_probe() -> None: + """The operator must never drain pre-existing workloads.""" + module = _load_script() + maintenance = json.loads(module._maintenance_manifest("maintenance-1", "default", "worker-1", "run-1", 30)) + deployment = json.loads( + module._deployment_manifest( + "probe-1", + "default", + "worker-1-host", + "run-1", + module.DEFAULT_IMAGE, + [{"key": "nvidia.com/gpu", "operator": "Equal", "value": "present", "effect": "NoSchedule"}], + ) + ) + + assert maintenance["spec"] == { + "requestorID": module.REQUESTOR_ID, + "nodeName": "worker-1", + "cordon": True, + "drainSpec": { + "force": False, + "deleteEmptyDir": False, + "podSelector": f"{module.RUN_LABEL}=run-1", + "timeoutSeconds": 30, + }, + } + pod_spec = deployment["spec"]["template"]["spec"] + assert pod_spec["nodeSelector"] == {"kubernetes.io/hostname": "worker-1-host"} + assert pod_spec["tolerations"][0]["key"] == "nvidia.com/gpu" + + +def test_create_owned_resource_captures_identity(monkeypatch: pytest.MonkeyPatch) -> None: + """Creation must return the API-assigned UID and exact ownership label.""" + module = _load_script() + payload = { + "metadata": { + "name": "probe-1", + "namespace": "default", + "uid": "uid-1", + "labels": {module.RUN_LABEL: "run-1"}, + } + } + monkeypatch.setattr( + module, + "_run", + lambda *args, **kwargs: _completed(payload=payload), + ) + + created = module._create_owned_resource( + ["kubectl"], + "deployment", + "probe-1", + "default", + "run-1", + "{}", + ) + assert created["metadata"]["uid"] == "uid-1" + + +def test_delete_owned_resource_rejects_replaced_uid(monkeypatch: pytest.MonkeyPatch) -> None: + """Cleanup must not delete a same-name object that replaced this run's UID.""" + module = _load_script() + payload = { + "metadata": { + "name": "probe-1", + "namespace": "default", + "uid": "replacement-uid", + "labels": {module.RUN_LABEL: "run-1"}, + } + } + monkeypatch.setattr(module, "_read_owned_resource", lambda *args, **kwargs: payload) + monkeypatch.setattr( + module, + "_delete_resource", + lambda *args, **kwargs: pytest.fail("replacement must not be deleted"), + ) + + with pytest.raises(module.MaintenanceTestError, match="replaced deployment"): + module._delete_owned_resource( + ["kubectl"], + "deployment", + "probe-1", + "default", + "run-1", + expected_uid="original-uid", + timeout_seconds=30, + ) + + +@pytest.mark.parametrize( + ("kind", "expected_uri"), + [ + ("deployment", "/apis/apps/v1/namespaces/test%20ns/deployments/probe%2F1"), + ( + "nodemaintenances.maintenance.nvidia.com", + "/apis/maintenance.nvidia.com/v1alpha1/namespaces/test%20ns/nodemaintenances/probe%2F1", + ), + ], +) +def test_delete_resource_uses_server_side_uid_precondition( + monkeypatch: pytest.MonkeyPatch, + kind: str, + expected_uri: str, +) -> None: + """The API server must reject deletion if a same-name object replaced ours.""" + module = _load_script() + observed: dict[str, Any] = {} + + def fake_run(kubectl: list[str], *args: str, **kwargs: Any) -> subprocess.CompletedProcess[str]: + observed["args"] = args + observed["body"] = json.loads(kwargs["input_text"]) + return _completed() + + monkeypatch.setattr(module, "_run", fake_run) + + module._delete_resource( + ["kubectl"], + kind, + "probe/1", + "test ns", + uid="owned-uid", + ) + + assert observed["args"] == ("delete", f"--raw={expected_uri}", "-f", "-") + assert observed["body"]["preconditions"] == {"uid": "owned-uid"} + + +def test_delete_owned_resource_waits_for_exact_uid_to_disappear(monkeypatch: pytest.MonkeyPatch) -> None: + """Cleanup should poll the exact owned object after its atomic delete.""" + module = _load_script() + payload = { + "metadata": { + "name": "probe-1", + "namespace": "default", + "uid": "owned-uid", + "labels": {module.RUN_LABEL: "run-1"}, + } + } + reads = iter([payload, payload, None]) + deleted: dict[str, Any] = {} + monkeypatch.setattr(module, "_read_owned_resource", lambda *args, **kwargs: next(reads)) + monkeypatch.setattr( + module, + "_delete_resource", + lambda *args, **kwargs: deleted.update(kwargs), + ) + monkeypatch.setattr(module.time, "sleep", lambda *_args: None) + + module._delete_owned_resource( + ["kubectl"], + "deployment", + "probe-1", + "default", + "run-1", + expected_uid="owned-uid", + timeout_seconds=30, + ) + + assert deleted["uid"] == "owned-uid" + + +def test_preflight_rejects_an_existing_request( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Do not compete with another requestor for the same node.""" + module = _load_script() + + def fake_run( + kubectl: list[str], + *args: str, + **kwargs: Any, + ) -> subprocess.CompletedProcess[str]: + if args[:2] == ("auth", "can-i"): + return subprocess.CompletedProcess(["kubectl", *args], 0, stdout="yes\n", stderr="") + return _completed(args) + + monkeypatch.setattr(module, "_run", fake_run) + responses = iter( + [ + _node(), + { + "items": [ + { + "metadata": {"name": "other", "namespace": "ops"}, + "spec": {"nodeName": "worker-1"}, + } + ] + }, + ] + ) + monkeypatch.setattr(module, "_get_json", lambda *args, **kwargs: next(responses)) + + with pytest.raises(module.MaintenanceTestError, match="already has a NodeMaintenance"): + module._preflight(["kubectl"], "worker-1", "default") + + +def test_ready_condition_must_match_current_generation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Ignore stale Ready evidence from an older object generation.""" + module = _load_script() + payload = { + "metadata": {"generation": 2}, + "status": { + "conditions": [ + { + "type": "Ready", + "status": "True", + "reason": "Ready", + "observedGeneration": 1, + } + ] + }, + } + calls = 0 + + def fake_get(*args: Any, **kwargs: Any) -> dict[str, Any]: + nonlocal calls + calls += 1 + if calls == 2: + payload["status"]["conditions"][0]["observedGeneration"] = 2 + return payload + + monkeypatch.setattr(module, "_get_json", fake_get) + monkeypatch.setattr(module.time, "sleep", lambda _: None) + + assert ( + module._wait_for_maintenance_ready( + ["kubectl"], + "default", + "maintenance-1", + module.time.monotonic() + 1, + 0.01, + ) + is payload + ) + assert calls == 2 + + +def test_requestor_failed_condition_aborts_maintenance( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Treat the operator's terminal RequestorFailed condition as failure.""" + module = _load_script() + payload = { + "metadata": {"generation": 3}, + "status": { + "conditions": [ + { + "type": "RequestorFailed", + "status": "True", + "reason": "FailedMaintenance", + "observedGeneration": 3, + } + ] + }, + } + monkeypatch.setattr(module, "_get_json", lambda *args, **kwargs: payload) + + with pytest.raises(module.MaintenanceTestError, match="FailedMaintenance"): + module._wait_for_maintenance_ready( + ["kubectl"], + "default", + "maintenance-1", + module.time.monotonic() + 1, + 0.01, + ) + + +def test_ambiguous_deployment_create_still_runs_cleanup( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """A timed-out create must not leak the uniquely named probe.""" + module = _load_script() + deleted: list[str] = [] + monkeypatch.setenv(module.MUTATION_OPT_IN_ENV, "1") + monkeypatch.setattr(sys, "argv", ["return_node_maintenance.py", "--node=worker-1"]) + monkeypatch.setattr(module, "_kubectl_command", lambda: ["kubectl"]) + monkeypatch.setattr(module, "_preflight", lambda *args: _node()) + + def timeout_create(*_args: Any, **_kwargs: Any) -> dict[str, Any]: + raise module.KubectlTimeoutError("create timed out") + + monkeypatch.setattr(module, "_create_owned_resource", timeout_create) + monkeypatch.setattr( + module, + "_delete_owned_resource", + lambda kubectl, kind, name, namespace, run_id, **kwargs: deleted.append(kind), + ) + monkeypatch.setattr(module, "_wait_for_probe_absent", lambda *args: True) + + assert module.main() == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is False + assert payload["error"] == "create timed out" + assert deleted == ["deployment"] + + +def test_successful_workflow_reports_behavior_and_restoration( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """A PASS requires operator readiness, evacuation, blocking, and recovery.""" + module = _load_script() + deleted: list[tuple[str, str]] = [] + monkeypatch.setenv(module.MUTATION_OPT_IN_ENV, "1") + monkeypatch.setattr(sys, "argv", ["return_node_maintenance.py", "--node=worker-1"]) + monkeypatch.setattr(module, "_kubectl_command", lambda: ["kubectl"]) + monkeypatch.setattr(module, "_preflight", lambda *args: _node()) + monkeypatch.setattr(module, "_require_unclaimed_node", lambda *args: _node()) + monkeypatch.setattr( + module, + "_create_owned_resource", + lambda kubectl, kind, *args, **kwargs: { + "metadata": {"uid": "maintenance-uid" if kind == module.NODE_MAINTENANCE_RESOURCE else "deployment-uid"} + }, + ) + monkeypatch.setattr( + module, + "_maintenance_requests", + lambda *args: [{"metadata": {"uid": "maintenance-uid"}}], + ) + monkeypatch.setattr(module, "_wait_for_initial_probe", lambda *args: "old-uid") + monkeypatch.setattr( + module, + "_wait_for_maintenance_ready", + lambda *args: {"status": {"drain": {"evictionPods": 1, "drainProgress": 100}}}, + ) + monkeypatch.setattr(module, "_get_json", lambda *args, **kwargs: _node(unschedulable=True)) + monkeypatch.setattr(module, "_wait_for_replacement_blocked", lambda *args: (True, True)) + monkeypatch.setattr(module, "_wait_for_node_restored", lambda *args: True) + monkeypatch.setattr(module, "_wait_for_recovery", lambda *args: True) + monkeypatch.setattr( + module, + "_delete_owned_resource", + lambda kubectl, kind, name, namespace, run_id, **kwargs: deleted.append((kind, name)), + ) + monkeypatch.setattr(module, "_wait_for_probe_absent", lambda *args: True) + + assert module.main() == 0 + payload = json.loads(capsys.readouterr().out) + + assert payload["success"] is True + assert payload["operation"] == { + "requested": True, + "accepted": True, + "maintenance_mode": "Maintenance", + "workload_evacuated": True, + "replacement_blocked": True, + "workload_recovered": True, + "restored": True, + "node_id": "worker-1", + } + assert len(deleted) == 2 + assert deleted[0][0] == module.NODE_MAINTENANCE_RESOURCE + assert deleted[1][0] == "deployment" + + +def test_cleanup_failure_forces_failed_result( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Ready maintenance evidence cannot pass if restoration is unconfirmed.""" + module = _load_script() + monkeypatch.setenv(module.MUTATION_OPT_IN_ENV, "1") + monkeypatch.setattr(sys, "argv", ["return_node_maintenance.py", "--node=worker-1"]) + monkeypatch.setattr(module, "_kubectl_command", lambda: ["kubectl"]) + monkeypatch.setattr(module, "_preflight", lambda *args: _node()) + monkeypatch.setattr(module, "_require_unclaimed_node", lambda *args: _node()) + monkeypatch.setattr( + module, + "_create_owned_resource", + lambda kubectl, kind, *args, **kwargs: { + "metadata": {"uid": "maintenance-uid" if kind == module.NODE_MAINTENANCE_RESOURCE else "deployment-uid"} + }, + ) + monkeypatch.setattr( + module, + "_maintenance_requests", + lambda *args: [{"metadata": {"uid": "maintenance-uid"}}], + ) + monkeypatch.setattr(module, "_wait_for_initial_probe", lambda *args: "old-uid") + monkeypatch.setattr( + module, + "_wait_for_maintenance_ready", + lambda *args: {"status": {"drain": {"evictionPods": 1, "drainProgress": 100}}}, + ) + monkeypatch.setattr(module, "_get_json", lambda *args, **kwargs: _node(unschedulable=True)) + monkeypatch.setattr(module, "_wait_for_replacement_blocked", lambda *args: (True, True)) + monkeypatch.setattr(module, "_wait_for_node_restored", lambda *args: False) + monkeypatch.setattr(module, "_delete_owned_resource", lambda *args, **kwargs: None) + monkeypatch.setattr(module, "_wait_for_probe_absent", lambda *args: True) + + assert module.main() == 1 + payload = json.loads(capsys.readouterr().out) + + assert payload["operation"]["accepted"] is True + assert payload["operation"]["restored"] is False + assert payload["success"] is False + assert "restore node schedulability" in payload["cleanup_errors"][0] diff --git a/isvtest/src/isvtest/validations/breakfix.py b/isvtest/src/isvtest/validations/breakfix.py index 99612620e..eb228cec6 100644 --- a/isvtest/src/isvtest/validations/breakfix.py +++ b/isvtest/src/isvtest/validations/breakfix.py @@ -278,7 +278,11 @@ class ReturnNodeMaintenanceCheck(_OperationCheck): """Validate returning an individual node for maintenance (BFX01-02). Step output: - success, operation: {requested, accepted, machine_id, maintenance_mode} + success, platform, operation: + {requested, accepted, machine_id|node_id, maintenance_mode, restored} + + Kubernetes implementations additionally report workload_evacuated, + replacement_blocked, and workload_recovered. """ description: ClassVar[str] = "Return an individual node to the provider for maintenance via the API" @@ -292,6 +296,43 @@ def _pass_message(self, label: str, operation: dict[str, Any]) -> str: """Report the maintenance mode the provider placed the node into.""" return f"{super()._pass_message(label, operation)} (maintenance_mode={operation.get('maintenance_mode')})" + def run(self) -> None: + """Require evidence of the request, observed maintenance state, and restoration.""" + step_output = _step_output(self) + if step_output is None: + return + operation = step_output.get("operation") or {} + if not isinstance(operation, dict): + self.set_failed("Node maintenance operation evidence must be an object") + return + if operation.get("requested") is not True: + self.set_failed("Node maintenance return was not requested") + return + if operation.get("accepted") is not True: + self.set_failed(operation.get("message") or self.failure_message) + return + label = _record_label(operation, *self.label_keys) + if label == "unknown": + self.set_failed("Node maintenance evidence is missing a machine or node identifier") + return + if operation.get("maintenance_mode") != "Maintenance": + self.set_failed("Provider did not report the node in Maintenance") + return + if operation.get("restored") is not True: + self.set_failed("Provider did not restore the node after maintenance validation") + return + if step_output.get("platform") == "kubernetes": + evidence = { + "workload_evacuated": "Owned workload was not evacuated", + "replacement_blocked": "Replacement workload was not blocked during maintenance", + "workload_recovered": "Owned workload did not recover after maintenance", + } + for field, message in evidence.items(): + if operation.get(field) is not True: + self.set_failed(message) + return + self.set_passed(self._pass_message(label, operation)) + class ReturnRackMaintenanceCheck(_OperationCheck): """Validate returning a rack for maintenance (BFX01-03). diff --git a/isvtest/tests/test_breakfix.py b/isvtest/tests/test_breakfix.py index 485c8cbcb..898480bf6 100644 --- a/isvtest/tests/test_breakfix.py +++ b/isvtest/tests/test_breakfix.py @@ -138,12 +138,84 @@ def test_host_replacement_uses_its_own_flag(self) -> None: step_output = {"success": True, "operation": {"completed": True, "node_removed_from_pool": False}} assert not _run(HostReplacementCheck, step_output).passed - def test_node_maintenance_reports_mode(self) -> None: - """BFX01-02 appends the maintenance mode the provider placed the node into.""" - step_output = {"success": True, "operation": {"accepted": True, "machine_id": "m-1", "maintenance_mode": "hw"}} + def test_node_maintenance_requires_complete_evidence(self) -> None: + """BFX01-02 passes only after the requested state was observed and restored.""" + step_output = { + "success": True, + "operation": { + "requested": True, + "accepted": True, + "machine_id": "m-1", + "maintenance_mode": "Maintenance", + "restored": True, + }, + } check = _run(ReturnNodeMaintenanceCheck, step_output) assert check.passed - assert "maintenance_mode=hw" in check.message + assert "maintenance_mode=Maintenance" in check.message + + def test_kubernetes_node_maintenance_requires_evacuation_and_recovery(self) -> None: + """Kubernetes must prove behavior beyond BFX01-04 cordoning.""" + step_output = { + "success": True, + "platform": "kubernetes", + "operation": { + "requested": True, + "accepted": True, + "node_id": "worker-1", + "maintenance_mode": "Maintenance", + "workload_evacuated": True, + "replacement_blocked": True, + "workload_recovered": True, + "restored": True, + }, + } + check = _run(ReturnNodeMaintenanceCheck, step_output) + assert check.passed + assert "worker-1" in check.message + + @pytest.mark.parametrize( + "field", + ["workload_evacuated", "replacement_blocked", "workload_recovered"], + ) + def test_kubernetes_node_maintenance_rejects_missing_behavior( + self, + field: str, + ) -> None: + """Every Kubernetes maintenance behavior flag is required.""" + operation = { + "requested": True, + "accepted": True, + "node_id": "worker-1", + "maintenance_mode": "Maintenance", + "workload_evacuated": True, + "replacement_blocked": True, + "workload_recovered": True, + "restored": True, + } + operation[field] = False + assert not _run( + ReturnNodeMaintenanceCheck, + {"success": True, "platform": "kubernetes", "operation": operation}, + ).passed + + @pytest.mark.parametrize( + "operation", + [ + {"accepted": True}, + { + "requested": True, + "accepted": True, + "maintenance_mode": "Maintenance", + "restored": True, + }, + {"requested": True, "accepted": True, "maintenance_mode": "Ready", "restored": True}, + {"requested": True, "accepted": True, "maintenance_mode": "Maintenance", "restored": False}, + ], + ) + def test_node_maintenance_rejects_incomplete_evidence(self, operation: dict[str, Any]) -> None: + """A provider's accepted flag alone cannot satisfy BFX01-02.""" + assert not _run(ReturnNodeMaintenanceCheck, {"success": True, "operation": operation}).passed class TestNodeHealthAgentCheck: