internal: testserver: reorg internals, fix error code (#3591)

## Changes
- Move jobs-related code to jobs.go.
- In jobs.go, update parsing error to return 400 instead of 500.
- Move secrets ACLs related code to secret_acls.go.

## Why
Match structure of other resources in testserver.

Author: denik

libs/testserver/fake_workspace.go

@@ -6,8 +6,6 @@ import (
 	"fmt"
 	"path"
 	"path/filepath"
-	"sort"
-	"strconv"
 	"strings"
 	"sync"
 
@@ -297,185 +295,6 @@ func (s *FakeWorkspace) WorkspaceFilesExportFile(path string) []byte {
 	return s.files[path].Data
 }
 
-func (s *FakeWorkspace) JobsCreate(request jobs.CreateJob) Response {
-	defer s.LockUnlock()()
-
-	jobId := s.nextJobId
-	s.nextJobId++
-
-	jobSettings := jobs.JobSettings{}
-	err := jsonConvert(request, &jobSettings)
-	if err != nil {
-		return Response{
-			StatusCode: 400,
-			Body:       fmt.Sprintf("Cannot convert request to jobSettings: %s", err),
-		}
-	}
-
-	s.Jobs[jobId] = jobs.Job{
-		JobId:    jobId,
-		Settings: &jobSettings,
-	}
-
-	return Response{
-		Body: jobs.CreateResponse{JobId: jobId},
-	}
-}
-
-func (s *FakeWorkspace) JobsReset(request jobs.ResetJob) Response {
-	defer s.LockUnlock()()
-
-	jobId := request.JobId
-
-	_, ok := s.Jobs[request.JobId]
-	if !ok {
-		return Response{
-			StatusCode: 403,
-			Body:       "{}",
-		}
-	}
-
-	s.Jobs[jobId] = jobs.Job{
-		JobId:    jobId,
-		Settings: &request.NewSettings,
-	}
-
-	return Response{
-		Body: "",
-	}
-}
-
-func (s *FakeWorkspace) JobsGet(jobId string) Response {
-	id := jobId
-
-	jobIdInt, err := strconv.ParseInt(id, 10, 64)
-	if err != nil {
-		return Response{
-			StatusCode: 400,
-			Body:       fmt.Sprintf("Failed to parse job id: %s: %v", err, id),
-		}
-	}
-
-	defer s.LockUnlock()()
-
-	job, ok := s.Jobs[jobIdInt]
-	if !ok {
-		return Response{
-			StatusCode: 404,
-		}
-	}
-
-	job = setSourceIfNotSet(job)
-	return Response{
-		Body: job,
-	}
-}
-
-func (s *FakeWorkspace) JobsRunNow(jobId int64) Response {
-	defer s.LockUnlock()()
-
-	_, ok := s.Jobs[jobId]
-	if !ok {
-		return Response{
-			StatusCode: 404,
-		}
-	}
-
-	runId := s.nextJobRunId
-	s.nextJobRunId++
-	s.JobRuns[runId] = jobs.Run{
-		RunId: runId,
-		State: &jobs.RunState{
-			LifeCycleState: jobs.RunLifeCycleStateRunning,
-		},
-		RunPageUrl: fmt.Sprintf("%s/job/run/%d", s.url, runId),
-		RunType:    jobs.RunTypeJobRun,
-		RunName:    "run-name",
-	}
-
-	return Response{
-		Body: jobs.RunNowResponse{
-			RunId: runId,
-		},
-	}
-}
-
-func (s *FakeWorkspace) JobsGetRun(runId int64) Response {
-	defer s.LockUnlock()()
-
-	run, ok := s.JobRuns[runId]
-	if !ok {
-		return Response{
-			StatusCode: 404,
-		}
-	}
-
-	// Mark the run as terminated.
-	run.State.LifeCycleState = jobs.RunLifeCycleStateTerminated
-	return Response{
-		Body: run,
-	}
-}
-
-func (s *FakeWorkspace) JobsList() Response {
-	defer s.LockUnlock()()
-
-	list := make([]jobs.BaseJob, 0, len(s.Jobs))
-	for _, job := range s.Jobs {
-		job = setSourceIfNotSet(job)
-		baseJob := jobs.BaseJob{}
-		err := jsonConvert(job, &baseJob)
-		if err != nil {
-			return Response{
-				StatusCode: 400,
-				Body:       fmt.Sprintf("failed to convert job to base job: %s", err),
-			}
-		}
-
-		list = append(list, baseJob)
-	}
-
-	// sort to have less non-determinism in tests
-	sort.Slice(list, func(i, j int) bool {
-		return list[i].JobId < list[j].JobId
-	})
-
-	return Response{
-		Body: jobs.ListJobsResponse{
-			Jobs: list,
-		},
-	}
-}
-
-func setSourceIfNotSet(job jobs.Job) jobs.Job {
-	// Setting the source field in the output of the Jobs List API same way as backend does
-	if job.Settings != nil {
-		source := "WORKSPACE"
-		if job.Settings.GitSource != nil {
-			source = "GIT"
-		}
-		for _, task := range job.Settings.Tasks {
-			if task.NotebookTask != nil {
-				if task.NotebookTask.Source == "" {
-					task.NotebookTask.Source = jobs.Source(source)
-				}
-				if task.DbtTask != nil {
-					if task.DbtTask.Source == "" {
-						task.DbtTask.Source = jobs.Source(source)
-					}
-				}
-				if task.SparkPythonTask != nil {
-					if task.SparkPythonTask.Source == "" {
-						task.SparkPythonTask.Source = jobs.Source(source)
-					}
-				}
-			}
-		}
-	}
-
-	return job
-}
-
 // jsonConvert saves input to a value pointed by output
 func jsonConvert(input, output any) error {
 	writer := new(bytes.Buffer)

libs/testserver/handlers.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
-	"strconv"
 
 	"github.com/databricks/databricks-sdk-go/service/catalog"
 	"github.com/databricks/databricks-sdk-go/service/compute"
@@ -180,77 +179,42 @@ func AddDefaultHandlers(server *Server) {
 	})
 
 	server.Handle("POST", "/api/2.2/jobs/create", func(req Request) any {
-		var request jobs.CreateJob
-		if err := json.Unmarshal(req.Body, &request); err != nil {
-			return Response{
-				Body:       fmt.Sprintf("internal error: %s", err),
-				StatusCode: 500,
-			}
-		}
-
-		return req.Workspace.JobsCreate(request)
+		return req.Workspace.JobsCreate(req)
 	})
 
 	server.Handle("POST", "/api/2.2/jobs/delete", func(req Request) any {
 		var request jobs.DeleteJob
 		if err := json.Unmarshal(req.Body, &request); err != nil {
 			return Response{
-				Body:       fmt.Sprintf("internal error: %s", err),
-				StatusCode: 500,
+				StatusCode: 400,
+				Body:       fmt.Sprintf("request parsing error: %s", err),
 			}
 		}
 		return MapDelete(req.Workspace, req.Workspace.Jobs, request.JobId)
 	})
 
 	server.Handle("POST", "/api/2.2/jobs/reset", func(req Request) any {
-		var request jobs.ResetJob
-		if err := json.Unmarshal(req.Body, &request); err != nil {
-			return Response{
-				Body:       fmt.Sprintf("internal error: %s", err),
-				StatusCode: 500,
-			}
-		}
-
-		return req.Workspace.JobsReset(request)
+		return req.Workspace.JobsReset(req)
 	})
 
 	server.Handle("GET", "/api/2.0/jobs/get", func(req Request) any {
-		jobId := req.URL.Query().Get("job_id")
-		return req.Workspace.JobsGet(jobId)
+		return req.Workspace.JobsGet(req)
 	})
 
 	server.Handle("GET", "/api/2.2/jobs/get", func(req Request) any {
-		jobId := req.URL.Query().Get("job_id")
-		return req.Workspace.JobsGet(jobId)
+		return req.Workspace.JobsGet(req)
 	})
 
 	server.Handle("GET", "/api/2.2/jobs/list", func(req Request) any {
 		return req.Workspace.JobsList()
 	})
 
 	server.Handle("POST", "/api/2.2/jobs/run-now", func(req Request) any {
-		var request jobs.RunNow
-		if err := json.Unmarshal(req.Body, &request); err != nil {
-			return Response{
-				Body:       fmt.Sprintf("internal error: %s", err),
-				StatusCode: 500,
-			}
-		}
-
-		return req.Workspace.JobsRunNow(request.JobId)
+		return req.Workspace.JobsRunNow(req)
 	})
 
 	server.Handle("GET", "/api/2.2/jobs/runs/get", func(req Request) any {
-		runId := req.URL.Query().Get("run_id")
-		runIdInt, err := strconv.ParseInt(runId, 10, 64)
-		if err != nil {
-			return Response{
-				Body:       fmt.Sprintf("internal error: %s", err),
-				StatusCode: 500,
-			}
-		}
-
-		return req.Workspace.JobsGetRun(runIdInt)
+		return req.Workspace.JobsGetRun(req)
 	})
 
 	server.Handle("GET", "/api/2.2/jobs/runs/list", func(req Request) any {
@@ -395,6 +359,8 @@ func AddDefaultHandlers(server *Server) {
 		return req.Workspace.VolumesCreate(req)
 	})
 
+	// Repos:
+
 	server.Handle("POST", "/api/2.0/repos", func(req Request) any {
 		return req.Workspace.ReposCreate(req)
 	})
@@ -420,6 +386,7 @@ func AddDefaultHandlers(server *Server) {
 	})
 
 	// SQL Warehouses:
+
 	server.Handle("GET", "/api/2.0/sql/warehouses/{warehouse_id}", func(req Request) any {
 		return MapGet(req.Workspace, req.Workspace.SqlWarehouses, req.Vars["warehouse_id"])
 	})
@@ -444,60 +411,24 @@ func AddDefaultHandlers(server *Server) {
 		return req.Workspace.SqlDataSourcesList(req)
 	})
 
+	// Secrets ACLs:
+
 	server.Handle("GET", "/api/2.0/secrets/acls/get", func(req Request) any {
-		defer req.Workspace.LockUnlock()()
-
-		scope := req.URL.Query().Get("scope")
-		principal := req.URL.Query().Get("principal")
-		scopeAcls := req.Workspace.Acls[scope]
-		for _, acl := range scopeAcls {
-			if acl.Principal == principal {
-				return acl
-			}
-		}
-		return Response{StatusCode: 404}
+		return req.Workspace.SecretsAclsGet(req)
 	})
 
 	server.Handle("GET", "/api/2.0/secrets/acls/list", func(req Request) any {
-		return MapGet(req.Workspace, req.Workspace.Acls, req.Vars["scope"])
+		return MapGet(req.Workspace, req.Workspace.Acls, req.URL.Query().Get("scope"))
 	})
 
 	server.Handle("POST", "/api/2.0/secrets/acls/put", func(req Request) any {
-		defer req.Workspace.LockUnlock()()
-
-		var request workspace.PutAcl
-		if err := json.Unmarshal(req.Body, &request); err != nil {
-			return Response{
-				Body:       fmt.Sprintf("internal error: %s", err),
-				StatusCode: 500,
-			}
-		}
-		req.Workspace.Acls[request.Scope] = append(req.Workspace.Acls[request.Scope], workspace.AclItem{
-			Principal:  request.Principal,
-			Permission: request.Permission,
-		})
-		return ""
+		return req.Workspace.SecretsAclsPut(req)
 	})
 
 	server.Handle("POST", "/api/2.0/secrets/acls/delete", func(req Request) any {
-		defer req.Workspace.LockUnlock()()
-
-		var request workspace.DeleteAcl
-		if err := json.Unmarshal(req.Body, &request); err != nil {
-			return Response{
-				Body:       fmt.Sprintf("internal error: %s", err),
-				StatusCode: 500,
-			}
-		}
-		scopeAcls := req.Workspace.Acls[request.Scope]
-		for i, acl := range scopeAcls {
-			if acl.Principal == request.Principal {
-				req.Workspace.Acls[request.Scope] = append(scopeAcls[:i], scopeAcls[i+1:]...)
-				return ""
-			}
-		}
-		return Response{StatusCode: 404}
+		return req.Workspace.SecretsAclsDelete(req)
 	})
+	// Database Instances:
 
 	server.Handle("POST", "/api/2.0/database/instances", func(req Request) any {
 		return req.Workspace.DatabaseInstanceCreate(req)
@@ -519,6 +450,8 @@ func AddDefaultHandlers(server *Server) {
 		return DatabaseInstanceMapDelete(req)
 	})
 
+	// Database Catalogs:
+
 	server.Handle("POST", "/api/2.0/database/catalogs", func(req Request) any {
 		return req.Workspace.DatabaseCatalogCreate(req)
 	})
@@ -535,6 +468,8 @@ func AddDefaultHandlers(server *Server) {
 		return MapDelete(req.Workspace, req.Workspace.DatabaseCatalogs, req.Vars["name"])
 	})
 
+	// Synced Database Tables:
+
 	server.Handle("POST", "/api/2.0/database/synced_tables", func(req Request) any {
 		return req.Workspace.SyncedDatabaseTableCreate(req)
 	})