Deploy the testing property of pipelines and show remote changes in bundle plan

acceptance/bundle/refschema/output/pipelines.txt

@@ -867,14 +867,14 @@ resources.pipelines.*.storage	string	ALL
 resources.pipelines.*.tags	map[string]string	ALL
 resources.pipelines.*.tags.*	string	ALL
 resources.pipelines.*.target	string	ALL
-resources.pipelines.*.testing	*resources.PipelineTesting	INPUT
-resources.pipelines.*.testing.environment	*resources.PipelineTestEnvironment	INPUT
-resources.pipelines.*.testing.environment.dependencies	[]string	INPUT
-resources.pipelines.*.testing.environment.dependencies[*]	string	INPUT
-resources.pipelines.*.testing.sources	[]resources.PipelineTestSource	INPUT
-resources.pipelines.*.testing.sources[*]	resources.PipelineTestSource	INPUT
-resources.pipelines.*.testing.sources[*].glob	*pipelines.PathPattern	INPUT
-resources.pipelines.*.testing.sources[*].glob.include	string	INPUT
+resources.pipelines.*.testing	*resources.PipelineTesting	ALL
+resources.pipelines.*.testing.environment	*resources.PipelineTestEnvironment	ALL
+resources.pipelines.*.testing.environment.dependencies	[]string	ALL
+resources.pipelines.*.testing.environment.dependencies[*]	string	ALL
+resources.pipelines.*.testing.sources	[]resources.PipelineTestSource	ALL
+resources.pipelines.*.testing.sources[*]	resources.PipelineTestSource	ALL
+resources.pipelines.*.testing.sources[*].glob	*pipelines.PathPattern	ALL
+resources.pipelines.*.testing.sources[*].glob.include	string	ALL
 resources.pipelines.*.trigger	*pipelines.PipelineTrigger	ALL
 resources.pipelines.*.trigger.cron	*pipelines.CronTrigger	ALL
 resources.pipelines.*.trigger.cron.quartz_cron_schedule	string	ALL

acceptance/bundle/resources/pipelines/testing/output.txt

@@ -33,3 +33,115 @@ Workspace:
   Path: /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default
 
 Found 1 error
+
+=== Deploy sends testing, with the local wheel uploaded
+
+>>> [CLI] bundle deploy
+Uploading dist/test_helpers-0.1-py3-none-any.whl...
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files...
+Created pipelines.orders
+Files: 6 uploaded, 0 deleted
+Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
+
+>>> print_requests.py //api/2.0/pipelines --nostamp
+{
+  "method": "POST",
+  "path": "/api/2.0/pipelines",
+  "testing": {
+    "sources": [
+      {
+        "glob": {
+          "include": "/Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files/tests/**"
+        }
+      }
+    ],
+    "environment": {
+      "dependencies": [
+        "-e /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files[test]",
+        "-r /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files/tests/requirements.txt",
+        "/Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/artifacts/.internal/test_helpers-0.1-py3-none-any.whl"
+      ]
+    }
+  }
+}
+
+=== No changes after the deploy
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
+
+=== A remote edit that drops testing shows in the plan
+
+>>> [CLI] bundle plan -o json
+"update"
+
+>>> [CLI] bundle deploy
+Uploading dist/test_helpers-0.1-py3-none-any.whl...
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files...
+Updated pipelines.orders
+Files: 1 uploaded, 0 deleted
+Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
+
+>>> [CLI] bundle plan
+Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
+
+=== Changing testing updates the pipeline
+
+>>> update_file.py databricks.yml ./tests/** ./tests/unit/**
+
+>>> [CLI] bundle plan
+update pipelines.orders
+
+Plan: 0 to add, 1 to change, 0 to delete, 0 unchanged
+
+>>> [CLI] bundle deploy
+Uploading dist/test_helpers-0.1-py3-none-any.whl...
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files...
+Updated pipelines.orders
+Files: 1 uploaded, 0 deleted
+Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
+
+>>> print_requests.py //api/2.0/pipelines --nostamp
+{
+  "method": "PUT",
+  "path": "/api/2.0/pipelines/[ORDERS_ID]",
+  "sources": [
+    {
+      "glob": {
+        "include": "/Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files/tests/unit/**"
+      }
+    }
+  ]
+}
+
+=== Removing testing sends an update without it
+
+>>> [CLI] bundle plan
+update pipelines.orders
+
+Plan: 0 to add, 1 to change, 0 to delete, 0 unchanged
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files...
+Updated pipelines.orders
+Files: 1 uploaded, 0 deleted
+Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
+
+>>> print_requests.py //api/2.0/pipelines --nostamp
+{
+  "method": "PUT",
+  "path": "/api/2.0/pipelines/[ORDERS_ID]",
+  "testing": null
+}
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.pipelines.orders
+
+This action will result in the deletion of the following Lakeflow Spark Declarative Pipelines along with the
+Streaming Tables (STs) and Materialized Views (MVs) managed by them. Set 'cascade_on_destroy: false' on a pipeline to retain datasets on pipeline deletion:
+  delete resources.pipelines.orders
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default
+
+Destroy: 1 deleted

acceptance/bundle/resources/pipelines/testing/script

@@ -3,3 +3,36 @@ trace $CLI bundle validate -o json | jq .resources.pipelines.orders.testing
 
 title "Without pipeline testing, the property is an error\n"
 trace musterr env -u DATABRICKS_BUNDLE_ENABLE_EXPERIMENTAL_PIPELINE_TESTING $CLI bundle validate
+
+title "Deploy sends testing, with the local wheel uploaded\n"
+trace $CLI bundle deploy
+trace print_requests.py //api/2.0/pipelines --nostamp | jq '{method, path, testing: .body.testing}'
+
+title "No changes after the deploy\n"
+trace $CLI bundle plan
+rm "$OUT_REQUESTS"
+
+title "A remote edit that drops testing shows in the plan\n"
+PIPELINE_ID=$(read_id.py orders)
+MSYS_NO_PATHCONV=1 $CLI api get "/api/2.0/pipelines/$PIPELINE_ID" | jq '.spec | del(.testing)' > edit.json
+MSYS_NO_PATHCONV=1 $CLI api put "/api/2.0/pipelines/$PIPELINE_ID" --json @edit.json &> LOG.edit
+rm edit.json
+trace $CLI bundle plan -o json | jq '.plan["resources.pipelines.orders"].changes.testing.action'
+trace $CLI bundle deploy
+trace $CLI bundle plan
+rm "$OUT_REQUESTS"
+
+title "Changing testing updates the pipeline\n"
+trace update_file.py databricks.yml './tests/**' './tests/unit/**'
+trace $CLI bundle plan
+trace $CLI bundle deploy
+trace print_requests.py //api/2.0/pipelines --nostamp | jq '{method, path, sources: .body.testing.sources}'
+
+title "Removing testing sends an update without it\n"
+update_file.py databricks.yml "$(sed -n '/^      testing:/,$p' databricks.yml)" ""
+trace $CLI bundle plan
+trace $CLI bundle deploy
+trace print_requests.py //api/2.0/pipelines --nostamp | jq '{method, path, testing: .body.testing}'
+
+trace $CLI bundle destroy --auto-approve
+rm "$OUT_REQUESTS"

bundle/direct/dresources/pipeline.go

@@ -2,13 +2,19 @@ package dresources
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
 
 	"github.com/databricks/cli/bundle/config/resources"
 	"github.com/databricks/cli/bundle/deployplan"
+	"github.com/databricks/cli/bundle/env"
+	"github.com/databricks/cli/libs/auth"
 	"github.com/databricks/cli/libs/structs/structdiff"
 	"github.com/databricks/cli/libs/structs/structpath"
 	"github.com/databricks/cli/libs/utils"
 	"github.com/databricks/databricks-sdk-go"
+	"github.com/databricks/databricks-sdk-go/client"
 	"github.com/databricks/databricks-sdk-go/marshal"
 	"github.com/databricks/databricks-sdk-go/service/pipelines"
 )
@@ -22,6 +28,9 @@ type PipelineState struct {
 	// STs, Views). Nil means the server default (cascade) applies. Read from persisted state at
 	// delete time; never sent on create/update.
 	CascadeOnDestroy *bool `json:"cascade_on_destroy,omitempty"`
+
+	// Testing is PipelineSpec.testing, which the Go SDK does not have yet.
+	Testing *resources.PipelineTesting `json:"testing,omitempty"`
 }
 
 func (s *PipelineState) UnmarshalJSON(b []byte) error {
@@ -52,6 +61,9 @@ type PipelineRemote struct {
 	PipelineId                   string                              `json:"pipeline_id,omitempty"`
 	RunAsUserName                string                              `json:"run_as_user_name,omitempty"`
 	State                        pipelines.PipelineState             `json:"state,omitempty"`
+
+	// Testing is spec.testing of the Get response, read by DoRead because the Go SDK does not have it yet.
+	Testing *resources.PipelineTesting `json:"testing,omitempty"`
 }
 
 // Custom marshaler needed because embedded CreatePipeline has its own MarshalJSON
@@ -78,6 +90,7 @@ func (*ResourcePipeline) PrepareState(input *resources.Pipeline) *PipelineState
 	return &PipelineState{
 		CreatePipeline:   input.CreatePipeline,
 		CascadeOnDestroy: input.CascadeOnDestroy,
+		Testing:          input.Testing,
 	}
 }
 
@@ -86,17 +99,48 @@ func (*ResourcePipeline) RemapState(remote *PipelineRemote) *PipelineState {
 		CreatePipeline: remote.CreatePipeline,
 		// cascade_on_destroy is input-only and absent from PipelineRemote, so it stays nil here.
 		CascadeOnDestroy: nil,
+		Testing:          remote.Testing,
 	}
 }
 
+// pipelineTestingResponse holds spec.testing of a Get Pipeline response.
+type pipelineTestingResponse struct {
+	Spec struct {
+		Testing *resources.PipelineTesting `json:"testing"`
+	} `json:"spec"`
+}
+
 func (r *ResourcePipeline) DoRead(ctx context.Context, id string) (*PipelineRemote, error) {
+	if env.ExperimentalPipelineTesting(ctx) {
+		return r.doReadWithTesting(ctx, id)
+	}
 	resp, err := r.client.Pipelines.GetByPipelineId(ctx, id)
 	if err != nil {
 		return nil, err
 	}
 	return makePipelineRemote(resp), nil
 }
 
+// doReadWithTesting sends the same request as GetByPipelineId, but raw, so that it can also read
+// spec.testing, which GetPipelineResponse does not have yet. Remove it once GetPipelineResponse has it.
+func (r *ResourcePipeline) doReadWithTesting(ctx context.Context, id string) (*PipelineRemote, error) {
+	var raw json.RawMessage
+	if err := r.doRaw(ctx, http.MethodGet, "/api/2.0/pipelines/"+id, nil, &raw); err != nil {
+		return nil, err
+	}
+	var resp pipelines.GetPipelineResponse
+	if err := json.Unmarshal(raw, &resp); err != nil {
+		return nil, err
+	}
+	var testing pipelineTestingResponse
+	if err := json.Unmarshal(raw, &testing); err != nil {
+		return nil, err
+	}
+	remote := makePipelineRemote(&resp)
+	remote.Testing = testing.Spec.Testing
+	return remote, nil
+}
+
 func makePipelineRemote(p *pipelines.GetPipelineResponse) *PipelineRemote {
 	var createPipeline pipelines.CreatePipeline
 	if p.Spec != nil {
@@ -155,10 +199,20 @@ func makePipelineRemote(p *pipelines.GetPipelineResponse) *PipelineRemote {
 		PipelineId:                   p.PipelineId,
 		RunAsUserName:                p.RunAsUserName,
 		State:                        p.State,
+		// GetPipelineResponse has no spec.testing; DoRead sets it from the raw response.
+		Testing: nil,
 	}
 }
 
 func (r *ResourcePipeline) DoCreate(ctx context.Context, config *PipelineState) (string, *PipelineRemote, error) {
+	if config.Testing != nil {
+		var response pipelines.CreatePipelineResponse
+		err := r.doWithTesting(ctx, http.MethodPost, "/api/2.0/pipelines", config.CreatePipeline, config.Testing, &response)
+		if err != nil {
+			return "", nil, err
+		}
+		return response.PipelineId, nil, nil
+	}
 	response, err := r.client.Pipelines.Create(ctx, config.CreatePipeline)
 	if err != nil {
 		return "", nil, err
@@ -212,6 +266,9 @@ func (r *ResourcePipeline) DoUpdate(ctx context.Context, id string, config *Pipe
 		ForceSendFields:      utils.FilterFields[pipelines.EditPipeline](config.ForceSendFields),
 	}
 
+	if config.Testing != nil {
+		return nil, r.doWithTesting(ctx, http.MethodPut, "/api/2.0/pipelines/"+id, request, config.Testing, nil)
+	}
 	return nil, r.client.Pipelines.Update(ctx, request)
 }
 
@@ -242,6 +299,35 @@ func (*ResourcePipeline) OverrideChangeDesc(_ context.Context, path *structpath.
 	return nil
 }
 
+// doWithTesting sends a Create or Edit request with the testing field, which the Go SDK does not
+// have yet. The body is the SDK's JSON for the request plus testing. Remove this once
+// CreatePipeline and EditPipeline have the field; pipelines without testing already use the SDK.
+func (r *ResourcePipeline) doWithTesting(ctx context.Context, method, path string, request any, testing *resources.PipelineTesting, response any) error {
+	data, err := json.Marshal(request)
+	if err != nil {
+		return err
+	}
+	// RawMessage values keep the SDK's encoding, including large integers, unchanged.
+	var body map[string]json.RawMessage
+	if err := json.Unmarshal(data, &body); err != nil {
+		return err
+	}
+	body["testing"], err = json.Marshal(testing)
+	if err != nil {
+		return err
+	}
+	return r.doRaw(ctx, method, path, body, response)
+}
+
+// doRaw sends a request that the Go SDK cannot express yet because it lacks the testing field.
+func (r *ResourcePipeline) doRaw(ctx context.Context, method, path string, request, response any) error {
+	apiClient, err := client.New(r.client.Config)
+	if err != nil {
+		return fmt.Errorf("failed to create API client: %w", err)
+	}
+	return apiClient.Do(ctx, method, path, auth.WorkspaceIDHeaders(r.client.Config), nil, request, response)
+}
+
 // Note, terraform provider either
 // a) reads back state at least once and fails create if state is "failed"
 // b) repeatededly reads state until state is "running" (if spec.Contionous is set).