Acceptance tests for pipeline run support in handlers and testserver files (#3178)

## Changes
Added pipeline run support for acceptance testing:
- Created `libs/testserver/pipelines_test.go` with test coverage
- Implemented pipeline operations in `libs/testserver/pipelines.go`
- Enhanced `libs/testserver/fake_workspace.go` for pipeline API support
- Updated `acceptance/internal/handlers.go` for pipeline requests

## Why
Pipeline functionality is needed for acceptance tests to validate
pipeline API calls without requiring actual Databricks workspace
connections. This provides test infrastructure for simulating pipeline
operations.

## Tests
Test coverage validates:
- Pipeline start update functionality with various parameters
- Error handling for invalid requests and non-existent pipelines
- Unique update ID generation
- Complex parameter combinations and edge cases

The test infrastructure ensures pipeline-related acceptance tests can
run.

---
**Follows up on:** #3132

Author: alyssa-db

acceptance/internal/handlers.go

@@ -324,6 +324,22 @@ func addDefaultHandlers(server *testserver.Server) {
 		return testserver.MapDelete(req.Workspace, req.Workspace.Pipelines, req.Vars["pipeline_id"])
 	})
 
+	server.Handle("POST", "/api/2.0/pipelines/{pipeline_id}/updates", func(req testserver.Request) any {
+		return req.Workspace.PipelineStartUpdate(req.Vars["pipeline_id"])
+	})
+
+	server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req testserver.Request) any {
+		return req.Workspace.PipelineEvents(req.Vars["pipeline_id"])
+	})
+
+	server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/updates/{update_id}", func(req testserver.Request) any {
+		return req.Workspace.PipelineGetUpdate(req.Vars["pipeline_id"], req.Vars["update_id"])
+	})
+
+	server.Handle("POST", "/api/2.0/pipelines/{pipeline_id}/stop", func(req testserver.Request) any {
+		return req.Workspace.PipelineStop(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

@@ -47,12 +47,13 @@ type FakeWorkspace struct {
 	Jobs         map[int64]jobs.Job
 	JobRuns      map[int64]jobs.Run
 
-	Pipelines  map[string]pipelines.GetPipelineResponse
-	Monitors   map[string]catalog.MonitorInfo
-	Apps       map[string]apps.App
-	Schemas    map[string]catalog.SchemaInfo
-	Volumes    map[string]catalog.VolumeInfo
-	Dashboards map[string]dashboards.Dashboard
+	Pipelines       map[string]pipelines.GetPipelineResponse
+	PipelineUpdates map[string]bool
+	Monitors        map[string]catalog.MonitorInfo
+	Apps            map[string]apps.App
+	Schemas         map[string]catalog.SchemaInfo
+	Volumes         map[string]catalog.VolumeInfo
+	Dashboards      map[string]dashboards.Dashboard
 }
 
 func (w *FakeWorkspace) LockUnlock() func() {
@@ -117,17 +118,18 @@ func NewFakeWorkspace(url string) *FakeWorkspace {
 		directories: map[string]bool{
 			"/Workspace": true,
 		},
-		files:        make(map[string]FileEntry),
-		Jobs:         map[int64]jobs.Job{},
-		JobRuns:      map[int64]jobs.Run{},
-		nextJobId:    TestJobID,
-		nextJobRunId: TestRunID,
-		Pipelines:    map[string]pipelines.GetPipelineResponse{},
-		Monitors:     map[string]catalog.MonitorInfo{},
-		Apps:         map[string]apps.App{},
-		Schemas:      map[string]catalog.SchemaInfo{},
-		Volumes:      map[string]catalog.VolumeInfo{},
-		Dashboards:   map[string]dashboards.Dashboard{},
+		files:           make(map[string]FileEntry),
+		Jobs:            map[int64]jobs.Job{},
+		JobRuns:         map[int64]jobs.Run{},
+		nextJobId:       TestJobID,
+		nextJobRunId:    TestRunID,
+		Pipelines:       map[string]pipelines.GetPipelineResponse{},
+		PipelineUpdates: map[string]bool{},
+		Monitors:        map[string]catalog.MonitorInfo{},
+		Apps:            map[string]apps.App{},
+		Schemas:         map[string]catalog.SchemaInfo{},
+		Volumes:         map[string]catalog.VolumeInfo{},
+		Dashboards:      map[string]dashboards.Dashboard{},
 	}
 }
 

libs/testserver/pipelines.go

@@ -94,3 +94,91 @@ func (s *FakeWorkspace) PipelineUpdate(req Request, pipelineId string) Response
 		Body: pipelines.EditPipelineResponse{},
 	}
 }
+
+func (s *FakeWorkspace) PipelineStartUpdate(pipelineId string) Response {
+	defer s.LockUnlock()()
+
+	_, exists := s.Pipelines[pipelineId]
+	if !exists {
+		return Response{
+			StatusCode: 404,
+			Body:       map[string]string{"message": fmt.Sprintf("The specified pipeline %s was not found.", pipelineId)},
+		}
+	}
+
+	updateId := uuid.New().String()
+	s.PipelineUpdates[updateId] = true
+
+	return Response{
+		Body: pipelines.StartUpdateResponse{
+			UpdateId: updateId,
+		},
+	}
+}
+
+func (s *FakeWorkspace) PipelineEvents(pipelineId string) Response {
+	defer s.LockUnlock()()
+
+	_, exists := s.Pipelines[pipelineId]
+	if !exists {
+		return Response{
+			StatusCode: 404,
+			Body:       map[string]string{"message": fmt.Sprintf("The specified pipeline %s was not found.", pipelineId)},
+		}
+	}
+
+	return Response{
+		Body: map[string]any{
+			"events": []pipelines.PipelineEvent{},
+		},
+	}
+}
+
+func (s *FakeWorkspace) PipelineGetUpdate(pipelineId, updateId string) Response {
+	defer s.LockUnlock()()
+
+	_, exists := s.Pipelines[pipelineId]
+	if !exists {
+		return Response{
+			StatusCode: 404,
+			Body:       map[string]string{"message": fmt.Sprintf("The specified pipeline %s was not found.", pipelineId)},
+		}
+	}
+
+	// Check if the update exists
+	_, updateExists := s.PipelineUpdates[updateId]
+	if !updateExists {
+		return Response{
+			StatusCode: 404,
+			Body:       map[string]string{"message": fmt.Sprintf("The specified update %s was not found.", updateId)},
+		}
+	}
+
+	return Response{
+		Body: pipelines.GetUpdateResponse{
+			Update: &pipelines.UpdateInfo{
+				UpdateId: updateId,
+				State:    pipelines.UpdateInfoStateCompleted,
+			},
+		},
+	}
+}
+
+func (s *FakeWorkspace) PipelineStop(pipelineId string) Response {
+	defer s.LockUnlock()()
+
+	_, exists := s.Pipelines[pipelineId]
+	if !exists {
+		return Response{
+			StatusCode: 404,
+			Body:       map[string]string{"message": fmt.Sprintf("The specified pipeline %s was not found.", pipelineId)},
+		}
+	}
+
+	return Response{
+		Body: pipelines.GetPipelineResponse{
+			PipelineId: pipelineId,
+			State:      pipelines.PipelineStateIdle,
+		},
+	}
+}

libs/testserver/pipelines_test.go

@@ -0,0 +1,64 @@
+package testserver
+
+import (
+	"testing"
+
+	"github.com/databricks/databricks-sdk-go/service/pipelines"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func createTestPipeline(t *testing.T, workspace *FakeWorkspace) string {
+	createReq := Request{
+		Body: []byte(`{
+			"name": "Test Pipeline",
+			"storage": "dbfs:/pipelines/test-pipeline"
+		}`),
+	}
+
+	createResponse := workspace.PipelineCreate(createReq)
+	// StatusCode 0 gets converted to 200 by normalizeResponse in the server
+	require.Equal(t, 0, createResponse.StatusCode)
+
+	createPipelineResponse, ok := createResponse.Body.(pipelines.CreatePipelineResponse)
+	require.True(t, ok)
+	return createPipelineResponse.PipelineId
+}
+
+func TestPipelineStartUpdate_HandlesNonExistentPipeline(t *testing.T) {
+	workspace := NewFakeWorkspace("http://test")
+
+	response := workspace.PipelineStartUpdate("non-existent-pipeline")
+	assert.Equal(t, 404, response.StatusCode)
+	assert.Contains(t, response.Body.(map[string]string)["message"], "The specified pipeline non-existent-pipeline was not found")
+}
+
+func TestPipelineGetUpdate_HandlesNonExistent(t *testing.T) {
+	workspace := NewFakeWorkspace("http://test")
+
+	response := workspace.PipelineGetUpdate("non-existent-pipeline", "some-update-id")
+	assert.Equal(t, 404, response.StatusCode)
+
+	pipelineId := createTestPipeline(t, workspace)
+
+	response = workspace.PipelineGetUpdate(pipelineId, "non-existent-update")
+	assert.Equal(t, 404, response.StatusCode)
+	assert.Contains(t, response.Body.(map[string]string)["message"], "The specified update non-existent-update was not found")
+}
+
+func TestPipelineStop_AfterUpdate(t *testing.T) {
+	workspace := NewFakeWorkspace("http://test")
+
+	pipelineId := createTestPipeline(t, workspace)
+
+	startResponse := workspace.PipelineStartUpdate(pipelineId)
+	assert.Equal(t, 0, startResponse.StatusCode)
+
+	stopResponse := workspace.PipelineStop(pipelineId)
+	assert.Equal(t, 0, stopResponse.StatusCode)
+
+	stopBody, ok := stopResponse.Body.(pipelines.GetPipelineResponse)
+	require.True(t, ok)
+	assert.Equal(t, pipelineId, stopBody.PipelineId)
+	assert.Equal(t, pipelines.PipelineStateIdle, stopBody.State)
+}