adjusted unit tests to mock k8s connectivity

Author: Mohamed Zeidan

test/unit_tests/cli/test_inference.py

@@ -7,6 +7,18 @@
 import hyperpod_jumpstart_inference_template.registry as jreg
 import hyperpod_custom_inference_template.registry as creg
 
+# Mock Kubernetes connectivity for all tests in this module
+@pytest.fixture(autouse=True)
+def mock_kubernetes_connectivity():
+    """Mock Kubernetes connectivity checks for all tests in this module."""
+    with patch('sagemaker.hyperpod.common.cli_decorators._check_kubernetes_connectivity') as mock_connectivity, \
+         patch('sagemaker.hyperpod.common.cli_decorators._is_kubernetes_operation') as mock_is_k8s_op:
+        # Always return successful connectivity
+        mock_connectivity.return_value = (True, "")
+        # Let the operation detection work normally
+        mock_is_k8s_op.side_effect = lambda func, **kwargs: True
+        yield
+
 # Import the non-create commands that don't need special handling
 from sagemaker.hyperpod.cli.commands.inference import (
     js_create, custom_create, custom_invoke,

test/unit_tests/cli/test_training.py

@@ -13,10 +13,26 @@
 import sys
 import os
 import importlib
+import pytest
 
 # Add the hyperpod-pytorch-job-template to the path for testing
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..', 'hyperpod-pytorch-job-template'))
 
+# Mock Kubernetes connectivity for all tests in this module
+@pytest.fixture(autouse=True)
+def mock_kubernetes_connectivity():
+    """Mock Kubernetes connectivity checks for all tests in this module."""
+    with patch('sagemaker.hyperpod.common.cli_decorators._check_kubernetes_connectivity') as mock_connectivity, \
+         patch('sagemaker.hyperpod.common.cli_decorators._is_kubernetes_operation') as mock_is_k8s_op, \
+         patch('sagemaker.hyperpod.common.cli_decorators._check_training_operator_exists') as mock_training_op:
+        # Always return successful connectivity
+        mock_connectivity.return_value = (True, "")
+        # Let the operation detection work normally
+        mock_is_k8s_op.side_effect = lambda func, **kwargs: True
+        # Always return that training operator exists
+        mock_training_op.return_value = True
+        yield
+
 try:
     from hyperpod_pytorch_job_template.v1_1.model import PyTorchJobConfig, VolumeConfig
     from pydantic import ValidationError

test/unit_tests/error_handling/test_cli_decorators.py

@@ -21,6 +21,18 @@
     _is_get_logs_operation
 )
 
+# Mock Kubernetes connectivity for all tests in this module
+@pytest.fixture(autouse=True)
+def mock_kubernetes_connectivity():
+    """Mock Kubernetes connectivity checks for all tests in this module."""
+    with patch('sagemaker.hyperpod.common.cli_decorators._check_kubernetes_connectivity') as mock_connectivity, \
+         patch('sagemaker.hyperpod.common.cli_decorators._is_kubernetes_operation') as mock_is_k8s_op:
+        # Always return successful connectivity
+        mock_connectivity.return_value = (True, "")
+        # Let the operation detection work normally
+        mock_is_k8s_op.side_effect = lambda func, **kwargs: True
+        yield
+
 
 class TestHandleCliExceptions:
     """Test template-agnostic handle_cli_exceptions decorator."""