From 5cab194ef7eead614c51a91ae45192b6f4d20fe2 Mon Sep 17 00:00:00 2001 From: aviruthen <91846056+aviruthen@users.noreply.github.com> Date: Tue, 13 Oct 2026 16:20:28 -0700 Subject: [PATCH 1/3] fix: Stabilize integration cluster tests --- .../hyperpod/cli/clients/kubernetes_client.py | 36 ++++++++++++++----- .../hyperpod/cli/commands/cluster.py | 7 ++-- .../hyperpod/cli/service/list_pods.py | 3 +- .../init/test_recipe_job_creation.py | 28 ++++++++++++--- .../clients/test_kubernetes_client.py | 14 ++++---- 5 files changed, 65 insertions(+), 23 deletions(-) diff --git a/src/sagemaker/hyperpod/cli/clients/kubernetes_client.py b/src/sagemaker/hyperpod/cli/clients/kubernetes_client.py index 96c92bb7..8df5002d 100644 --- a/src/sagemaker/hyperpod/cli/clients/kubernetes_client.py +++ b/src/sagemaker/hyperpod/cli/clients/kubernetes_client.py @@ -67,9 +67,18 @@ class KubernetesClient: _kube_client = None def __new__(cls, config_file: Optional[str] = None) -> "KubernetesClient": + if config_file is not None: + instance = super(KubernetesClient, cls).__new__(cls) + instance._config_file = config_file + instance._kube_client = config.new_client_from_config( + config_file=config_file + ) + return instance + if cls._instance is None: cls._instance = super(KubernetesClient, cls).__new__(cls) - config.load_kube_config(config_file=config_file or KUBE_CONFIG_PATH) + cls._instance._config_file = KUBE_CONFIG_PATH + config.load_kube_config(config_file=KUBE_CONFIG_PATH) cls._instance._kube_client = client.ApiClient() return cls._instance @@ -110,8 +119,10 @@ def set_context( with open(KUBE_CONFIG_PATH, "w") as file: yaml.safe_dump(kubeconfig, file) - # Load the updated kubeconfig + # Load the updated kubeconfig and refresh this client's configuration. config.load_kube_config(config_file=KUBE_CONFIG_PATH) + self._config_file = KUBE_CONFIG_PATH + self._kube_client = client.ApiClient() logger.debug(f"Current context set to '{context_name}'") def get_core_v1_api(self) -> client.CoreV1Api: @@ -182,13 +193,16 @@ def list_node_with_temp_config( file (str): The path to the kubeconfig file selector_str (str): Selector to filter """ - config.load_kube_config(config_file=file) - v1Client = client.CoreV1Api() + api_client = self._kube_client + if getattr(self, "_config_file", None) != file: + api_client = config.new_client_from_config(config_file=file) + + v1_client = client.CoreV1Api(api_client) _continue = None nodes = [] while True: - response = v1Client.list_node( + response = v1_client.list_node( label_selector=selector_str, limit=200, # Set a reasonable limit _continue=_continue, @@ -236,7 +250,9 @@ def get_sagemaker_managed_namespace(self, namespace: Optional[str]): if namespace is None: return None try: - response = client.CoreV1Api().read_namespace(name=namespace) + response = client.CoreV1Api(self._kube_client).read_namespace( + name=namespace + ) labels = response.metadata.labels if labels and SAGEMAKER_MANAGED_QUEUE_LABEL in labels and labels[SAGEMAKER_MANAGED_QUEUE_LABEL] == "true": return response @@ -254,12 +270,12 @@ def list_pods_with_labels(self, namespace: str, label_selector: str): ) def list_pods_in_all_namespaces_with_labels(self, label_selector: str): - v1Client = client.CoreV1Api() + v1_client = client.CoreV1Api(self._kube_client) pods = [] _continue = None while True: - response = v1Client.list_pod_for_all_namespaces( + response = v1_client.list_pod_for_all_namespaces( label_selector=label_selector, limit=200, # Set a reasonable limit _continue=_continue, @@ -367,7 +383,9 @@ def list_workload_priority_classes(self): ) def get_cluster_queue(self, cluster_queue_name: str): - return client.CustomObjectsApi().get_cluster_custom_object( + return client.CustomObjectsApi( + self._kube_client + ).get_cluster_custom_object( group=KUEUE_CUSTOM_OBJECT_GROUP, version=KUEUE_CUSTOM_OBJECT_VERSION, plural=CLUSTER_QUEUE_PRIORITY_CLASS_CUSTOM_OBJECT_PLURAL, diff --git a/src/sagemaker/hyperpod/cli/commands/cluster.py b/src/sagemaker/hyperpod/cli/commands/cluster.py index 65323e2a..2aebdd71 100644 --- a/src/sagemaker/hyperpod/cli/commands/cluster.py +++ b/src/sagemaker/hyperpod/cli/commands/cluster.py @@ -307,7 +307,7 @@ def rate_limited_operation( nodes = k8s_client.list_node_with_temp_config( temp_config_file, SAGEMAKER_HYPERPOD_NAME_LABEL ) - nodes_info = _aggregate_nodes_info(nodes) + nodes_info = _aggregate_nodes_info(nodes, k8s_client) ns_nominal_quota = {} ns_quota_usage = {} @@ -466,10 +466,13 @@ def _restructure_output(summary_list, namespaces): def _aggregate_nodes_info( nodes: List[client.V1Node], + k8s_client: Optional[KubernetesClient] = None, ) -> Dict[str, Dict[str, Any]]: list_pods_service = ListPods() nodes_resource_allocated_dict = ( - list_pods_service.list_pods_and_get_requested_resources_group_by_node_name() + list_pods_service.list_pods_and_get_requested_resources_group_by_node_name( + k8s_client + ) ) nodes_summary: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int)) diff --git a/src/sagemaker/hyperpod/cli/service/list_pods.py b/src/sagemaker/hyperpod/cli/service/list_pods.py index a34f03da..cc2bf554 100644 --- a/src/sagemaker/hyperpod/cli/service/list_pods.py +++ b/src/sagemaker/hyperpod/cli/service/list_pods.py @@ -66,13 +66,14 @@ def list_pods_for_training_job( def list_pods_and_get_requested_resources_group_by_node_name( self, + k8s_client: Optional[KubernetesClient] = None, ): """ List pods for all namespaces and initialized by kubeflow. Group by the node_name of the pod and the value is the accelerator devices resources requested """ - k8s_client = KubernetesClient() + k8s_client = k8s_client or KubernetesClient() label_filter = f"training.kubeflow.org/job-name" pods = k8s_client.list_pods_in_all_namespaces_with_labels(label_filter) diff --git a/test/integration_tests/init/test_recipe_job_creation.py b/test/integration_tests/init/test_recipe_job_creation.py index a0f89b27..a425d09a 100644 --- a/test/integration_tests/init/test_recipe_job_creation.py +++ b/test/integration_tests/init/test_recipe_job_creation.py @@ -180,22 +180,40 @@ def test_wait_for_recipe_job_running(job_name, test_directory): pytest.fail(f"Job {job_name} pods are crash-looping") if "Running" in output and "Status: True" in output: - print(f"[INFO] Job {job_name} is Running") + remaining_seconds = max(1, int(deadline - time.time())) + wait_timeout = min(POLL_INTERVAL_SECONDS, remaining_seconds) + print( + f"[INFO] Job {job_name} is Running; waiting for its pods " + "to be Ready..." + ) + execute_command([ + "kubectl", "wait", + "--for=condition=Ready", + "pod", + "--selector", f"HPJob={job_name}", + "--namespace", NAMESPACE, + f"--timeout={wait_timeout}s", + ]) + print(f"[INFO] Job {job_name} pods are Ready") return except RuntimeError as e: print(f"[DEBUG] Exception during polling: {e}") - time.sleep(POLL_INTERVAL_SECONDS) + remaining_seconds = deadline - time.time() + if remaining_seconds <= 0: + break + time.sleep(min(POLL_INTERVAL_SECONDS, remaining_seconds)) - pytest.fail(f"[ERROR] Timed out waiting for job {job_name} to be Running after {TIMEOUT_MINUTES}m") + pytest.fail( + f"[ERROR] Timed out waiting for job {job_name} pods to be Ready " + f"after {TIMEOUT_MINUTES}m" + ) @pytest.mark.dependency(name="list_pods", depends=["wait"]) def test_list_pods_recipe_job(job_name, test_directory): """List pods associated with the recipe job.""" - time.sleep(10) - result = execute_command([ "hyp", "list-pods", "hyp-recipe-job", "--job-name", job_name, diff --git a/test/unit_tests/clients/test_kubernetes_client.py b/test/unit_tests/clients/test_kubernetes_client.py index baa0670a..3fa46de6 100644 --- a/test/unit_tests/clients/test_kubernetes_client.py +++ b/test/unit_tests/clients/test_kubernetes_client.py @@ -418,7 +418,7 @@ def test_get_auth_v1_api( auth_v1_api = test_client.get_auth_v1_api() self.assertIsInstance(auth_v1_api, client.AuthorizationV1Api) - @patch("kubernetes.config.load_kube_config") + @patch("kubernetes.config.new_client_from_config") @patch( "kubernetes.client.CoreV1Api.list_node", return_value=nodes_no_next_token, @@ -426,13 +426,14 @@ def test_get_auth_v1_api( def test_list_node_with_temp_config_no_next_token( self, mock_core_client: Mock, - mock_kube_config: Mock, + mock_new_client: Mock, ): - mock_kube_config.return_value = None + mock_new_client.return_value = MagicMock() test_client = KubernetesClient() test_client.list_node_with_temp_config("temp-file", "temp-label") + mock_new_client.assert_called_once_with(config_file="temp-file") - @patch("kubernetes.config.load_kube_config") + @patch("kubernetes.config.new_client_from_config") @patch( "kubernetes.client.CoreV1Api.list_node", side_effect=[ @@ -443,11 +444,12 @@ def test_list_node_with_temp_config_no_next_token( def test_list_node_with_temp_config_no_next_token_with_pagination( self, mock_core_client: Mock, - mock_kube_config: Mock, + mock_new_client: Mock, ): - mock_kube_config.return_value = None + mock_new_client.return_value = MagicMock() test_client = KubernetesClient() test_client.list_node_with_temp_config("temp-file", "temp-label") + mock_new_client.assert_called_once_with(config_file="temp-file") @patch("kubernetes.client.CustomObjectsApi") def test_delete_training_job_with_namepsace( From 8131b3dcce35c42b03d4ed03d3ad651103490d91 Mon Sep 17 00:00:00 2001 From: aviruthen <91846056+aviruthen@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:01:52 -0700 Subject: [PATCH 2/3] fix: Isolate recipe test storage --- .../hyperpod/cli/clients/kubernetes_client.py | 25 +-- .../hyperpod/cli/commands/cluster.py | 11 +- .../hyperpod/cli/service/list_pods.py | 63 ++++--- .../init/test_recipe_job_creation.py | 173 +++++++++++++++--- .../clients/test_kubernetes_client.py | 4 - .../service/test_list_pods_service.py | 3 +- test/unit_tests/test_cluster.py | 36 ++-- 7 files changed, 222 insertions(+), 93 deletions(-) diff --git a/src/sagemaker/hyperpod/cli/clients/kubernetes_client.py b/src/sagemaker/hyperpod/cli/clients/kubernetes_client.py index 8df5002d..38ad45ae 100644 --- a/src/sagemaker/hyperpod/cli/clients/kubernetes_client.py +++ b/src/sagemaker/hyperpod/cli/clients/kubernetes_client.py @@ -67,21 +67,22 @@ class KubernetesClient: _kube_client = None def __new__(cls, config_file: Optional[str] = None) -> "KubernetesClient": - if config_file is not None: - instance = super(KubernetesClient, cls).__new__(cls) - instance._config_file = config_file - instance._kube_client = config.new_client_from_config( - config_file=config_file - ) - return instance - if cls._instance is None: cls._instance = super(KubernetesClient, cls).__new__(cls) - cls._instance._config_file = KUBE_CONFIG_PATH - config.load_kube_config(config_file=KUBE_CONFIG_PATH) + cls._instance._config_file = config_file or KUBE_CONFIG_PATH + config.load_kube_config(config_file=config_file or KUBE_CONFIG_PATH) cls._instance._kube_client = client.ApiClient() return cls._instance + @classmethod + def _from_config_file(cls, config_file: str) -> "KubernetesClient": + instance = super(KubernetesClient, cls).__new__(cls) + instance._config_file = config_file + instance._kube_client = config.new_client_from_config( + config_file=config_file + ) + return instance + def set_context( self, context_name: str, @@ -209,7 +210,7 @@ def list_node_with_temp_config( ) nodes.extend(response.items) - _continue = response._metadata._continue # Get the continue token + _continue = response.metadata._continue # Get the continue token if not _continue: # If there's no more data, break the loop break @@ -282,7 +283,7 @@ def list_pods_in_all_namespaces_with_labels(self, label_selector: str): ) pods.extend(response.items) - _continue = response._metadata._continue # Get the continue token + _continue = response.metadata._continue # Get the continue token if not _continue: # If there's no more data, break the loop break diff --git a/src/sagemaker/hyperpod/cli/commands/cluster.py b/src/sagemaker/hyperpod/cli/commands/cluster.py index 2aebdd71..bef7b82a 100644 --- a/src/sagemaker/hyperpod/cli/commands/cluster.py +++ b/src/sagemaker/hyperpod/cli/commands/cluster.py @@ -48,7 +48,7 @@ get_user_agent_extra_suffix, ) from sagemaker.hyperpod.cli.service.list_pods import ( - ListPods, + _list_pods_and_get_requested_resources_group_by_node_name, ) from sagemaker.hyperpod.cli.utils import ( get_name_from_arn, @@ -303,7 +303,7 @@ def rate_limited_operation( return None eks_cluster_name = get_name_from_arn(eks_cluster_arn) _update_kube_config(eks_cluster_name, region, temp_config_file) - k8s_client = KubernetesClient(config_file=temp_config_file) + k8s_client = KubernetesClient._from_config_file(temp_config_file) nodes = k8s_client.list_node_with_temp_config( temp_config_file, SAGEMAKER_HYPERPOD_NAME_LABEL ) @@ -466,13 +466,10 @@ def _restructure_output(summary_list, namespaces): def _aggregate_nodes_info( nodes: List[client.V1Node], - k8s_client: Optional[KubernetesClient] = None, + k8s_client: KubernetesClient, ) -> Dict[str, Dict[str, Any]]: - list_pods_service = ListPods() nodes_resource_allocated_dict = ( - list_pods_service.list_pods_and_get_requested_resources_group_by_node_name( - k8s_client - ) + _list_pods_and_get_requested_resources_group_by_node_name(k8s_client) ) nodes_summary: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int)) diff --git a/src/sagemaker/hyperpod/cli/service/list_pods.py b/src/sagemaker/hyperpod/cli/service/list_pods.py index cc2bf554..f4ef8839 100644 --- a/src/sagemaker/hyperpod/cli/service/list_pods.py +++ b/src/sagemaker/hyperpod/cli/service/list_pods.py @@ -27,6 +27,36 @@ from sagemaker.hyperpod.cli.constants.command_constants import NVIDIA_GPU_RESOURCE_LIMIT_KEY from sagemaker.hyperpod.cli.service.discover_namespaces import DiscoverNamespaces +def _list_pods_and_get_requested_resources_group_by_node_name( + k8s_client: KubernetesClient, +): + label_filter = "training.kubeflow.org/job-name" + pods = k8s_client.list_pods_in_all_namespaces_with_labels(label_filter) + + accelerator_devices_requests_by_node = defaultdict(int) + for pod in pods: + node_name = pod.spec.node_name + if node_name: + for container in pod.spec.containers: + if container.resources and container.resources.requests: + gpu_request = container.resources.requests.get( + NVIDIA_GPU_RESOURCE_LIMIT_KEY + ) + neuron_request = container.resources.requests.get( + "aws.amazon.com/neurondevice" + ) + if gpu_request: + accelerator_devices_requests_by_node[node_name] += int( + gpu_request + ) + if neuron_request: + accelerator_devices_requests_by_node[node_name] += int( + neuron_request + ) + + return accelerator_devices_requests_by_node + + class ListPods: def __init__(self): return @@ -66,42 +96,15 @@ def list_pods_for_training_job( def list_pods_and_get_requested_resources_group_by_node_name( self, - k8s_client: Optional[KubernetesClient] = None, ): """ List pods for all namespaces and initialized by kubeflow. Group by the node_name of the pod and the value is the accelerator devices resources requested """ - k8s_client = k8s_client or KubernetesClient() - - label_filter = f"training.kubeflow.org/job-name" - pods = k8s_client.list_pods_in_all_namespaces_with_labels(label_filter) - - # Dictionary to hold total GPU/Neuron requests per node - accelerator_devices_requests_by_node = defaultdict(int) - - # Loop through each pod and aggregate the GPU resource requests per node - for pod in pods: - node_name = pod.spec.node_name - # Check if the pod has a node_name assigned, if pod is not scheduled, it may not have node_name - if node_name: - for container in pod.spec.containers: - if container.resources and container.resources.requests: - gpu_request = container.resources.requests.get(NVIDIA_GPU_RESOURCE_LIMIT_KEY) - neuron_request = container.resources.requests.get( - "aws.amazon.com/neurondevice" - ) - if gpu_request: - accelerator_devices_requests_by_node[node_name] += int( - gpu_request - ) - if neuron_request: - accelerator_devices_requests_by_node[node_name] += int( - neuron_request - ) - - return accelerator_devices_requests_by_node + return _list_pods_and_get_requested_resources_group_by_node_name( + KubernetesClient() + ) def _generate_list_pods_output(self, pods: V1PodList) -> Optional[str]: output_pods = {"pods": []} diff --git a/test/integration_tests/init/test_recipe_job_creation.py b/test/integration_tests/init/test_recipe_job_creation.py index a425d09a..545015ae 100644 --- a/test/integration_tests/init/test_recipe_job_creation.py +++ b/test/integration_tests/init/test_recipe_job_creation.py @@ -7,15 +7,18 @@ Tests complete user workflow: init -> configure -> validate -> create -> wait -> delete. Uses real AWS resources (SageMaker Hub, S3, Kubernetes) with cost implications. """ -import time -import yaml -import pytest import os +import re import sys import tempfile +import time +import uuid from pathlib import Path from unittest.mock import patch +import boto3 +import pytest +import yaml from click.testing import CliRunner from sagemaker.hyperpod.cli.commands.init import init, configure, validate, reset from test.integration_tests.init.utils import ( @@ -42,7 +45,7 @@ def runner(): @pytest.fixture(scope="module") def job_name(): - return "recipe-integ-" + get_time_str() + return "recipe-integ-{}-{}".format(get_time_str(), uuid.uuid4().hex[:8]) @pytest.fixture(scope="module") @@ -56,6 +59,121 @@ def test_directory(): os.chdir(original_cwd) +def _run_cleanup_command(command, errors): + try: + execute_command(command) + except RuntimeError as error: + errors.append(str(error)) + + +@pytest.fixture(scope="module") +def recipe_fsx_resources(job_name, test_directory): + """Create an isolated static FSx PV/PVC for this recipe test run.""" + fsx_id = os.getenv("FSX_ID") + assert fsx_id, "FSX_ID must be set for recipe integration tests" + + response = boto3.client("fsx", region_name=REGION).describe_file_systems( + FileSystemIds=[fsx_id] + ) + file_systems = response.get("FileSystems", []) + assert len(file_systems) == 1, "Expected exactly one FSx filesystem" + + file_system = file_systems[0] + assert file_system.get("FileSystemType") == "LUSTRE" + assert file_system.get("Lifecycle") == "AVAILABLE" + + lustre_config = file_system.get("LustreConfiguration", {}) + dns_name = file_system.get("DNSName") + mount_name = lustre_config.get("MountName") + storage = "{}Gi".format(file_system.get("StorageCapacity")) + assert dns_name and mount_name + + pv_name = "{}-fsx-pv".format(job_name) + pvc_name = "{}-fsx-pvc".format(job_name) + assert pv_name != "fsx-pv-new" + assert pvc_name != "fsx-claim" + + manifest = [ + { + "apiVersion": "v1", + "kind": "PersistentVolume", + "metadata": {"name": pv_name}, + "spec": { + "capacity": {"storage": storage}, + "volumeMode": "Filesystem", + "accessModes": ["ReadWriteMany"], + "mountOptions": ["flock"], + "persistentVolumeReclaimPolicy": "Retain", + "storageClassName": "", + "csi": { + "driver": "fsx.csi.aws.com", + "volumeHandle": fsx_id, + "volumeAttributes": { + "dnsname": dns_name, + "mountname": mount_name, + }, + }, + }, + }, + { + "apiVersion": "v1", + "kind": "PersistentVolumeClaim", + "metadata": {"name": pvc_name, "namespace": NAMESPACE}, + "spec": { + "accessModes": ["ReadWriteMany"], + "volumeMode": "Filesystem", + "storageClassName": "", + "volumeName": pv_name, + "resources": {"requests": {"storage": storage}}, + }, + }, + ] + manifest_path = Path(test_directory) / ".recipe-fsx-resources.yaml" + manifest_path.write_text(yaml.safe_dump_all(manifest, sort_keys=False)) + + try: + execute_command(["kubectl", "apply", "-f", str(manifest_path)]) + execute_command([ + "kubectl", "wait", + "--for=jsonpath={.status.phase}=Bound", + "persistentvolumeclaim/{}".format(pvc_name), + "--namespace", NAMESPACE, + "--timeout=120s", + ]) + yield {"pv_name": pv_name, "pvc_name": pvc_name} + finally: + cleanup_errors = [] + _run_cleanup_command([ + "kubectl", "delete", "hyperpodpytorchjob", job_name, + "--namespace", NAMESPACE, + "--ignore-not-found=true", "--wait=true", "--timeout=180s", + ], cleanup_errors) + _run_cleanup_command([ + "kubectl", "delete", "pod", + "--selector", "HPJob={}".format(job_name), + "--namespace", NAMESPACE, + "--ignore-not-found=true", "--wait=true", "--timeout=180s", + ], cleanup_errors) + _run_cleanup_command([ + "kubectl", "delete", "configmap", "training-config-{}".format(job_name), + "--namespace", NAMESPACE, + "--ignore-not-found=true", "--wait=true", "--timeout=180s", + ], cleanup_errors) + _run_cleanup_command([ + "kubectl", "delete", "persistentvolumeclaim", pvc_name, + "--namespace", NAMESPACE, + "--ignore-not-found=true", "--wait=true", "--timeout=180s", + ], cleanup_errors) + _run_cleanup_command([ + "kubectl", "delete", "persistentvolume", pv_name, + "--ignore-not-found=true", "--wait=true", "--timeout=180s", + ], cleanup_errors) + if cleanup_errors: + raise RuntimeError("Recipe resource cleanup failed: {}".format( + "; ".join(cleanup_errors) + )) + + # --------- Recipe Job Tests --------- @pytest.mark.dependency(name="init") @@ -73,17 +191,28 @@ def test_init_recipe_job(runner, job_name, test_directory): assert (base / "config.yaml").exists(), f"config.yaml not found in {test_directory}. Output: {result.output}" assert (base / ".override_spec.json").exists(), f".override_spec.json not found in {test_directory}" assert (base / "k8s.jinja").exists(), f"k8s.jinja not found in {test_directory}" + assert "claimName: fsx-claim" in (base / "k8s.jinja").read_text() @pytest.mark.dependency(name="configure", depends=["init"]) -def test_configure_recipe_job(runner, job_name, test_directory): +def test_configure_recipe_job( + runner, job_name, test_directory, recipe_fsx_resources +): """Configure recipe job with training parameters.""" - # Patch EFA to 1 in k8s.jinja — the S3 template may ship with a higher value - import re + # Patch this run's PVC and reduce EFA in the downloaded template. k8s_jinja = Path("k8s.jinja") - k8s_jinja.write_text( - re.sub(r'(vpc\.amazonaws\.com/efa:\s*)\d+', r'\g<1>1', k8s_jinja.read_text()) + template = k8s_jinja.read_text() + default_claim = "claimName: fsx-claim" + assert template.count(default_claim) == 1 + template = template.replace( + default_claim, + "claimName: {}".format(recipe_fsx_resources["pvc_name"]), + 1, + ) + template = re.sub( + r"(vpc\.amazonaws\.com/efa:\s*)\d+", r"\g<1>1", template ) + k8s_jinja.write_text(template) with patch.object(sys, 'argv', ['hyp', 'configure']): import importlib @@ -99,9 +228,9 @@ def test_configure_recipe_job(runner, job_name, test_directory): "--learning-rate", "0.0001", "--lr-warmup-steps-ratio", "0.1", "--max-epochs", "20", - "--output-path", "/data/output/qwen3-sft", - "--results-directory", "/data/results/qwen3-sft", - "--resume-from-path", "/data/output/qwen3-sft", + "--output-path", "/data/output/{}".format(job_name), + "--results-directory", "/data/results/{}".format(job_name), + "--resume-from-path", "/data/output/{}".format(job_name), "--training-data-name", "zc_train_256", "--validation-data-name", "zc_train_256", "--validation-data-path", "/data/recipes-data/sft/zc_train_256.jsonl", @@ -125,7 +254,9 @@ def test_validate_recipe_job(runner, job_name, test_directory): @pytest.mark.dependency(name="create", depends=["validate", "configure", "init"]) -def test_create_recipe_job(runner, job_name, test_directory): +def test_create_recipe_job( + runner, job_name, test_directory, recipe_fsx_resources +): """Submit recipe job to Kubernetes and verify submission.""" from sagemaker.hyperpod.cli.commands.init import _default_create as create_cmd @@ -140,6 +271,8 @@ def test_create_recipe_job(runner, job_name, test_directory): # Verify job name appears in rendered output k8s_yaml = (run_dir / "k8s.yaml").read_text() assert job_name in k8s_yaml + assert "claimName: {}".format(recipe_fsx_resources["pvc_name"]) in k8s_yaml + assert "claimName: fsx-claim" not in k8s_yaml @pytest.mark.dependency(name="wait", depends=["create"]) @@ -229,16 +362,14 @@ def test_list_pods_recipe_job(job_name, test_directory): @pytest.mark.dependency(name="get_logs", depends=["wait"]) def test_get_logs_recipe_job(job_name, test_directory): """Get logs from the first pod of the recipe job.""" - pods_result = execute_command([ - "hyp", "list-pods", "hyp-recipe-job", - "--job-name", job_name, + pod_result = execute_command([ + "kubectl", "get", "pods", "--namespace", NAMESPACE, + "--selector", "HPJob={}".format(job_name), + "--output", "jsonpath={.items[0].metadata.name}", ]) - assert pods_result.returncode == 0 - # Extract first pod name — pod rows start with the job name - lines = [l for l in pods_result.stdout.splitlines() if l.strip().startswith(job_name)] - assert lines, "No pod found for job" - pod_name = lines[0].split()[0] + pod_name = pod_result.stdout.strip() + assert pod_name, "No pod found for job" result = execute_command([ "hyp", "get-logs", "hyp-recipe-job", diff --git a/test/unit_tests/clients/test_kubernetes_client.py b/test/unit_tests/clients/test_kubernetes_client.py index 3fa46de6..44fb0e2d 100644 --- a/test/unit_tests/clients/test_kubernetes_client.py +++ b/test/unit_tests/clients/test_kubernetes_client.py @@ -80,7 +80,6 @@ metadata=V1ObjectMeta( name="test-name", namespace="kubeflow", - creation_timestamp="timestamp", ) ) ], @@ -95,7 +94,6 @@ metadata=V1ObjectMeta( name="test-name", namespace="kubeflow", - creation_timestamp="timestamp", ) ) ], @@ -110,7 +108,6 @@ metadata=V1ObjectMeta( name="test-name", namespace="kubeflow", - creation_timestamp="timestamp", ) ) ], @@ -125,7 +122,6 @@ metadata=V1ObjectMeta( name="test-name", namespace="kubeflow", - creation_timestamp="timestamp", ) ) ], diff --git a/test/unit_tests/service/test_list_pods_service.py b/test/unit_tests/service/test_list_pods_service.py index 6bee69c5..e3a9cc35 100644 --- a/test/unit_tests/service/test_list_pods_service.py +++ b/test/unit_tests/service/test_list_pods_service.py @@ -10,6 +10,7 @@ # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. +from datetime import datetime, timezone import unittest from unittest import mock from unittest.mock import MagicMock @@ -40,7 +41,7 @@ metadata=V1ObjectMeta( name="test-name", namespace="kubeflow", - creation_timestamp="timestamp", + creation_timestamp=datetime(2024, 1, 1, tzinfo=timezone.utc), ), spec=V1PodSpec( node_name="test-node-name", diff --git a/test/unit_tests/test_cluster.py b/test/unit_tests/test_cluster.py index 37d52ce3..2621bbf7 100644 --- a/test/unit_tests/test_cluster.py +++ b/test/unit_tests/test_cluster.py @@ -266,7 +266,7 @@ def test_connect_validator_failure(self, mock_open_test, mock_validator_cls): @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_get_clusters( self, @@ -319,7 +319,7 @@ def test_get_clusters( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") @mock.patch("logging.Logger.debug") def test_get_clusters_debug_mode( @@ -375,7 +375,7 @@ def test_get_clusters_debug_mode( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_get_clusters_maximum_number( self, @@ -434,7 +434,7 @@ def test_get_clusters_maximum_number( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_get_clusters_no_cluster_summary( self, @@ -486,7 +486,7 @@ def test_get_clusters_no_cluster_summary( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_get_clusters_table_output( self, @@ -540,10 +540,10 @@ def test_get_clusters_table_output( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") @mock.patch( - "sagemaker.hyperpod.cli.service.list_pods.ListPods.list_pods_and_get_requested_resources_group_by_node_name" + "sagemaker.hyperpod.cli.commands.cluster._list_pods_and_get_requested_resources_group_by_node_name" ) def test_get_clusters_with_deep_health_check_enabled_and_gpu_devices( self, @@ -599,10 +599,10 @@ def test_get_clusters_with_deep_health_check_enabled_and_gpu_devices( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") @mock.patch( - "sagemaker.hyperpod.cli.service.list_pods.ListPods.list_pods_and_get_requested_resources_group_by_node_name" + "sagemaker.hyperpod.cli.commands.cluster._list_pods_and_get_requested_resources_group_by_node_name" ) def test_get_clusters_with_unexpected_health_status( self, @@ -659,10 +659,10 @@ def test_get_clusters_with_unexpected_health_status( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") @mock.patch( - "sagemaker.hyperpod.cli.service.list_pods.ListPods.list_pods_and_get_requested_resources_group_by_node_name" + "sagemaker.hyperpod.cli.commands.cluster._list_pods_and_get_requested_resources_group_by_node_name" ) def test_get_clusters_with_no_status( self, @@ -753,7 +753,7 @@ def test_get_clusters_sm_client_unexpected_error( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_list_clusters_with_clusters_list( self, @@ -808,7 +808,7 @@ def test_list_clusters_with_clusters_list( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_list_clusters_failed_list_cluster_error( self, @@ -858,7 +858,7 @@ def test_list_clusters_failed_list_cluster_error( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_list_clusters_failed_unexpected_error( self, @@ -911,7 +911,7 @@ def test_list_clusters_failed_unexpected_error( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_list_clusters_skipped_not_eks_clusters( self, @@ -962,7 +962,7 @@ def test_list_clusters_skipped_not_eks_clusters( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_get_clusters_with_sm_managed_namespace( self, @@ -1043,7 +1043,7 @@ def test_get_clusters_with_sm_managed_namespace( @mock.patch( "sagemaker.hyperpod.cli.commands.cluster.ClusterValidator.validate_cluster_and_get_eks_arn" ) - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("subprocess.run") def test_get_clusters_with_not_sm_managed_namespace( self, @@ -1105,7 +1105,7 @@ def test_get_clusters_with_not_sm_managed_namespace( json.loads(result.output) @mock.patch("subprocess.run") - @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient.__new__") + @mock.patch("sagemaker.hyperpod.cli.clients.kubernetes_client.KubernetesClient._from_config_file") @mock.patch("sagemaker.hyperpod.cli.validators.cluster_validator.ClusterValidator.validate_cluster_and_get_eks_arn") @mock.patch("sagemaker.hyperpod.cli.validators.cluster_validator.ClusterValidator.validate_aws_credential") @mock.patch("boto3.Session") From 46890062b828a379a23db68b68cb4410e2795907 Mon Sep 17 00:00:00 2001 From: aviruthen <91846056+aviruthen@users.noreply.github.com> Date: Thu, 15 Oct 2026 11:07:36 -0700 Subject: [PATCH 3/3] fix: Support Kubernetes 37 models --- .../hyperpod/cli/service/self_subject_access_review.py | 1 - .../service/test_self_subject_access_review.py | 10 ++++++++-- test/unit_tests/test_cluster.py | 6 +++--- 3 files changed, 11 insertions(+), 6 deletions(-) diff --git a/src/sagemaker/hyperpod/cli/service/self_subject_access_review.py b/src/sagemaker/hyperpod/cli/service/self_subject_access_review.py index 6bf49307..332d81cb 100644 --- a/src/sagemaker/hyperpod/cli/service/self_subject_access_review.py +++ b/src/sagemaker/hyperpod/cli/service/self_subject_access_review.py @@ -38,7 +38,6 @@ def self_subject_access_review( spec=V1SelfSubjectAccessReviewSpec( resource_attributes=resource_attributes, non_resource_attributes=non_resource_attributes, - local_vars_configuration=local_vars_configuration, ) ) diff --git a/test/unit_tests/service/test_self_subject_access_review.py b/test/unit_tests/service/test_self_subject_access_review.py index a73b7fb2..4f872454 100644 --- a/test/unit_tests/service/test_self_subject_access_review.py +++ b/test/unit_tests/service/test_self_subject_access_review.py @@ -27,7 +27,7 @@ def test_self_subject_access_review_success(self, mock_kubernetes_client): mock_kubernetes_client().get_auth_v1_api.return_value = mock_auth_v1_api # Create a mock response for create_self_subject_access_review - mock_response = MagicMock(spec=V1SelfSubjectAccessReview) + mock_response = MagicMock() mock_response.status.allowed = True mock_auth_v1_api.create_self_subject_access_review.return_value = mock_response @@ -53,7 +53,13 @@ def test_self_subject_access_review_success(self, mock_kubernetes_client): # Capture and inspect the arguments passed in the call args, kwargs = mock_auth_v1_api.create_self_subject_access_review.call_args self.assertIsInstance(kwargs['body'], V1SelfSubjectAccessReview) - self.assertEqual(kwargs['body'].spec.resource_attributes['namespace'], 'test-namespace') + resource_attributes_body = kwargs["body"].spec.resource_attributes + namespace = ( + resource_attributes_body["namespace"] + if isinstance(resource_attributes_body, dict) + else resource_attributes_body.namespace + ) + self.assertEqual(namespace, "test-namespace") if __name__ == "__main__": diff --git a/test/unit_tests/test_cluster.py b/test/unit_tests/test_cluster.py index 2621bbf7..6241195b 100644 --- a/test/unit_tests/test_cluster.py +++ b/test/unit_tests/test_cluster.py @@ -1190,7 +1190,7 @@ def _generate_deep_health_check_nodes_list(): DEEP_HEALTH_CHECK_STATUS_LABEL: "Passed", }, ), - status=client.V1NodeStatus(allocatable={"nvidia.com/gpu": 24}), + status=client.V1NodeStatus(allocatable={"nvidia.com/gpu": "24"}), ), client.V1Node( metadata=client.V1ObjectMeta( @@ -1201,7 +1201,7 @@ def _generate_deep_health_check_nodes_list(): INSTANCE_TYPE_LABEL: "ml.trn1.32xlarge", }, ), - status=client.V1NodeStatus(allocatable={"aws.amazon.com/neurondevice": 16}), + status=client.V1NodeStatus(allocatable={"aws.amazon.com/neurondevice": "16"}), ), client.V1Node( metadata=client.V1ObjectMeta( @@ -1211,7 +1211,7 @@ def _generate_deep_health_check_nodes_list(): INSTANCE_TYPE_LABEL: "ml.trn1.32xlarge", }, ), - status=client.V1NodeStatus(allocatable={"aws.amazon.com/neurondevice": 16}), + status=client.V1NodeStatus(allocatable={"aws.amazon.com/neurondevice": "16"}), ), client.V1Node( metadata=client.V1ObjectMeta(