acc: run ssh/connect-serverless-gpu locally

acceptance/ssh/connect-serverless-gpu/out.stdout.txt

@@ -1 +0,0 @@
-Connection successful

acceptance/ssh/connect-serverless-gpu/output.txt

@@ -0,0 +1,40 @@
+
+>>> print_requests.py //api/2.2/jobs/runs/submit
+{
+  "method": "POST",
+  "path": "/api/2.2/jobs/runs/submit",
+  "body": {
+    "environments": [
+      {
+        "environment_key": "ssh_tunnel_serverless",
+        "spec": {
+          "environment_version": "4"
+        }
+      }
+    ],
+    "run_name": "ssh-server-bootstrap-serverless-gpu-test",
+    "tasks": [
+      {
+        "compute": {
+          "hardware_accelerator": "GPU_1xA10"
+        },
+        "environment_key": "ssh_tunnel_serverless",
+        "notebook_task": {
+          "base_parameters": {
+            "authorizedKeySecretName": "client-public-key",
+            "maxClients": "10",
+            "secretScopeName": "[USERNAME]-serverless-gpu-test-ssh-tunnel-keys",
+            "serverless": "true",
+            "sessionId": "serverless-gpu-test",
+            "shutdownDelay": "10m0s",
+            "version": "[CLI_VERSION]"
+          },
+          "notebook_path": "/Workspace/Users/[USERNAME]/.databricks/ssh-tunnel/[CLI_VERSION]/serverless-gpu-test/ssh-server-bootstrap"
+        },
+        "task_key": "start_ssh_server",
+        "timeout_seconds": 86400
+      }
+    ],
+    "timeout_seconds": 86400
+  }
+}

acceptance/ssh/connect-serverless-gpu/script

@@ -1,6 +1,22 @@
+if [ -z "${CLOUD_ENV:-}" ]; then
+  # Local runs have no release artifacts and never execute the uploaded
+  # binaries, so stand in empty archives (dev name omits the version).
+  mkdir -p releases
+  : >releases/databricks_cli_linux_amd64.zip
+  : >releases/databricks_cli_linux_arm64.zip
+  CLI_RELEASES_DIR="$PWD/releases"
+fi
+
 errcode $CLI ssh connect --name serverless-gpu-test --accelerator=GPU_1xA10 --releases-dir=$CLI_RELEASES_DIR -- "echo 'Connection successful'" >out.stdout.txt 2>LOG.stderr
 
-if ! grep -q "Connection successful" out.stdout.txt; then
-  run_id=$(cat LOG.stderr | grep -o "Job submitted successfully with run ID: [0-9]*" | grep -o "[0-9]*$")
-  trace $CLI jobs get-run "$run_id" > LOG.job
+if [ -n "${CLOUD_ENV:-}" ]; then
+  # On cloud the connection runs the remote command; dump the run on failure.
+  if ! grep -q "Connection successful" out.stdout.txt; then
+    run_id=$(cat LOG.stderr | grep -o "Job submitted successfully with run ID: [0-9]*" | grep -o "[0-9]*$")
+    trace $CLI jobs get-run "$run_id" > LOG.job
+  fi
+else
+  # Locally the handshake can't complete (no compute); just assert the CLI
+  # submitted the right serverless GPU bootstrap job.
+  trace print_requests.py //api/2.2/jobs/runs/submit
 fi

acceptance/ssh/connect-serverless-gpu/test.toml

@@ -1,9 +1,17 @@
-Local = false
+Local = true
 Cloud = false
 
 # Serverless GPU is only available in newer environments
 RequiresUnityCatalog = true
 
+# Assert the serverless GPU bootstrap job the CLI submits on a local run.
+RecordRequests = true
+
+# Local runs stage stand-in release archives here; not part of the golden output.
+Ignore = [
+  "releases",
+]
+
 # Serverless GPU is not available in GCP yet
 [CloudEnvs]
   gcp = false

libs/testserver/handlers.go

@@ -105,6 +105,15 @@ func AddDefaultHandlers(server *Server) {
 		path := req.URL.Query().Get("path")
 		data := req.Workspace.WorkspaceExport(path)
 
+		// A missing object returns 404, matching the real API; returning the nil
+		// body otherwise trips the response normalizer.
+		if data == nil {
+			return Response{
+				StatusCode: 404,
+				Body:       map[string]string{"message": fmt.Sprintf("Path (%s) doesn't exist.", path)},
+			}
+		}
+
 		// The filer reads the raw object body via ?direct_download=true, while
 		// the SDK's Workspace.Export (used by `databricks workspace export`)
 		// requests JSON and expects the base64-encoded content field.
@@ -730,6 +739,20 @@ func AddDefaultHandlers(server *Server) {
 		return req.Workspace.SecretsGet(req)
 	})
 
+	server.Handle("GET", "/api/2.0/secrets/list", func(req Request) any {
+		return req.Workspace.SecretsList(req)
+	})
+
+	// SSH tunnel server behind the driver proxy: /metadata returns the remote
+	// login user, /logs is a best-effort error tail fetched on failure.
+	server.Handle("GET", "/driver-proxy-api/o/{workspace_id}/{cluster_id}/{port}/metadata", func(req Request) any {
+		return Response{Body: sshTunnelRemoteUser}
+	})
+
+	server.Handle("GET", "/driver-proxy-api/o/{workspace_id}/{cluster_id}/{port}/logs", func(req Request) any {
+		return Response{Body: ""}
+	})
+
 	// Secrets ACLs:
 	server.Handle("GET", "/api/2.0/secrets/acls/get", func(req Request) any {
 		return req.Workspace.SecretsAclsGet(req)

libs/testserver/jobs.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/databricks/databricks-sdk-go/service/compute"
 	"github.com/databricks/databricks-sdk-go/service/jobs"
+	"github.com/databricks/databricks-sdk-go/service/workspace"
 )
 
 const missingJobGitProviderMessage = "git_source.git_provider must be one of: github,gitlab,bitbucketcloud,gitlabenterpriseedition,bitbucketserver,azuredevopsservices,githubenterprise,awscodecommit"
@@ -519,9 +520,45 @@ func (s *FakeWorkspace) JobsSubmit(req Request) Response {
 		Tasks:      tasks,
 	}
 
+	// No tunnel server runs locally, so synthesize the metadata.json it would
+	// publish; `ssh connect` polls for it before connecting.
+	if strings.HasPrefix(runName, sshTunnelBootstrapRunPrefix) {
+		s.writeSSHTunnelMetadata(request)
+	}
+
 	return Response{Body: jobs.SubmitRunResponse{RunId: runId}}
 }
 
+const (
+	sshTunnelBootstrapRunPrefix = "ssh-server-bootstrap-"
+	sshTunnelBootstrapNotebook  = "ssh-server-bootstrap"
+	sshTunnelServerPort         = 7772
+	sshTunnelClusterID          = "1234-567890-serverless"
+	sshTunnelRemoteUser         = "spark"
+)
+
+// writeSSHTunnelMetadata publishes the metadata.json a real tunnel server would
+// write next to the bootstrap notebook. Callers must hold the workspace lock.
+func (s *FakeWorkspace) writeSSHTunnelMetadata(request jobs.SubmitRun) {
+	for _, t := range request.Tasks {
+		if t.NotebookTask == nil {
+			continue
+		}
+		metadataPath := strings.TrimSuffix(t.NotebookTask.NotebookPath, sshTunnelBootstrapNotebook) + "metadata.json"
+		metadata, err := json.Marshal(map[string]any{
+			"port":       sshTunnelServerPort,
+			"cluster_id": sshTunnelClusterID,
+		})
+		if err != nil {
+			continue
+		}
+		s.files[metadataPath] = FileEntry{
+			Info: workspace.ObjectInfo{ObjectType: "FILE", Path: metadataPath},
+			Data: metadata,
+		}
+	}
+}
+
 // executePythonWheelTask runs a python wheel task locally using uv.
 // For tasks using existing_cluster_id, the venv is cached per cluster to match
 // cloud behavior where libraries are cached on running clusters.

libs/testserver/secrets.go

@@ -4,6 +4,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"slices"
 
 	"github.com/databricks/databricks-sdk-go/service/workspace"
 )
@@ -40,6 +41,38 @@ func (s *FakeWorkspace) SecretsPut(req Request) Response {
 	return Response{}
 }
 
+// SecretsList models GET /api/2.0/secrets/list. A missing scope must return
+// RESOURCE_DOES_NOT_EXIST so callers can branch on apierr.ErrResourceDoesNotExist.
+func (s *FakeWorkspace) SecretsList(req Request) Response {
+	defer s.LockUnlock()()
+
+	scope := req.URL.Query().Get("scope")
+
+	if _, exists := s.SecretScopes[scope]; !exists {
+		return Response{
+			StatusCode: 404,
+			Body:       map[string]string{"error_code": "RESOURCE_DOES_NOT_EXIST", "message": fmt.Sprintf("Scope %s does not exist", scope)},
+		}
+	}
+
+	keys := make([]string, 0, len(s.Secrets[scope]))
+	for key := range s.Secrets[scope] {
+		keys = append(keys, key)
+	}
+	slices.Sort(keys)
+
+	secrets := make([]workspace.SecretMetadata, 0, len(keys))
+	for _, key := range keys {
+		secrets = append(secrets, workspace.SecretMetadata{Key: key})
+	}
+
+	return Response{
+		Body: workspace.ListSecretsResponse{
+			Secrets: secrets,
+		},
+	}
+}
+
 func (s *FakeWorkspace) SecretsGet(req Request) Response {
 	defer s.LockUnlock()()