Skip to content

Commit 35031d3

Browse files
tharunbhuk-dbIsaac
andcommitted
Deploy the testing property of pipelines and show remote changes in bundle plan
bundle deploy sends a pipeline's testing on Create and Edit, and bundle plan reads it back, so a change made outside the bundle shows in the plan and the next deploy undoes it. The Go SDK does not have the field yet, so these calls are raw: a pipeline without testing is still created and edited with the SDK, and the raw read is used only with DATABRICKS_BUNDLE_ENABLE_EXPERIMENTAL_PIPELINE_TESTING=true. The test server stores spec.testing on Create and Edit and returns it on Get. Co-authored-by: Isaac <no-reply@databricks.com>
1 parent 858d4df commit 35031d3

9 files changed

Lines changed: 370 additions & 12 deletions

File tree

‎acceptance/bundle/refschema/output/pipelines.txt‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -867,14 +867,14 @@ resources.pipelines.*.storage string ALL
867867
resources.pipelines.*.tags map[string]string ALL
868868
resources.pipelines.*.tags.* string ALL
869869
resources.pipelines.*.target string ALL
870-
resources.pipelines.*.testing *resources.PipelineTesting INPUT
871-
resources.pipelines.*.testing.environment *resources.PipelineTestEnvironment INPUT
872-
resources.pipelines.*.testing.environment.dependencies []string INPUT
873-
resources.pipelines.*.testing.environment.dependencies[*] string INPUT
874-
resources.pipelines.*.testing.sources []resources.PipelineTestSource INPUT
875-
resources.pipelines.*.testing.sources[*] resources.PipelineTestSource INPUT
876-
resources.pipelines.*.testing.sources[*].glob *pipelines.PathPattern INPUT
877-
resources.pipelines.*.testing.sources[*].glob.include string INPUT
870+
resources.pipelines.*.testing *resources.PipelineTesting ALL
871+
resources.pipelines.*.testing.environment *resources.PipelineTestEnvironment ALL
872+
resources.pipelines.*.testing.environment.dependencies []string ALL
873+
resources.pipelines.*.testing.environment.dependencies[*] string ALL
874+
resources.pipelines.*.testing.sources []resources.PipelineTestSource ALL
875+
resources.pipelines.*.testing.sources[*] resources.PipelineTestSource ALL
876+
resources.pipelines.*.testing.sources[*].glob *pipelines.PathPattern ALL
877+
resources.pipelines.*.testing.sources[*].glob.include string ALL
878878
resources.pipelines.*.trigger *pipelines.PipelineTrigger ALL
879879
resources.pipelines.*.trigger.cron *pipelines.CronTrigger ALL
880880
resources.pipelines.*.trigger.cron.quartz_cron_schedule string ALL

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

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,3 +33,115 @@ Workspace:
3333
Path: /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default
3434

3535
Found 1 error
36+
37+
=== Deploy sends testing, with the local wheel uploaded
38+
39+
>>> [CLI] bundle deploy
40+
Uploading dist/test_helpers-0.1-py3-none-any.whl...
41+
Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files...
42+
Created pipelines.orders
43+
Files: 6 uploaded, 0 deleted
44+
Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
45+
46+
>>> print_requests.py //api/2.0/pipelines --nostamp
47+
{
48+
"method": "POST",
49+
"path": "/api/2.0/pipelines",
50+
"testing": {
51+
"sources": [
52+
{
53+
"glob": {
54+
"include": "/Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files/tests/**"
55+
}
56+
}
57+
],
58+
"environment": {
59+
"dependencies": [
60+
"-e /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files[test]",
61+
"-r /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files/tests/requirements.txt",
62+
"/Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/artifacts/.internal/test_helpers-0.1-py3-none-any.whl"
63+
]
64+
}
65+
}
66+
}
67+
68+
=== No changes after the deploy
69+
70+
>>> [CLI] bundle plan
71+
Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
72+
73+
=== A remote edit that drops testing shows in the plan
74+
75+
>>> [CLI] bundle plan -o json
76+
"update"
77+
78+
>>> [CLI] bundle deploy
79+
Uploading dist/test_helpers-0.1-py3-none-any.whl...
80+
Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files...
81+
Updated pipelines.orders
82+
Files: 1 uploaded, 0 deleted
83+
Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
84+
85+
>>> [CLI] bundle plan
86+
Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
87+
88+
=== Changing testing updates the pipeline
89+
90+
>>> update_file.py databricks.yml ./tests/** ./tests/unit/**
91+
92+
>>> [CLI] bundle plan
93+
update pipelines.orders
94+
95+
Plan: 0 to add, 1 to change, 0 to delete, 0 unchanged
96+
97+
>>> [CLI] bundle deploy
98+
Uploading dist/test_helpers-0.1-py3-none-any.whl...
99+
Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files...
100+
Updated pipelines.orders
101+
Files: 1 uploaded, 0 deleted
102+
Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
103+
104+
>>> print_requests.py //api/2.0/pipelines --nostamp
105+
{
106+
"method": "PUT",
107+
"path": "/api/2.0/pipelines/[ORDERS_ID]",
108+
"sources": [
109+
{
110+
"glob": {
111+
"include": "/Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files/tests/unit/**"
112+
}
113+
}
114+
]
115+
}
116+
117+
=== Removing testing sends an update without it
118+
119+
>>> [CLI] bundle plan
120+
update pipelines.orders
121+
122+
Plan: 0 to add, 1 to change, 0 to delete, 0 unchanged
123+
124+
>>> [CLI] bundle deploy
125+
Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default/files...
126+
Updated pipelines.orders
127+
Files: 1 uploaded, 0 deleted
128+
Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
129+
130+
>>> print_requests.py //api/2.0/pipelines --nostamp
131+
{
132+
"method": "PUT",
133+
"path": "/api/2.0/pipelines/[ORDERS_ID]",
134+
"testing": null
135+
}
136+
137+
>>> [CLI] bundle destroy --auto-approve
138+
The following resources will be deleted:
139+
delete resources.pipelines.orders
140+
141+
This action will result in the deletion of the following Lakeflow Spark Declarative Pipelines along with the
142+
Streaming Tables (STs) and Materialized Views (MVs) managed by them. Set 'cascade_on_destroy: false' on a pipeline to retain datasets on pipeline deletion:
143+
delete resources.pipelines.orders
144+
145+
All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/pipeline-testing/default
146+
147+
Destroy: 1 deleted

‎acceptance/bundle/resources/pipelines/testing/script‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,3 +3,36 @@ trace $CLI bundle validate -o json | jq .resources.pipelines.orders.testing
33

44
title "Without pipeline testing, the property is an error\n"
55
trace musterr env -u DATABRICKS_BUNDLE_ENABLE_EXPERIMENTAL_PIPELINE_TESTING $CLI bundle validate
6+
7+
title "Deploy sends testing, with the local wheel uploaded\n"
8+
trace $CLI bundle deploy
9+
trace print_requests.py //api/2.0/pipelines --nostamp | jq '{method, path, testing: .body.testing}'
10+
11+
title "No changes after the deploy\n"
12+
trace $CLI bundle plan
13+
rm "$OUT_REQUESTS"
14+
15+
title "A remote edit that drops testing shows in the plan\n"
16+
PIPELINE_ID=$(read_id.py orders)
17+
MSYS_NO_PATHCONV=1 $CLI api get "/api/2.0/pipelines/$PIPELINE_ID" | jq '.spec | del(.testing)' > edit.json
18+
MSYS_NO_PATHCONV=1 $CLI api put "/api/2.0/pipelines/$PIPELINE_ID" --json @edit.json &> LOG.edit
19+
rm edit.json
20+
trace $CLI bundle plan -o json | jq '.plan["resources.pipelines.orders"].changes.testing.action'
21+
trace $CLI bundle deploy
22+
trace $CLI bundle plan
23+
rm "$OUT_REQUESTS"
24+
25+
title "Changing testing updates the pipeline\n"
26+
trace update_file.py databricks.yml './tests/**' './tests/unit/**'
27+
trace $CLI bundle plan
28+
trace $CLI bundle deploy
29+
trace print_requests.py //api/2.0/pipelines --nostamp | jq '{method, path, sources: .body.testing.sources}'
30+
31+
title "Removing testing sends an update without it\n"
32+
update_file.py databricks.yml "$(sed -n '/^ testing:/,$p' databricks.yml)" ""
33+
trace $CLI bundle plan
34+
trace $CLI bundle deploy
35+
trace print_requests.py //api/2.0/pipelines --nostamp | jq '{method, path, testing: .body.testing}'
36+
37+
trace $CLI bundle destroy --auto-approve
38+
rm "$OUT_REQUESTS"

‎bundle/direct/dresources/pipeline.go‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,19 @@ package dresources
22

33
import (
44
"context"
5+
"encoding/json"
6+
"fmt"
7+
"net/http"
58

69
"github.com/databricks/cli/bundle/config/resources"
710
"github.com/databricks/cli/bundle/deployplan"
11+
"github.com/databricks/cli/bundle/env"
12+
"github.com/databricks/cli/libs/auth"
813
"github.com/databricks/cli/libs/structs/structdiff"
914
"github.com/databricks/cli/libs/structs/structpath"
1015
"github.com/databricks/cli/libs/utils"
1116
"github.com/databricks/databricks-sdk-go"
17+
"github.com/databricks/databricks-sdk-go/client"
1218
"github.com/databricks/databricks-sdk-go/marshal"
1319
"github.com/databricks/databricks-sdk-go/service/pipelines"
1420
)
@@ -22,6 +28,9 @@ type PipelineState struct {
2228
// STs, Views). Nil means the server default (cascade) applies. Read from persisted state at
2329
// delete time; never sent on create/update.
2430
CascadeOnDestroy *bool `json:"cascade_on_destroy,omitempty"`
31+
32+
// Testing is PipelineSpec.testing, which the Go SDK does not have yet.
33+
Testing *resources.PipelineTesting `json:"testing,omitempty"`
2534
}
2635

2736
func (s *PipelineState) UnmarshalJSON(b []byte) error {
@@ -52,6 +61,9 @@ type PipelineRemote struct {
5261
PipelineId string `json:"pipeline_id,omitempty"`
5362
RunAsUserName string `json:"run_as_user_name,omitempty"`
5463
State pipelines.PipelineState `json:"state,omitempty"`
64+
65+
// Testing is spec.testing of the Get response, read by DoRead because the Go SDK does not have it yet.
66+
Testing *resources.PipelineTesting `json:"testing,omitempty"`
5567
}
5668

5769
// Custom marshaler needed because embedded CreatePipeline has its own MarshalJSON
@@ -78,6 +90,7 @@ func (*ResourcePipeline) PrepareState(input *resources.Pipeline) *PipelineState
7890
return &PipelineState{
7991
CreatePipeline: input.CreatePipeline,
8092
CascadeOnDestroy: input.CascadeOnDestroy,
93+
Testing: input.Testing,
8194
}
8295
}
8396

@@ -86,17 +99,48 @@ func (*ResourcePipeline) RemapState(remote *PipelineRemote) *PipelineState {
8699
CreatePipeline: remote.CreatePipeline,
87100
// cascade_on_destroy is input-only and absent from PipelineRemote, so it stays nil here.
88101
CascadeOnDestroy: nil,
102+
Testing: remote.Testing,
89103
}
90104
}
91105

106+
// pipelineTestingResponse holds spec.testing of a Get Pipeline response.
107+
type pipelineTestingResponse struct {
108+
Spec struct {
109+
Testing *resources.PipelineTesting `json:"testing"`
110+
} `json:"spec"`
111+
}
112+
92113
func (r *ResourcePipeline) DoRead(ctx context.Context, id string) (*PipelineRemote, error) {
114+
if env.ExperimentalPipelineTesting(ctx) {
115+
return r.doReadWithTesting(ctx, id)
116+
}
93117
resp, err := r.client.Pipelines.GetByPipelineId(ctx, id)
94118
if err != nil {
95119
return nil, err
96120
}
97121
return makePipelineRemote(resp), nil
98122
}
99123

124+
// doReadWithTesting sends the same request as GetByPipelineId, but raw, so that it can also read
125+
// spec.testing, which GetPipelineResponse does not have yet. Remove it once GetPipelineResponse has it.
126+
func (r *ResourcePipeline) doReadWithTesting(ctx context.Context, id string) (*PipelineRemote, error) {
127+
var raw json.RawMessage
128+
if err := r.doRaw(ctx, http.MethodGet, "/api/2.0/pipelines/"+id, nil, &raw); err != nil {
129+
return nil, err
130+
}
131+
var resp pipelines.GetPipelineResponse
132+
if err := json.Unmarshal(raw, &resp); err != nil {
133+
return nil, err
134+
}
135+
var testing pipelineTestingResponse
136+
if err := json.Unmarshal(raw, &testing); err != nil {
137+
return nil, err
138+
}
139+
remote := makePipelineRemote(&resp)
140+
remote.Testing = testing.Spec.Testing
141+
return remote, nil
142+
}
143+
100144
func makePipelineRemote(p *pipelines.GetPipelineResponse) *PipelineRemote {
101145
var createPipeline pipelines.CreatePipeline
102146
if p.Spec != nil {
@@ -155,10 +199,20 @@ func makePipelineRemote(p *pipelines.GetPipelineResponse) *PipelineRemote {
155199
PipelineId: p.PipelineId,
156200
RunAsUserName: p.RunAsUserName,
157201
State: p.State,
202+
// GetPipelineResponse has no spec.testing; DoRead sets it from the raw response.
203+
Testing: nil,
158204
}
159205
}
160206

161207
func (r *ResourcePipeline) DoCreate(ctx context.Context, config *PipelineState) (string, *PipelineRemote, error) {
208+
if config.Testing != nil {
209+
var response pipelines.CreatePipelineResponse
210+
err := r.doWithTesting(ctx, http.MethodPost, "/api/2.0/pipelines", config.CreatePipeline, config.Testing, &response)
211+
if err != nil {
212+
return "", nil, err
213+
}
214+
return response.PipelineId, nil, nil
215+
}
162216
response, err := r.client.Pipelines.Create(ctx, config.CreatePipeline)
163217
if err != nil {
164218
return "", nil, err
@@ -212,6 +266,9 @@ func (r *ResourcePipeline) DoUpdate(ctx context.Context, id string, config *Pipe
212266
ForceSendFields: utils.FilterFields[pipelines.EditPipeline](config.ForceSendFields),
213267
}
214268

269+
if config.Testing != nil {
270+
return nil, r.doWithTesting(ctx, http.MethodPut, "/api/2.0/pipelines/"+id, request, config.Testing, nil)
271+
}
215272
return nil, r.client.Pipelines.Update(ctx, request)
216273
}
217274

@@ -242,6 +299,35 @@ func (*ResourcePipeline) OverrideChangeDesc(_ context.Context, path *structpath.
242299
return nil
243300
}
244301

302+
// doWithTesting sends a Create or Edit request with the testing field, which the Go SDK does not
303+
// have yet. The body is the SDK's JSON for the request plus testing. Remove this once
304+
// CreatePipeline and EditPipeline have the field; pipelines without testing already use the SDK.
305+
func (r *ResourcePipeline) doWithTesting(ctx context.Context, method, path string, request any, testing *resources.PipelineTesting, response any) error {
306+
data, err := json.Marshal(request)
307+
if err != nil {
308+
return err
309+
}
310+
// RawMessage values keep the SDK's encoding, including large integers, unchanged.
311+
var body map[string]json.RawMessage
312+
if err := json.Unmarshal(data, &body); err != nil {
313+
return err
314+
}
315+
body["testing"], err = json.Marshal(testing)
316+
if err != nil {
317+
return err
318+
}
319+
return r.doRaw(ctx, method, path, body, response)
320+
}
321+
322+
// doRaw sends a request that the Go SDK cannot express yet because it lacks the testing field.
323+
func (r *ResourcePipeline) doRaw(ctx context.Context, method, path string, request, response any) error {
324+
apiClient, err := client.New(r.client.Config)
325+
if err != nil {
326+
return fmt.Errorf("failed to create API client: %w", err)
327+
}
328+
return apiClient.Do(ctx, method, path, auth.WorkspaceIDHeaders(r.client.Config), nil, request, response)
329+
}
330+
245331
// Note, terraform provider either
246332
// a) reads back state at least once and fails create if state is "failed"
247333
// b) repeatededly reads state until state is "running" (if spec.Contionous is set).

0 commit comments

Comments
 (0)