acc: support pipeline update (#2934)

## Changes
- Support editing pipelines in testserver.
- Rewrite pipeline code to match best practice in testserver (separate
file in libs/testserver and separate block in handlers.go).

## Why
Enables new acceptance tests, I'm using this in terraform-removal branch
#2926

## Tests
New edit functionality is tested via
#2926

Author: denik

acceptance/internal/handlers.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/databricks/databricks-sdk-go/service/catalog"
 	"github.com/databricks/databricks-sdk-go/service/iam"
-	"github.com/databricks/databricks-sdk-go/service/pipelines"
 
 	"github.com/databricks/databricks-sdk-go/service/compute"
 	"github.com/databricks/databricks-sdk-go/service/jobs"
@@ -188,32 +187,11 @@ func addDefaultHandlers(server *testserver.Server) {
 		return req.Workspace.JobsReset(request)
 	})
 
-	server.Handle("POST", "/api/2.0/pipelines", func(req testserver.Request) any {
-		var request pipelines.PipelineSpec
-		if err := json.Unmarshal(req.Body, &request); err != nil {
-			return testserver.Response{
-				Body:       fmt.Sprintf("internal error: %s", err),
-				StatusCode: 400,
-			}
-		}
-
-		return req.Workspace.PipelinesCreate(request)
-	})
-
-	server.Handle("DELETE", "/api/2.0/pipelines/{pipeline_id}", func(req testserver.Request) any {
-		return testserver.MapDelete(req.Workspace, req.Workspace.Pipelines, req.Vars["pipeline_id"])
-	})
-
 	server.Handle("GET", "/api/2.2/jobs/get", func(req testserver.Request) any {
 		jobId := req.URL.Query().Get("job_id")
 		return req.Workspace.JobsGet(jobId)
 	})
 
-	server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}", func(req testserver.Request) any {
-		pipelineId := req.Vars["pipeline_id"]
-		return req.Workspace.PipelinesGet(pipelineId)
-	})
-
 	server.Handle("GET", "/api/2.2/jobs/list", func(req testserver.Request) any {
 		return req.Workspace.JobsList()
 	})
@@ -274,6 +252,24 @@ func addDefaultHandlers(server *testserver.Server) {
 		}
 	})
 
+	// Pipelines:
+
+	server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}", func(req testserver.Request) any {
+		return req.Workspace.PipelineGet(req.Vars["pipeline_id"])
+	})
+
+	server.Handle("POST", "/api/2.0/pipelines", func(req testserver.Request) any {
+		return req.Workspace.PipelineCreate(req)
+	})
+
+	server.Handle("PUT", "/api/2.0/pipelines/{pipeline_id}", func(req testserver.Request) any {
+		return req.Workspace.PipelineUpdate(req, req.Vars["pipeline_id"])
+	})
+
+	server.Handle("DELETE", "/api/2.0/pipelines/{pipeline_id}", func(req testserver.Request) any {
+		return testserver.MapDelete(req.Workspace, req.Workspace.Pipelines, req.Vars["pipeline_id"])
+	})
+
 	// Quality monitors:
 
 	server.Handle("GET", "/api/2.1/unity-catalog/tables/{table_name}/monitor", func(req testserver.Request) any {

libs/testserver/fake_workspace.go

@@ -15,7 +15,6 @@ import (
 	"github.com/databricks/databricks-sdk-go/service/jobs"
 	"github.com/databricks/databricks-sdk-go/service/pipelines"
 	"github.com/databricks/databricks-sdk-go/service/workspace"
-	"github.com/google/uuid"
 )
 
 // FakeWorkspace holds a state of a workspace for acceptance tests.
@@ -232,28 +231,6 @@ func (s *FakeWorkspace) JobsReset(request jobs.ResetJob) Response {
 	}
 }
 
-func (s *FakeWorkspace) PipelinesCreate(r pipelines.PipelineSpec) Response {
-	defer s.LockUnlock()()
-
-	pipelineId := uuid.New().String()
-
-	r.Id = pipelineId
-
-	// If the pipeline definition does not specify a catalog, it switches to Hive metastore mode
-	// and if the storage location is not specified, API automatically generates a storage location
-	// (ref: https://docs.databricks.com/gcp/en/dlt/hive-metastore#specify-a-storage-location)
-	if r.Storage == "" && r.Catalog == "" {
-		r.Storage = "dbfs:/pipelines/" + pipelineId
-	}
-	s.Pipelines[pipelineId] = r
-
-	return Response{
-		Body: pipelines.CreatePipelineResponse{
-			PipelineId: pipelineId,
-		},
-	}
-}
-
 func (s *FakeWorkspace) JobsGet(jobId string) Response {
 	id := jobId
 
@@ -325,24 +302,6 @@ func (s *FakeWorkspace) JobsGetRun(runId int64) Response {
 	}
 }
 
-func (s *FakeWorkspace) PipelinesGet(pipelineId string) Response {
-	defer s.LockUnlock()()
-
-	spec, ok := s.Pipelines[pipelineId]
-	if !ok {
-		return Response{
-			StatusCode: 404,
-		}
-	}
-
-	return Response{
-		Body: pipelines.GetPipelineResponse{
-			PipelineId: pipelineId,
-			Spec:       &spec,
-		},
-	}
-}
-
 func (s *FakeWorkspace) JobsList() Response {
 	defer s.LockUnlock()()
 

libs/testserver/pipelines.go

@@ -0,0 +1,83 @@
+package testserver
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/databricks/databricks-sdk-go/service/pipelines"
+	"github.com/google/uuid"
+)
+
+func (s *FakeWorkspace) PipelineGet(pipelineId string) Response {
+	defer s.LockUnlock()()
+
+	spec, ok := s.Pipelines[pipelineId]
+	if !ok {
+		return Response{
+			StatusCode: 404,
+		}
+	}
+
+	return Response{
+		Body: pipelines.GetPipelineResponse{
+			PipelineId: pipelineId,
+			Spec:       &spec,
+		},
+	}
+}
+
+func (s *FakeWorkspace) PipelineCreate(req Request) Response {
+	defer s.LockUnlock()()
+
+	var r pipelines.PipelineSpec
+	err := json.Unmarshal(req.Body, &r)
+	if err != nil {
+		return Response{
+			Body:       fmt.Sprintf("cannot unmarshal request body: %s", err),
+			StatusCode: 400,
+		}
+	}
+
+	pipelineId := uuid.New().String()
+	r.Id = pipelineId
+
+	// If the pipeline definition does not specify a catalog, it switches to Hive metastore mode
+	// and if the storage location is not specified, API automatically generates a storage location
+	// (ref: https://docs.databricks.com/gcp/en/dlt/hive-metastore#specify-a-storage-location)
+	if r.Storage == "" && r.Catalog == "" {
+		r.Storage = "dbfs:/pipelines/" + pipelineId
+	}
+	s.Pipelines[pipelineId] = r
+
+	return Response{
+		Body: pipelines.CreatePipelineResponse{
+			PipelineId: pipelineId,
+		},
+	}
+}
+
+func (s *FakeWorkspace) PipelineUpdate(req Request, pipelineId string) Response {
+	defer s.LockUnlock()()
+
+	var request pipelines.PipelineSpec
+	err := json.Unmarshal(req.Body, &request)
+	if err != nil {
+		return Response{
+			Body:       fmt.Sprintf("internal error: %s", err),
+			StatusCode: 400,
+		}
+	}
+
+	_, exists := s.Pipelines[pipelineId]
+	if !exists {
+		return Response{
+			StatusCode: 404,
+		}
+	}
+
+	s.Pipelines[pipelineId] = request
+
+	return Response{
+		Body: pipelines.EditPipelineResponse{},
+	}
+}