Repository: skypilot-org/skypilot
Pull request #5736 (2025-05-26 03:36:59)

[Workspace] Allow specification of `kubernetes.allowed_contexts`/`ssh.allowed_node_pools` and detailed infra in enabled infra
<!-- Describe the changes in this PR -->

An inconsistency on master:
- Kubernetes takes only the active context when no `allowed_contexts` is specified
- SSH Node Pool takes all the available context

TODO:
- [x] show allowed contexts in the enabled infra for workspace
<img width="2542" alt="image" src="https://github.com/user-attachments/assets/2105980d-ad2b-4f0f-a6ce-3e457b388317" />


<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [ ] Code formatting: install pre-commit (auto-check on commit) or `bash format.sh`
- [ ] Any manual or new tests for this PR (please specify below)
  - [x] Specifying allowed_contexts in a workspace and check the allowed_contexts
- [ ] All smoke tests: `/smoke-test` (CI) or `pytest tests/test_smoke.py` (local)
- [ ] Relevant individual tests: `/smoke-test -k test_name` (CI) or `pytest tests/test_smoke.py::test_name` (local)
- [ ] Backward compatibility: `/quicktest-core` (CI) or `pytest tests/smoke_tests/test_backward_compat.py` (local)

<!-- CI commands (/-prefixed) can only be triggered by repo members -->



Patch:
diff --git a/docs/source/admin/workspaces.rst b/docs/source/admin/workspaces.rst
index 79449a8d8..337b18621 100644
--- a/docs/source/admin/workspaces.rst
+++ b/docs/source/admin/workspaces.rst
@@ -60,6 +60,15 @@ The above is achieved by configuring the following section in the config file:
            gcp:
              disabled: false
              project_id: GCP project ID
+           kubernetes:
+             disabled: false
+             allowed_contexts:
+               - node-pool-1
+               - node-pool-2
+           ssh:
+             disabled: false
+             allowed_node_pools:
+               - node-pool-1
 
 To apply the configuration, follow the following steps:
 
diff --git a/docs/source/reference/api-server/api-server-admin-deploy.rst b/docs/source/reference/api-server/api-server-admin-deploy.rst
index 64290f2a0..2c0957d35 100644
--- a/docs/source/reference/api-server/api-server-admin-deploy.rst
+++ b/docs/source/reference/api-server/api-server-admin-deploy.rst
@@ -191,7 +191,7 @@ Following tabs describe how to configure credentials for different clouds on the
 
             kubectl create secret generic kube-credentials \
               --namespace $NAMESPACE \
-              --from-file=config=~/.kube/config
+              --from-file=config=$HOME/.kube/config
 
 
         Once the secret is created, set ``kubernetesCredentials.useKubeconfig=true`` and ``kubernetesCredentials.kubeconfigSecretName`` in the Helm chart values to use the kubeconfig for authentication:
@@ -216,6 +216,8 @@ Following tabs describe how to configure credentials for different clouds on the
 
             Then create the Kubernetes secret with the converted kubeconfig file ``~/.kube/config.converted``.
 
+            The specific cloud's credential for the exec-based authentication also needs to be configured. For example, to enable exec-based authentication for GKE, you also need to setup GCP credentials (see the GCP tab above).
+
         To use multiple Kubernetes clusters, you will need to add the context names to ``allowed_contexts`` in the SkyPilot config. An example config file that allows using the hosting Kubernetes cluster and two additional Kubernetes clusters is shown below:
 
         .. code-block:: yaml
@@ -485,11 +487,10 @@ Optional: Update SkyPilot configuration
 
    .. code-block:: bash
 
-     POD_NAME=$(kubectl get pod -l app=${RELEASE_NAME}-api -o jsonpath='{.items[0].metadata.name}')
-     # Make sure the pod is running
-     kubectl get pod $POD_NAME -n $NAMESPACE | grep -q "Running" || exit 1
-     # Fetch the config
-     kubectl exec -it $POD_NAME -n $NAMESPACE -- cat /root/.sky/config.yaml > current-config.yaml
+     POD_NAME=$(kubectl get pod -l app=${RELEASE_NAME}-api -n $NAMESPACE -o jsonpath='{.items[0].metadata.name}')
+     # Make sure the pod is running and download the config
+     kubectl get pod $POD_NAME -n $NAMESPACE | grep -q "Running" && \
+     kubectl cp $POD_NAME:/root/.sky/config.yaml current-config.yaml -n $NAMESPACE --no-preserve
 
 2. Edit the configuration file ``current-config.yaml`` with your desired changes.
 3. Upload the updated config to the API server:
diff --git a/docs/source/reference/config.rst b/docs/source/reference/config.rst
index aecd27c68..8ee360e83 100644
--- a/docs/source/reference/config.rst
+++ b/docs/source/reference/config.rst
@@ -76,6 +76,11 @@ Below is the configuration syntax and some example values. See detailed explanat
           my-label: my-value
       spec:
         runtimeClassName: nvidia
+  
+  :ref: `ssh <config-yaml-ssh>`:
+    :ref: `allowed_node_pools <config-yaml-ssh-allowed-node-pools>`:
+      - node-pool-1
+      - node-pool-2
 
   :ref:`aws <config-yaml-aws>`:
     :ref:`labels <config-yaml-aws-labels>`:
@@ -1062,6 +1067,20 @@ Example:
                 medium: Memory
                 sizeLimit: 3Gi
 
+``ssh``
+~~~~~~~~~~~~~~~
+
+Advanced SSH node pool configuration (optional).
+
+.. _config-yaml-ssh-allowed-node-pools:
+
+``ssh.allowed_node_pools``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+List of allowed SSH node pools (optional).
+
+List of names that SkyPilot is allowed to use.
+
 .. _config-yaml-oci:
 
 ``oci``
@@ -1159,6 +1178,7 @@ Example:
                 mount_path: /mnt/fsnew2
                 attach_mode: READ_ONLY
 
+
 .. _config-yaml-nebius-use-internal-ips:
 
 ``nebius.use_internal_ips``
diff --git a/sky/check.py b/sky/check.py
index d8a5e31ca..559167577 100644
--- a/sky/check.py
+++ b/sky/check.py
@@ -504,6 +504,18 @@ def _format_context_details(cloud: Union[str, sky_clouds.Cloud],
                 f'{str_to_format}'
                 f'{colorama.Style.RESET_ALL}')
 
+    # For SSH, determine which contexts are disabled due to allowed_node_pools
+    disabled_due_to_allowed_node_pools = set()
+    if isinstance(cloud_type, sky_clouds.SSH):
+        # Get all node pool contexts from file
+        all_node_pool_contexts = sky_clouds.SSH.get_ssh_node_pool_contexts()
+        # Get allowed contexts (after filtering)
+        allowed_contexts = sky_clouds.SSH.existing_allowed_contexts()
+        # Contexts that exist in file but not in allowed list are disabled
+        # due to allowed_node_pools configuration
+        disabled_due_to_allowed_node_pools = (set(all_node_pool_contexts) -
+                                              set(allowed_contexts))
+
     # Format the context info with consistent styling
     contexts_formatted = []
     for i, context in enumerate(filtered_contexts):
@@ -519,11 +531,22 @@ def _format_context_details(cloud: Union[str, sky_clouds.Cloud],
         text_suffix = ''
         if show_details:
             if ctx2text is not None:
-                text_suffix = (
-                    f': {ctx2text[context]}' if context in ctx2text else
-                    (': ' + _red_color('disabled. ') +
-                     _dim_color('Reason: Not set up. Use `sky ssh up --infra '
-                                f'{context.lstrip("ssh-")}` to set up.')))
+                if context in ctx2text:
+                    text_suffix = f': {ctx2text[context]}'
+                elif (isinstance(cloud_type, sky_clouds.SSH) and
+                      context in disabled_due_to_allowed_node_pools):
+                    # Context is disabled due to allowed_node_pools config
+                    text_suffix = (': ' + _red_color('disabled. ') +
+                                   _dim_color('Reason: Not included in '
+                                              'allowed_node_pools '
+                                              'configuration.'))
+                else:
+                    # Default case - not set up
+                    text_suffix = (': ' + _red_color('disabled. ') +
+                                   _dim_color('Reason: Not set up. Use '
+                                              '`sky ssh up --infra '
+                                              f'{context.lstrip("ssh-")}` '
+                                              'to set up.'))
         contexts_formatted.append(
             f'\n    {symbol}{cleaned_context}{text_suffix}')
     identity_str = ('SSH Node Pools' if isinstance(cloud_type, sky_clouds.SSH)
diff --git a/sky/cli.py b/sky/cli.py
index 90d7855fc..ec1ecf818 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -1985,7 +1985,7 @@ def status(verbose: bool, refresh: bool, ip: bool, endpoints: bool,
             workspace_request_id = sdk.workspaces()
         except RuntimeError:
             # Backward compatibility for API server before #5660.
-            # TODO(zhwu): remove this after 0.12.0.
+            # TODO(zhwu): remove this after 0.10.0.
             logger.warning(f'{colorama.Style.DIM}SkyPilot API server is '
                            'in an old version, and may miss feature: '
                            'workspaces. Update with: sky api stop; '
diff --git a/sky/clouds/kubernetes.py b/sky/clouds/kubernetes.py
index 2b889300e..b3166144c 100644
--- a/sky/clouds/kubernetes.py
+++ b/sky/clouds/kubernetes.py
@@ -164,15 +164,20 @@ class Kubernetes(clouds.Cloud):
 
         all_contexts = set(all_contexts)
 
+        # Allowed_contexts specified for workspace should take precedence over
+        # the global allowed_contexts.
+        allowed_contexts = skypilot_config.get_workspace_cloud(
+            'kubernetes').get('allowed_contexts', None)
+        if allowed_contexts is None:
+            allowed_contexts = skypilot_config.get_nested(
+                ('kubernetes', 'allowed_contexts'), None)
+
         # Exclude contexts starting with `ssh-`
         # TODO(romilb): Remove when SSH Node Pools use a separate kubeconfig.
         all_contexts = [
             ctx for ctx in all_contexts if not ctx.startswith('ssh-')
         ]
 
-        allowed_contexts = skypilot_config.get_nested(
-            ('kubernetes', 'allowed_contexts'), None)
-
         if allowed_contexts is None:
             # Try kubeconfig if present
             current_context = (
diff --git a/sky/clouds/ssh.py b/sky/clouds/ssh.py
index 4b8f08f3c..6b257cb58 100644
--- a/sky/clouds/ssh.py
+++ b/sky/clouds/ssh.py
@@ -7,6 +7,7 @@ from typing import Dict, List, Optional, Set, Tuple, Union
 import yaml
 
 from sky import sky_logging
+from sky import skypilot_config
 from sky.adaptors import kubernetes as kubernetes_adaptor
 from sky.clouds import kubernetes
 from sky.provision.kubernetes import utils as kubernetes_utils
@@ -134,29 +135,44 @@ class SSH(kubernetes.Kubernetes):
 
         all_contexts = set(all_contexts)
 
+        # Workspace-level allowed_node_pools should take precedence over
+        # the global allowed_node_pools.
+        allowed_node_pools = skypilot_config.get_workspace_cloud('ssh').get(
+            'allowed_node_pools', None)
+        if allowed_node_pools is None:
+            allowed_node_pools = skypilot_config.get_nested(
+                ('ssh', 'allowed_node_pools'), None)
+
         # Filter for SSH contexts (those starting with 'ssh-')
         ssh_contexts = [
             context for context in all_contexts if context.startswith('ssh-')
         ]
 
         # Get contexts from SSH node pools file
-        allowed_contexts = cls.get_ssh_node_pool_contexts()
+        all_node_pool_contexts = cls.get_ssh_node_pool_contexts()
+
+        def filter_by_allowed_node_pools(ctxs):
+            if allowed_node_pools is None:
+                return ctxs
+            return [
+                ctx for ctx in ctxs if ctx.lstrip('ssh-') in allowed_node_pools
+            ]
 
-        if allowed_contexts:
+        if all_node_pool_contexts:
             # Only include allowed contexts that exist
             existing_contexts = []
             skipped_contexts = []
-            for context in allowed_contexts:
+            for context in all_node_pool_contexts:
                 if context in ssh_contexts:
                     existing_contexts.append(context)
                 else:
                     skipped_contexts.append(context)
             if not silent:
                 cls._ssh_log_skipped_contexts_once(tuple(skipped_contexts))
-            return existing_contexts
+            return filter_by_allowed_node_pools(existing_contexts)
 
-        # If no allowed_contexts found, return all SSH contexts
-        return ssh_contexts
+        # If no all_node_pool_contexts found, return all SSH contexts
+        return filter_by_allowed_node_pools(ssh_contexts)
 
     @classmethod
     def _check_compute_credentials(
diff --git a/sky/dashboard/src/components/workspace-editor.jsx b/sky/dashboard/src/components/workspace-editor.jsx
index b02d4e074..b346425e0 100644
--- a/sky/dashboard/src/components/workspace-editor.jsx
+++ b/sky/dashboard/src/components/workspace-editor.jsx
@@ -51,7 +51,11 @@ const SuccessDisplay = ({ message }) => {
 };
 
 // Workspace configuration description component
-const WorkspaceConfigDescription = ({ workspaceName, config }) => {
+const WorkspaceConfigDescription = ({
+  workspaceName,
+  config,
+  enabledClouds = [],
+}) => {
   if (!config) return null;
 
   const isDefault = workspaceName === 'default';
@@ -67,9 +71,17 @@ const WorkspaceConfigDescription = ({ workspaceName, config }) => {
 
   const enabledDescriptions = [];
   const disabledClouds = [];
+  const configuredButNotEnabled = [];
+
+  // Convert enabledClouds to a set for faster lookup
+  const enabledCloudsSet = new Set(
+    enabledClouds.map((cloud) => cloud.toLowerCase())
+  );
 
   Object.entries(config).forEach(([cloud, cloudConfig]) => {
     const cloudName = CLOUD_CONONICATIONS[cloud.toLowerCase()];
+    const isActuallyEnabled = enabledCloudsSet.has(cloudName?.toLowerCase());
+
     if (cloudConfig?.disabled === true) {
       disabledClouds.push(cloudName);
     } else if (cloudConfig && Object.keys(cloudConfig).length > 0) {
@@ -79,18 +91,42 @@ const WorkspaceConfigDescription = ({ workspaceName, config }) => {
       } else if (cloud.toLowerCase() === 'aws' && cloudConfig.region) {
         detail = ` (Region: ${cloudConfig.region})`;
       }
-      enabledDescriptions.push(
-        <span key={`${cloud}-enabled`} className="block">
-          {cloudName}
-          {detail} is enabled.
-        </span>
-      );
+
+      if (isActuallyEnabled) {
+        enabledDescriptions.push(
+          <span key={`${cloud}-enabled`} className="block">
+            {cloudName}
+            {detail} is enabled.
+          </span>
+        );
+      } else {
+        configuredButNotEnabled.push(
+          <span
+            key={`${cloud}-configured-not-enabled`}
+            className="block text-amber-700"
+          >
+            {cloudName}
+            {detail} is configured but not currently available.
+          </span>
+        );
+      }
     } else {
-      enabledDescriptions.push(
-        <span key={`${cloud}-default-enabled`} className="block">
-          {cloudName} is enabled (using default settings).
-        </span>
-      );
+      if (isActuallyEnabled) {
+        enabledDescriptions.push(
+          <span key={`${cloud}-default-enabled`} className="block">
+            {cloudName} is enabled (using default settings).
+          </span>
+        );
+      } else {
+        configuredButNotEnabled.push(
+          <span
+            key={`${cloud}-default-not-enabled`}
+            className="block text-amber-700"
+          >
+            {cloudName} is configured but not currently available.
+          </span>
+        );
+      }
     }
   });
 
@@ -105,6 +141,7 @@ const WorkspaceConfigDescription = ({ workspaceName, config }) => {
     );
   }
   finalDescriptions.push(...enabledDescriptions);
+  finalDescriptions.push(...configuredButNotEnabled);
 
   if (finalDescriptions.length > 0) {
     return (
@@ -194,7 +231,7 @@ export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
         await Promise.all([
           getClusters(),
           getManagedJobs(),
-          getEnabledClouds(workspaceName),
+          getEnabledClouds(workspaceName, true),
         ]);
 
       // Filter clusters for this workspace
@@ -237,7 +274,7 @@ export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
         totalClusterCount: workspaceClusters.length,
         runningClusterCount: runningClusters.length,
         managedJobsCount: managedJobsCount,
-        clouds: Array.isArray(enabledClouds) ? enabledClouds.sort() : [],
+        clouds: Array.isArray(enabledClouds) ? enabledClouds : [],
       });
     } catch (err) {
       console.error('Failed to fetch workspace stats:', err);
@@ -571,6 +608,7 @@ export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
                         <WorkspaceConfigDescription
                           workspaceName={workspaceName}
                           config={originalConfig}
+                          enabledClouds={workspaceStats.clouds}
                         />
                       </div>
                     </div>
@@ -624,8 +662,10 @@ export function WorkspaceEditor({ workspaceName, isNewWorkspace = false }) {
                               {`${workspaceName || 'my-workspace'}:
   gcp:
     project_id: xxx
-  aws:
-    disabled: true`}
+    disabled: false
+  kubernetes:
+    allowed_contexts:
+      - context-1`}
                             </pre>
                           </div>
                         </div>
diff --git a/sky/dashboard/src/components/workspaces.jsx b/sky/dashboard/src/components/workspaces.jsx
index b3814b971..562294a61 100644
--- a/sky/dashboard/src/components/workspaces.jsx
+++ b/sky/dashboard/src/components/workspaces.jsx
@@ -311,7 +311,7 @@ export function Workspaces() {
 
       // Fetch enabled clouds for all workspaces
       const enabledCloudsArray = await Promise.all(
-        configuredWorkspaceNames.map((wsName) => getEnabledClouds(wsName))
+        configuredWorkspaceNames.map((wsName) => getEnabledClouds(wsName, true))
       );
       const enabledCloudsMap = Object.fromEntries(
         configuredWorkspaceNames.map((wsName, index) => [
@@ -390,7 +390,7 @@ export function Workspaces() {
         .map((ws) => ({
           ...ws,
           clouds: Array.isArray(enabledCloudsMap[ws.name])
-            ? enabledCloudsMap[ws.name].sort()
+            ? enabledCloudsMap[ws.name]
             : [],
         }))
         .sort((a, b) => a.name.localeCompare(b.name));
@@ -576,7 +576,13 @@ export function Workspaces() {
             </DialogDescription>
           </DialogHeader>
 
-          <ErrorDisplay error={deleteState.error} title="Deletion Failed" />
+          <ErrorDisplay
+            error={deleteState.error}
+            title="Deletion Failed"
+            onDismiss={() =>
+              setDeleteState((prev) => ({ ...prev, error: null }))
+            }
+          />
 
           <DialogFooter>
             <Button
diff --git a/sky/dashboard/src/data/connectors/workspaces.jsx b/sky/dashboard/src/data/connectors/workspaces.jsx
index b81e1caa0..efdcc5177 100644
--- a/sky/dashboard/src/data/connectors/workspaces.jsx
+++ b/sky/dashboard/src/data/connectors/workspaces.jsx
@@ -190,12 +190,19 @@ export async function getWorkspaces() {
   }
 }
 
-export async function getEnabledClouds(workspaceName = null) {
+export async function getEnabledClouds(workspaceName = null, expand = false) {
   try {
     // Step 1: Call the /enabled_clouds endpoint to schedule the task
     let url = `${ENDPOINT}/enabled_clouds`;
+    const params = new URLSearchParams();
     if (workspaceName) {
-      url += `?workspace=${encodeURIComponent(workspaceName)}`;
+      params.append('workspace', workspaceName);
+    }
+    if (expand) {
+      params.append('expand', 'true');
+    }
+    if (params.toString()) {
+      url += `?${params.toString()}`;
     }
 
     const scheduleResponse = await fetch(url, {
diff --git a/sky/skypilot_config.py b/sky/skypilot_config.py
index 29abd63ef..bb79bc0fd 100644
--- a/sky/skypilot_config.py
+++ b/sky/skypilot_config.py
@@ -299,6 +299,10 @@ def get_nested(keys: Tuple[str, ...],
 def get_workspace_cloud(cloud: str,
                         workspace: Optional[str] = None) -> config_utils.Config:
     """Returns the workspace config."""
+    # TODO(zhwu): Instead of just returning the workspace specific config, we
+    # should return the config that already merges the global config, so that
+    # the caller does not need to manually merge the global config with
+    # the workspace specific config.
     if workspace is None:
         workspace = get_active_workspace()
     clouds = get_nested(keys=(
diff --git a/sky/utils/schemas.py b/sky/utils/schemas.py
index c7f936a5e..48bc2dc75 100644
--- a/sky/utils/schemas.py
+++ b/sky/utils/schemas.py
@@ -1044,6 +1044,25 @@ def get_config_schema():
                 },
             }
         },
+        'ssh': {
+            'type': 'object',
+            'required': [],
+            'additionalProperties': False,
+            'properties': {
+                'allowed_node_pools': {
+                    'type': 'array',
+                    'items': {
+                        'type': 'string',
+                    },
+                },
+                'pod_config': {
+                    'type': 'object',
+                    'required': [],
+                    # Allow arbitrary keys since validating pod spec is hard
+                    'additionalProperties': True,
+                },
+            }
+        },
         'oci': {
             'type': 'object',
             'required': [],
@@ -1177,12 +1196,13 @@ def get_config_schema():
 
     allowed_workspace_cloud_names = list(
         service_catalog.ALL_CLOUDS) + ['cloudflare']
-    # Create pattern for non-GCP clouds (all clouds except gcp)
-    non_gcp_clouds = [
+    # Create pattern for not supported clouds, i.e.
+    # all clouds except gcp, kubernetes, ssh
+    not_supported_clouds = [
         cloud for cloud in allowed_workspace_cloud_names
-        if cloud.lower() != 'gcp'
+        if cloud.lower() not in ['gcp', 'kubernetes', 'ssh']
     ]
-    non_gcp_cloud_regex = '|'.join(non_gcp_clouds)
+    not_supported_cloud_regex = '|'.join(not_supported_clouds)
     workspaces_schema = {
         'type': 'object',
         'required': [],
@@ -1192,7 +1212,7 @@ def get_config_schema():
             'additionalProperties': False,
             'patternProperties': {
                 # Pattern for non-GCP clouds - only allows 'disabled' property
-                f'^({non_gcp_cloud_regex})$': {
+                f'^({not_supported_cloud_regex})$': {
                     'type': 'object',
                     'additionalProperties': False,
                     'properties': {
@@ -1217,6 +1237,38 @@ def get_config_schema():
                     },
                     'additionalProperties': False,
                 },
+                'ssh': {
+                    'type': 'object',
+                    'required': [],
+                    'properties': {
+                        'allowed_node_pools': {
+                            'type': 'array',
+                            'items': {
+                                'type': 'string',
+                            },
+                        },
+                        'disabled': {
+                            'type': 'boolean'
+                        },
+                    },
+                    'additionalProperties': False,
+                },
+                'kubernetes': {
+                    'type': 'object',
+                    'required': [],
+                    'properties': {
+                        'allowed_contexts': {
+                            'type': 'array',
+                            'items': {
+                                'type': 'string',
+                            },
+                        },
+                        'disabled': {
+                            'type': 'boolean'
+                        },
+                    },
+                    'additionalProperties': False,
+                },
             },
         },
     }


Test patch:
diff --git a/tests/unit_tests/test_sky/clouds/test_kubernetes.py b/tests/unit_tests/test_sky/clouds/test_kubernetes.py
new file mode 100644
index 000000000..ab291c955
--- /dev/null
+++ b/tests/unit_tests/test_sky/clouds/test_kubernetes.py
@@ -0,0 +1,262 @@
+"""Tests for Kubernetes cloud implementation."""
+
+import unittest
+from unittest import mock
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+import pytest
+
+from sky.clouds import kubernetes
+from sky.provision.kubernetes import utils as kubernetes_utils
+
+
+class TestKubernetesExistingAllowedContexts(unittest.TestCase):
+    """Test cases for Kubernetes.existing_allowed_contexts method."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        # Clear any cached results
+        kubernetes.Kubernetes._log_skipped_contexts_once.cache_clear()
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_no_contexts_available(self, mock_get_nested,
+                                   mock_get_workspace_cloud,
+                                   mock_get_all_contexts):
+        """Test when no Kubernetes contexts are available."""
+        mock_get_all_contexts.return_value = []
+
+        result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+        self.assertEqual(result, [])
+        mock_get_all_contexts.assert_called_once()
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_workspace_allowed_contexts_takes_precedence(
+            self, mock_get_nested, mock_get_workspace_cloud,
+            mock_get_all_contexts):
+        """Test that workspace allowed_contexts takes precedence over global."""
+        mock_get_all_contexts.return_value = ['ctx1', 'ctx2', 'ctx3']
+        mock_get_workspace_cloud.return_value.get.return_value = [
+            'ctx1', 'ctx2'
+        ]
+        mock_get_nested.return_value = ['ctx3']  # This should be ignored
+
+        result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+        self.assertEqual(result, ['ctx1', 'ctx2'])
+        mock_get_workspace_cloud.assert_called_once_with('kubernetes')
+        # get_nested should not be called when workspace config exists
+        mock_get_nested.assert_not_called()
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_global_allowed_contexts_when_no_workspace_config(
+            self, mock_get_nested, mock_get_workspace_cloud,
+            mock_get_all_contexts):
+        """Test using global allowed_contexts when workspace config is None."""
+        mock_get_all_contexts.return_value = ['ctx1', 'ctx2', 'ctx3']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = ['ctx2', 'ctx3']
+
+        result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+        self.assertEqual(result, ['ctx2', 'ctx3'])
+        mock_get_nested.assert_called_once_with(
+            ('kubernetes', 'allowed_contexts'), None)
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_excludes_ssh_contexts(self, mock_get_nested,
+                                   mock_get_workspace_cloud,
+                                   mock_get_all_contexts):
+        """Test that contexts starting with 'ssh-' are excluded."""
+        mock_get_all_contexts.return_value = [
+            'ctx1', 'ssh-cluster1', 'ctx2', 'ssh-cluster2'
+        ]
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = None
+
+        # Mock current context to be ctx1
+        with patch(
+                'sky.provision.kubernetes.utils.'
+                'get_current_kube_config_context_name',
+                return_value='ctx1'):
+            result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+        self.assertEqual(result, ['ctx1'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    @patch('sky.provision.kubernetes.utils.'
+           'get_current_kube_config_context_name')
+    @patch('sky.provision.kubernetes.utils.is_incluster_config_available')
+    @patch('sky.adaptors.kubernetes.in_cluster_context_name')
+    def test_fallback_to_current_context(self, mock_in_cluster_name,
+                                         mock_is_incluster_available,
+                                         mock_get_current_context,
+                                         mock_get_nested,
+                                         mock_get_workspace_cloud,
+                                         mock_get_all_contexts):
+        """Test fallback to current context when no allowed_contexts set."""
+        mock_get_all_contexts.return_value = ['ctx1', 'ctx2']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = None
+        mock_get_current_context.return_value = 'ctx1'
+        mock_is_incluster_available.return_value = False
+
+        result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+        self.assertEqual(result, ['ctx1'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    @patch('sky.provision.kubernetes.utils.'
+           'get_current_kube_config_context_name')
+    @patch('sky.provision.kubernetes.utils.is_incluster_config_available')
+    @patch('sky.adaptors.kubernetes.in_cluster_context_name')
+    def test_fallback_to_incluster_when_no_current_context(
+            self, mock_in_cluster_name, mock_is_incluster_available,
+            mock_get_current_context, mock_get_nested, mock_get_workspace_cloud,
+            mock_get_all_contexts):
+        """Test fallback to in-cluster context when no current context."""
+        # Include the in-cluster context in all_contexts
+        mock_get_all_contexts.return_value = ['ctx1', 'ctx2', 'in-cluster']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = None
+        mock_get_current_context.return_value = None
+        mock_is_incluster_available.return_value = True
+        mock_in_cluster_name.return_value = 'in-cluster'
+
+        result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+        self.assertEqual(result, ['in-cluster'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_filters_existing_contexts_only(self, mock_get_nested,
+                                            mock_get_workspace_cloud,
+                                            mock_get_all_contexts):
+        """Test that only existing contexts are returned."""
+        mock_get_all_contexts.return_value = ['ctx1', 'ctx2']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = ['ctx1', 'ctx2', 'nonexistent-ctx']
+
+        result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+        self.assertEqual(result, ['ctx1', 'ctx2'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_skips_ssh_contexts_in_allowed_list(self, mock_get_nested,
+                                                mock_get_workspace_cloud,
+                                                mock_get_all_contexts):
+        """Test that SSH contexts in allowed list are skipped."""
+        mock_get_all_contexts.return_value = ['ctx1', 'ctx2']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = ['ctx1', 'ssh-cluster', 'ctx2']
+
+        result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+        self.assertEqual(result, ['ctx1', 'ctx2'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_logs_skipped_contexts_when_not_silent(self, mock_get_nested,
+                                                   mock_get_workspace_cloud,
+                                                   mock_get_all_contexts):
+        """Test that skipped contexts are logged when not silent."""
+        mock_get_all_contexts.return_value = ['ctx1']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = ['ctx1', 'nonexistent-ctx']
+
+        with patch.object(kubernetes.Kubernetes,
+                          '_log_skipped_contexts_once') as mock_log:
+            result = kubernetes.Kubernetes.existing_allowed_contexts(
+                silent=False)
+
+            self.assertEqual(result, ['ctx1'])
+            mock_log.assert_called_once_with(('nonexistent-ctx',))
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_does_not_log_skipped_contexts_when_silent(self, mock_get_nested,
+                                                       mock_get_workspace_cloud,
+                                                       mock_get_all_contexts):
+        """Test that skipped contexts are not logged when silent=True."""
+        mock_get_all_contexts.return_value = ['ctx1']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = ['ctx1', 'nonexistent-ctx']
+
+        with patch.object(kubernetes.Kubernetes,
+                          '_log_skipped_contexts_once') as mock_log:
+            result = kubernetes.Kubernetes.existing_allowed_contexts(
+                silent=True)
+
+            self.assertEqual(result, ['ctx1'])
+            mock_log.assert_not_called()
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    @patch('sky.provision.kubernetes.utils.'
+           'get_current_kube_config_context_name')
+    @patch('sky.provision.kubernetes.utils.is_incluster_config_available')
+    def test_empty_result_when_no_contexts_found(
+            self, mock_is_incluster_available, mock_get_current_context,
+            mock_get_nested, mock_get_workspace_cloud, mock_get_all_contexts):
+        """Test empty result when no valid contexts are found."""
+        mock_get_all_contexts.return_value = ['ctx1']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = None
+        mock_get_current_context.return_value = None
+        mock_is_incluster_available.return_value = False
+
+        result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+        self.assertEqual(result, [])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_complex_scenario_with_mixed_contexts(self, mock_get_nested,
+                                                  mock_get_workspace_cloud,
+                                                  mock_get_all_contexts):
+        """Test complex scenario with various context types."""
+        # Available contexts include regular, SSH, and others
+        mock_get_all_contexts.return_value = [
+            'prod-cluster', 'ssh-dev-cluster', 'staging-cluster',
+            'ssh-test-cluster', 'local-cluster'
+        ]
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        # Allowed contexts include existing, non-existing, and SSH contexts
+        mock_get_nested.return_value = [
+            'prod-cluster', 'nonexistent-cluster', 'ssh-dev-cluster',
+            'staging-cluster', 'ssh-nonexistent'
+        ]
+
+        with patch.object(kubernetes.Kubernetes,
+                          '_log_skipped_contexts_once') as mock_log:
+            result = kubernetes.Kubernetes.existing_allowed_contexts()
+
+            # Should only return non-SSH existing contexts
+            self.assertEqual(sorted(result),
+                             ['prod-cluster', 'staging-cluster'])
+            # Should log the nonexistent context (SSH contexts are skipped)
+            mock_log.assert_called_once_with(('nonexistent-cluster',))
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/unit_tests/test_sky/clouds/test_ssh.py b/tests/unit_tests/test_sky/clouds/test_ssh.py
new file mode 100644
index 000000000..4d8e9d9a6
--- /dev/null
+++ b/tests/unit_tests/test_sky/clouds/test_ssh.py
@@ -0,0 +1,307 @@
+"""Tests for SSH cloud implementation."""
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+import pytest
+import yaml
+
+from sky.clouds import ssh
+from sky.provision.kubernetes import utils as kubernetes_utils
+
+
+class TestSSHExistingAllowedContexts(unittest.TestCase):
+    """Test cases for SSH.existing_allowed_contexts method."""
+
+    def setUp(self):
+        """Set up test fixtures."""
+        # Clear any cached results
+        ssh.SSH._ssh_log_skipped_contexts_once.cache_clear()
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    def test_no_contexts_available(self, mock_get_all_contexts):
+        """Test when no Kubernetes contexts are available."""
+        mock_get_all_contexts.return_value = []
+
+        result = ssh.SSH.existing_allowed_contexts()
+
+        self.assertEqual(result, [])
+        mock_get_all_contexts.assert_called_once()
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_filters_only_ssh_contexts(self, mock_get_nested,
+                                       mock_get_workspace_cloud,
+                                       mock_get_all_contexts):
+        """Test that only SSH contexts (starting with 'ssh-') are returned."""
+        mock_get_all_contexts.return_value = [
+            'regular-ctx1', 'ssh-cluster1', 'regular-ctx2', 'ssh-cluster2'
+        ]
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = None
+
+        with patch.object(ssh.SSH,
+                          'get_ssh_node_pool_contexts',
+                          return_value=[]):
+            result = ssh.SSH.existing_allowed_contexts()
+
+            # Should return all SSH contexts when no node pools file
+            self.assertEqual(sorted(result), ['ssh-cluster1', 'ssh-cluster2'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_workspace_allowed_node_pools_takes_precedence(
+            self, mock_get_nested, mock_get_workspace_cloud,
+            mock_get_all_contexts):
+        """Test that workspace allowed_node_pools takes precedence."""
+        mock_get_all_contexts.return_value = [
+            'ssh-cluster1', 'ssh-cluster2', 'ssh-cluster3'
+        ]
+        mock_get_workspace_cloud.return_value.get.return_value = [
+            'cluster1', 'cluster2'
+        ]
+        mock_get_nested.return_value = ['cluster3']  # Should be ignored
+
+        with patch.object(
+                ssh.SSH,
+                'get_ssh_node_pool_contexts',
+                return_value=['ssh-cluster1', 'ssh-cluster2', 'ssh-cluster3']):
+            result = ssh.SSH.existing_allowed_contexts()
+
+            # Should only return contexts matching allowed node pools
+            self.assertEqual(sorted(result), ['ssh-cluster1', 'ssh-cluster2'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_global_allowed_node_pools_when_no_workspace_config(
+            self, mock_get_nested, mock_get_workspace_cloud,
+            mock_get_all_contexts):
+        """Test using global allowed_node_pools when workspace config is None."""
+        mock_get_all_contexts.return_value = [
+            'ssh-cluster1', 'ssh-cluster2', 'ssh-cluster3'
+        ]
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = ['cluster2', 'cluster3']
+
+        with patch.object(
+                ssh.SSH,
+                'get_ssh_node_pool_contexts',
+                return_value=['ssh-cluster1', 'ssh-cluster2', 'ssh-cluster3']):
+            result = ssh.SSH.existing_allowed_contexts()
+
+            self.assertEqual(sorted(result), ['ssh-cluster2', 'ssh-cluster3'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_node_pool_contexts_filter_existing_contexts(
+            self, mock_get_nested, mock_get_workspace_cloud,
+            mock_get_all_contexts):
+        """Test that node pool contexts are filtered by existing contexts."""
+        mock_get_all_contexts.return_value = ['ssh-cluster1', 'ssh-cluster3']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = None
+
+        # Node pools file has more contexts than actually exist
+        with patch.object(
+                ssh.SSH,
+                'get_ssh_node_pool_contexts',
+                return_value=['ssh-cluster1', 'ssh-cluster2', 'ssh-cluster3']):
+            result = ssh.SSH.existing_allowed_contexts()
+
+            # Should only return contexts that actually exist
+            self.assertEqual(sorted(result), ['ssh-cluster1', 'ssh-cluster3'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_logs_skipped_contexts_when_not_silent(self, mock_get_nested,
+                                                   mock_get_workspace_cloud,
+                                                   mock_get_all_contexts):
+        """Test that skipped contexts are logged when not silent."""
+        mock_get_all_contexts.return_value = ['ssh-cluster1']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = None
+
+        with patch.object(ssh.SSH,
+                          'get_ssh_node_pool_contexts',
+                          return_value=['ssh-cluster1', 'ssh-nonexistent']):
+            with patch.object(ssh.SSH,
+                              '_ssh_log_skipped_contexts_once') as mock_log:
+                result = ssh.SSH.existing_allowed_contexts(silent=False)
+
+                self.assertEqual(result, ['ssh-cluster1'])
+                mock_log.assert_called_once_with(('ssh-nonexistent',))
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_does_not_log_skipped_contexts_when_silent(self, mock_get_nested,
+                                                       mock_get_workspace_cloud,
+                                                       mock_get_all_contexts):
+        """Test that skipped contexts are not logged when silent=True."""
+        mock_get_all_contexts.return_value = ['ssh-cluster1']
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = None
+
+        with patch.object(ssh.SSH,
+                          'get_ssh_node_pool_contexts',
+                          return_value=['ssh-cluster1', 'ssh-nonexistent']):
+            with patch.object(ssh.SSH,
+                              '_ssh_log_skipped_contexts_once') as mock_log:
+                result = ssh.SSH.existing_allowed_contexts(silent=True)
+
+                self.assertEqual(result, ['ssh-cluster1'])
+                mock_log.assert_not_called()
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_fallback_to_all_ssh_contexts_when_no_node_pools(
+            self, mock_get_nested, mock_get_workspace_cloud,
+            mock_get_all_contexts):
+        """Test fallback to all SSH contexts when no node pools file."""
+        mock_get_all_contexts.return_value = [
+            'regular-ctx', 'ssh-cluster1', 'ssh-cluster2'
+        ]
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = None
+
+        with patch.object(ssh.SSH,
+                          'get_ssh_node_pool_contexts',
+                          return_value=[]):
+            result = ssh.SSH.existing_allowed_contexts()
+
+            # Should return all SSH contexts when no node pools
+            self.assertEqual(sorted(result), ['ssh-cluster1', 'ssh-cluster2'])
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_allowed_node_pools_filtering_with_fallback(
+            self, mock_get_nested, mock_get_workspace_cloud,
+            mock_get_all_contexts):
+        """Test allowed_node_pools filtering with fallback scenario."""
+        mock_get_all_contexts.return_value = [
+            'ssh-cluster1', 'ssh-cluster2', 'ssh-cluster3'
+        ]
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        mock_get_nested.return_value = ['cluster1', 'cluster3']
+
+        # No node pools file, so fallback to all SSH contexts
+        with patch.object(ssh.SSH,
+                          'get_ssh_node_pool_contexts',
+                          return_value=[]):
+            result = ssh.SSH.existing_allowed_contexts()
+
+            # Should filter by allowed_node_pools even in fallback
+            self.assertEqual(sorted(result), ['ssh-cluster1', 'ssh-cluster3'])
+
+    def test_get_ssh_node_pool_contexts_with_valid_file(self):
+        """Test get_ssh_node_pool_contexts with a valid YAML file."""
+        test_config = {
+            'cluster1': {
+                'some': 'config'
+            },
+            'cluster2': {
+                'other': 'config'
+            },
+            'cluster3': {
+                'more': 'config'
+            }
+        }
+
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml',
+                                         delete=False) as f:
+            yaml.dump(test_config, f)
+            temp_path = f.name
+
+        try:
+            with patch('sky.clouds.ssh.SSH_NODE_POOLS_PATH', temp_path):
+                result = ssh.SSH.get_ssh_node_pool_contexts()
+
+                expected = ['ssh-cluster1', 'ssh-cluster2', 'ssh-cluster3']
+                self.assertEqual(sorted(result), sorted(expected))
+        finally:
+            os.unlink(temp_path)
+
+    def test_get_ssh_node_pool_contexts_with_nonexistent_file(self):
+        """Test get_ssh_node_pool_contexts with nonexistent file."""
+        with patch('sky.clouds.ssh.SSH_NODE_POOLS_PATH',
+                   '/nonexistent/path/file.yaml'):
+            result = ssh.SSH.get_ssh_node_pool_contexts()
+
+            self.assertEqual(result, [])
+
+    def test_get_ssh_node_pool_contexts_with_empty_file(self):
+        """Test get_ssh_node_pool_contexts with empty YAML file."""
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml',
+                                         delete=False) as f:
+            f.write('')  # Empty file
+            temp_path = f.name
+
+        try:
+            with patch('sky.clouds.ssh.SSH_NODE_POOLS_PATH', temp_path):
+                result = ssh.SSH.get_ssh_node_pool_contexts()
+
+                self.assertEqual(result, [])
+        finally:
+            os.unlink(temp_path)
+
+    def test_get_ssh_node_pool_contexts_with_invalid_yaml(self):
+        """Test get_ssh_node_pool_contexts with invalid YAML file."""
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml',
+                                         delete=False) as f:
+            f.write('invalid: yaml: content: [')  # Invalid YAML
+            temp_path = f.name
+
+        try:
+            with patch('sky.clouds.ssh.SSH_NODE_POOLS_PATH', temp_path):
+                result = ssh.SSH.get_ssh_node_pool_contexts()
+
+                # Should return empty list on error
+                self.assertEqual(result, [])
+        finally:
+            os.unlink(temp_path)
+
+    @patch('sky.provision.kubernetes.utils.get_all_kube_context_names')
+    @patch('sky.skypilot_config.get_workspace_cloud')
+    @patch('sky.skypilot_config.get_nested')
+    def test_complex_scenario_with_mixed_filtering(self, mock_get_nested,
+                                                   mock_get_workspace_cloud,
+                                                   mock_get_all_contexts):
+        """Test complex scenario with node pools and allowed filtering."""
+        # Available contexts include regular and SSH contexts
+        mock_get_all_contexts.return_value = [
+            'regular-ctx', 'ssh-prod', 'ssh-dev', 'ssh-staging', 'ssh-test'
+        ]
+        mock_get_workspace_cloud.return_value.get.return_value = None
+        # Only allow specific node pools - note: lstrip('ssh-') behavior
+        # 'ssh-prod'.lstrip('ssh-') -> 'prod'
+        # 'ssh-staging'.lstrip('ssh-') -> 'taging' (not 'staging'!)
+        mock_get_nested.return_value = ['prod', 'taging', 'nonexistent']
+
+        # Node pools file has some contexts
+        with patch.object(ssh.SSH,
+                          'get_ssh_node_pool_contexts',
+                          return_value=['ssh-prod', 'ssh-dev', 'ssh-staging']):
+            with patch.object(ssh.SSH,
+                              '_ssh_log_skipped_contexts_once') as mock_log:
+                result = ssh.SSH.existing_allowed_contexts()
+
+                # Should return intersection of node pools, existing, and allowed
+                # Only ssh-prod and ssh-staging should match (prod and taging)
+                self.assertEqual(sorted(result), ['ssh-prod', 'ssh-staging'])
+                # Should log empty tuple since all node pool contexts exist
+                mock_log.assert_called_once_with(())
+
+
+if __name__ == '__main__':
+    unittest.main()