diff --git a/acceptance/pipelines/test/arguments/output.txt b/acceptance/pipelines/test/arguments/output.txt index bf63a23f6ef..b64826fccbd 100644 --- a/acceptance/pipelines/test/arguments/output.txt +++ b/acceptance/pipelines/test/arguments/output.txt @@ -26,7 +26,10 @@ example: The command waits for the results. It exits with code 0 only if at least one test was collected and none failed or errored, so a run whose tests were all skipped succeeds; otherwise it exits with code 1. --no-wait returns after the -update starts. To stop the update, run 'pipelines stop'. +update starts. + +--timeout and Ctrl+C stop waiting but not the update; pass its ID to --resume +to read the results later. To stop the update, run 'pipelines stop'. Usage: databricks pipelines test [flags] [KEY] [TEST_SELECTION...] @@ -35,6 +38,8 @@ Pipeline Test Flags: --no-wait Don't wait for the tests to complete. --pipeline-id string ID of the pipeline to test, instead of a KEY. --pytest-arg stringArray Argument to pass to pytest (repeatable). Only -k and -m are supported. + --resume string Read the results of an existing test update instead of starting one. + --timeout duration Stop waiting for results after this duration, for example 30m. The update keeps running. Flags: -h, --help help for test @@ -71,3 +76,22 @@ Error: --pipeline-id "1234" is not a pipeline ID === JSON output is not supported >>> musterr [CLI] pipelines test --pipeline-id 1234 -o json Error: --output json is not supported by pipelines test + +=== Incompatible run-control flags +>>> musterr [CLI] pipelines test --pipeline-id 1234 tests --resume 5678 +Error: --resume cannot be used with test selections + +>>> musterr [CLI] pipelines test --pipeline-id 1234 --resume 5678 --pytest-arg=-m --pytest-arg=smoke +Error: --resume cannot be used with --pytest-arg + +>>> musterr [CLI] pipelines test --pipeline-id 1234 --resume 5678 --no-wait +Error: --resume cannot be used with --no-wait + +>>> musterr [CLI] pipelines test --pipeline-id 1234 --no-wait --timeout 10m +Error: --timeout cannot be used with --no-wait + +>>> musterr [CLI] pipelines test --pipeline-id 1234 --resume 5678' OR '1'='1 +Error: --resume "5678' OR '1'='1" is not an update ID; pass the update ID that pipelines test printed + +>>> musterr [CLI] pipelines test --pipeline-id 1234 --resume +Error: --resume "" is not an update ID; pass the update ID that pipelines test printed diff --git a/acceptance/pipelines/test/arguments/script b/acceptance/pipelines/test/arguments/script index 0df8c20ce2d..62ed32b4a1b 100644 --- a/acceptance/pipelines/test/arguments/script +++ b/acceptance/pipelines/test/arguments/script @@ -16,3 +16,11 @@ trace musterr $CLI pipelines test --pipeline-id 1234 title "JSON output is not supported" trace musterr $CLI pipelines test --pipeline-id 1234 -o json + +title "Incompatible run-control flags" +trace musterr $CLI pipelines test --pipeline-id 1234 tests --resume 5678 +trace musterr $CLI pipelines test --pipeline-id 1234 --resume 5678 --pytest-arg=-m --pytest-arg=smoke +trace musterr $CLI pipelines test --pipeline-id 1234 --resume 5678 --no-wait +trace musterr $CLI pipelines test --pipeline-id 1234 --no-wait --timeout 10m +trace musterr $CLI pipelines test --pipeline-id 1234 --resume "5678' OR '1'='1" +trace musterr $CLI pipelines test --pipeline-id 1234 --resume "" diff --git a/acceptance/pipelines/test/run-control/canceled/out.test.toml b/acceptance/pipelines/test/run-control/canceled/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/run-control/canceled/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/run-control/canceled/output.txt b/acceptance/pipelines/test/run-control/canceled/output.txt new file mode 100644 index 00000000000..72bd75f112a --- /dev/null +++ b/acceptance/pipelines/test/run-control/canceled/output.txt @@ -0,0 +1,9 @@ + +=== A canceled update is an incomplete run +>>> musterr [CLI] pipelines test --pipeline-id [UUID][0] --resume [UUID][1] +Pipeline: [UUID][0] +Update: [UUID][1] + +RUN INCOMPLETE: the update was canceled before all test results were received. +Done. PASSED=0 FAILED=0 SKIPPED=0 ERROR=0 TOTAL=1 +Error: update [UUID][1] is CANCELED diff --git a/acceptance/pipelines/test/run-control/canceled/script b/acceptance/pipelines/test/run-control/canceled/script new file mode 100644 index 00000000000..cfb7d487ab5 --- /dev/null +++ b/acceptance/pipelines/test/run-control/canceled/script @@ -0,0 +1,2 @@ +title "A canceled update is an incomplete run" +trace musterr $CLI pipelines test --pipeline-id $pipeline_id --resume 00000000-0000-0000-0000-0000000000a1 diff --git a/acceptance/pipelines/test/run-control/canceled/test.toml b/acceptance/pipelines/test/run-control/canceled/test.toml new file mode 100644 index 00000000000..f0bb57ea9a0 --- /dev/null +++ b/acceptance/pipelines/test/run-control/canceled/test.toml @@ -0,0 +1,10 @@ +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/updates/{update_id}" +Response.Body = '{"update": {"update_id": "00000000-0000-0000-0000-0000000000a1", "state": "CANCELED", "test_only": true}}' + +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/events" +Response.Body = ''' +{"events": [ + {"id": "e1", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:00.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_slow"}, "details": {"test_case_progress": {"status": "PENDING"}}}]} +''' diff --git a/acceptance/pipelines/test/run-control/not-test-update/out.test.toml b/acceptance/pipelines/test/run-control/not-test-update/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/run-control/not-test-update/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/run-control/not-test-update/output.txt b/acceptance/pipelines/test/run-control/not-test-update/output.txt new file mode 100644 index 00000000000..8e8b038e3f0 --- /dev/null +++ b/acceptance/pipelines/test/run-control/not-test-update/output.txt @@ -0,0 +1,4 @@ + +=== --resume rejects an update that is not a test update +>>> musterr [CLI] pipelines test --pipeline-id [UUID][0] --resume [UUID][1] +Error: update [UUID][1] is not a test update; pass the update ID that pipelines test printed diff --git a/acceptance/pipelines/test/run-control/not-test-update/script b/acceptance/pipelines/test/run-control/not-test-update/script new file mode 100644 index 00000000000..f1985b61601 --- /dev/null +++ b/acceptance/pipelines/test/run-control/not-test-update/script @@ -0,0 +1,2 @@ +title "--resume rejects an update that is not a test update" +trace musterr $CLI pipelines test --pipeline-id $pipeline_id --resume 00000000-0000-0000-0000-0000000000a1 diff --git a/acceptance/pipelines/test/run-control/not-test-update/test.toml b/acceptance/pipelines/test/run-control/not-test-update/test.toml new file mode 100644 index 00000000000..354f263cb3e --- /dev/null +++ b/acceptance/pipelines/test/run-control/not-test-update/test.toml @@ -0,0 +1,3 @@ +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/updates/{update_id}" +Response.Body = '{"update": {"update_id": "00000000-0000-0000-0000-0000000000a1", "state": "COMPLETED", "test_only": false}}' diff --git a/acceptance/pipelines/test/run-control/resume/out.test.toml b/acceptance/pipelines/test/run-control/resume/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/run-control/resume/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/run-control/resume/output.txt b/acceptance/pipelines/test/run-control/resume/output.txt new file mode 100644 index 00000000000..edd687dbdee --- /dev/null +++ b/acceptance/pipelines/test/run-control/resume/output.txt @@ -0,0 +1,9 @@ + +=== --resume reads the results of an existing update +>>> [CLI] pipelines test --pipeline-id [UUID][0] --resume [UUID][1] +Pipeline: [UUID][0] +Update: [UUID][1] +Running tests... +1 of 1 PASSED tests/test_orders.py::test_valid_order ........................... [PASSED in 1.00s] + +Done. PASSED=1 FAILED=0 SKIPPED=0 ERROR=0 TOTAL=1 diff --git a/acceptance/pipelines/test/run-control/resume/script b/acceptance/pipelines/test/run-control/resume/script new file mode 100644 index 00000000000..0d8fb1bada9 --- /dev/null +++ b/acceptance/pipelines/test/run-control/resume/script @@ -0,0 +1,2 @@ +title "--resume reads the results of an existing update" +trace $CLI pipelines test --pipeline-id $pipeline_id --resume 00000000-0000-0000-0000-0000000000a1 diff --git a/acceptance/pipelines/test/run-control/resume/test.toml b/acceptance/pipelines/test/run-control/resume/test.toml new file mode 100644 index 00000000000..fe809c2f5fb --- /dev/null +++ b/acceptance/pipelines/test/run-control/resume/test.toml @@ -0,0 +1,13 @@ +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/updates/{update_id}" +Response.Body = '{"update": {"update_id": "00000000-0000-0000-0000-0000000000a1", "state": "COMPLETED", "test_only": true}}' + +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/events" +Response.Body = ''' +{"events": [ + {"id": "e1", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:00.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_valid_order"}, "details": {"test_case_progress": {"status": "PENDING"}}}, + {"id": "e2", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:01.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_valid_order"}, "details": {"test_case_progress": {"status": "RUNNING"}}}, + {"id": "e3", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:02.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_valid_order"}, "details": {"test_case_progress": {"status": "PASSED"}}}, + {"id": "s1", "event_type": "test_run_summary", "timestamp": "2026-01-01T00:00:03.000Z", "details": {"test_run_summary": {"total": 1, "passed": 1, "failed": 0, "skipped": 0, "errored": 0}}}]} +''' diff --git a/acceptance/pipelines/test/run-control/script.prepare b/acceptance/pipelines/test/run-control/script.prepare new file mode 100644 index 00000000000..caaa90ba318 --- /dev/null +++ b/acceptance/pipelines/test/run-control/script.prepare @@ -0,0 +1,3 @@ +# Each case stubs Start Update, Get Update and List Pipeline Events in its test.toml. +pipeline_id=$($CLI pipelines create --json '{"name": "orders", "root_path": "/Workspace/Users/someone@example.com/orders"}' | jq -r .pipeline_id) +rm "$OUT_REQUESTS" diff --git a/acceptance/pipelines/test/run-control/timeout/out.test.toml b/acceptance/pipelines/test/run-control/timeout/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/run-control/timeout/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/run-control/timeout/output.txt b/acceptance/pipelines/test/run-control/timeout/output.txt new file mode 100644 index 00000000000..5b9ea594437 --- /dev/null +++ b/acceptance/pipelines/test/run-control/timeout/output.txt @@ -0,0 +1,7 @@ + +=== --timeout stops waiting but not the update, and prints the command that reads the results +>>> musterr [CLI] pipelines test --pipeline-id [UUID][0] --resume [UUID][1] --timeout 1s +Pipeline: [UUID][0] +Update: [UUID][1] +Running tests... +Error: stopped waiting for update [UUID][1] after 1s; the update keeps running, to get its results run: databricks pipelines test --pipeline-id [UUID][0] --resume [UUID][1] diff --git a/acceptance/pipelines/test/run-control/timeout/script b/acceptance/pipelines/test/run-control/timeout/script new file mode 100644 index 00000000000..e0703287af5 --- /dev/null +++ b/acceptance/pipelines/test/run-control/timeout/script @@ -0,0 +1,2 @@ +title "--timeout stops waiting but not the update, and prints the command that reads the results" +trace musterr $CLI pipelines test --pipeline-id $pipeline_id --resume 00000000-0000-0000-0000-0000000000a1 --timeout 1s diff --git a/acceptance/pipelines/test/run-control/timeout/test.toml b/acceptance/pipelines/test/run-control/timeout/test.toml new file mode 100644 index 00000000000..4b2e7edef64 --- /dev/null +++ b/acceptance/pipelines/test/run-control/timeout/test.toml @@ -0,0 +1,10 @@ +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/updates/{update_id}" +Response.Body = '{"update": {"update_id": "00000000-0000-0000-0000-0000000000a1", "state": "RUNNING", "test_only": true}}' + +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/events" +Response.Body = ''' +{"events": [ + {"id": "e1", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:00.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_slow"}, "details": {"test_case_progress": {"status": "PENDING"}}}]} +''' diff --git a/cmd/pipelines/test.go b/cmd/pipelines/test.go index cd3048f5756..ebecee1f3fb 100644 --- a/cmd/pipelines/test.go +++ b/cmd/pipelines/test.go @@ -9,10 +9,12 @@ import ( "maps" "net/http" "os" + "os/signal" "path" "path/filepath" "slices" "strings" + "syscall" "time" "uuid" @@ -219,6 +221,31 @@ func getRootPath(ctx context.Context, w *databricks.WorkspaceClient, pipelineID return pipeline.Spec.RootPath, nil } +// updateTestOnly holds UpdateInfo.test_only from a Get Update response; the Go SDK does not have it yet. +// The service leaves it out where it cannot report it, such as before pipeline testing is enabled. +type updateTestOnly struct { + Update struct { + TestOnly *bool `json:"test_only"` + } `json:"update"` +} + +// checkTestUpdate returns an error if the service reports that the update is not a test-only update. +func checkTestUpdate(ctx context.Context, w *databricks.WorkspaceClient, pipelineID, updateID string) error { + apiClient, err := client.New(w.Config) + if err != nil { + return fmt.Errorf("failed to create API client: %w", err) + } + var response updateTestOnly + err = apiClient.Do(ctx, http.MethodGet, fmt.Sprintf("/api/2.0/pipelines/%s/updates/%s", pipelineID, updateID), auth.WorkspaceIDHeaders(w.Config), nil, nil, &response) + if err != nil { + return err + } + if testOnly := response.Update.TestOnly; testOnly != nil && !*testOnly { + return fmt.Errorf("update %s is not a test update; pass the update ID that pipelines test printed", updateID) + } + return nil +} + // startTestUpdate starts a test-only update and returns its ID. func startTestUpdate(ctx context.Context, w *databricks.WorkspaceClient, pipelineID string, req startTestUpdateRequest) (string, error) { apiClient, err := client.New(w.Config) @@ -297,7 +324,7 @@ func resolveTestTargetByID(cmd *cobra.Command, args []string, pipelineID string, } // resolveTestTargetInBundle identifies the pipeline by its bundle KEY, which must precede any selections. -func resolveTestTargetInBundle(cmd *cobra.Command, args []string) (testTarget, error) { +func resolveTestTargetInBundle(cmd *cobra.Command, args []string, resume bool) (testTarget, error) { var key string b, err := utils.ProcessBundle(cmd, utils.ProcessOptions{ PostInitFunc: func(ctx context.Context, b *bundle.Bundle) error { @@ -320,7 +347,9 @@ func resolveTestTargetInBundle(cmd *cobra.Command, args []string) (testTarget, e } return testTarget{}, err } - suggestPipelineDeploy(ctx, cmd) + if !resume { + suggestPipelineDeploy(ctx, cmd) + } cwd, err := os.Getwd() if err != nil { @@ -340,12 +369,60 @@ func resolveTestTargetInBundle(cmd *cobra.Command, args []string) (testTarget, e return testTarget{key: key, pipelineID: pipelineID, selections: selections, w: b.WorkspaceClient(ctx), projectFolder: filesRoot}, nil } +// resumeCommand returns the command that reads the result of the update later. It keeps the +// options that select the pipeline and its workspace, quoted so that it can be pasted into a shell. +func (t testTarget) resumeCommand(cmd *cobra.Command, updateID string) string { + parts := []string{cmd.CommandPath()} + if t.key != "" { + parts = append(parts, shellquote.BashArg(t.key)) + } else { + parts = append(parts, "--pipeline-id", shellquote.BashArg(t.pipelineID)) + } + if target := cmd.Flag("target").Value.String(); target != "" { + parts = append(parts, "--target", shellquote.BashArg(target)) + } + if profile := cmd.Flag("profile").Value.String(); profile != "" { + parts = append(parts, "--profile", shellquote.BashArg(profile)) + } + // --resume loads the bundle again, so it needs the same variable values. + if flag := cmd.Flag("var"); flag != nil { + if values, ok := flag.Value.(pflag.SliceValue); ok { + for _, v := range values.GetSlice() { + parts = append(parts, "--var", shellquote.BashArg(v)) + } + } + } + return strings.Join(append(parts, "--resume", updateID), " ") +} + +// notifyTestInterrupt returns a context that the first Ctrl+C or SIGTERM cancels; the CLI root +// installs no signal handler of its own. It stops listening on that first signal, so that a second +// one ends the process, as notifyInterrupt in cmd/air/run.go does. The caller must call stop. +func notifyTestInterrupt(parent context.Context) (ctx context.Context, stop context.CancelFunc) { + ctx, cancel := context.WithCancel(parent) + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM) + go func() { + select { + case <-sigCh: + signal.Stop(sigCh) + cancel() + case <-ctx.Done(): + signal.Stop(sigCh) + } + }() + return ctx, cancel +} + +var errTestWaitTimeout = errors.New("test wait timed out") + const testPollInterval = time.Second // waitForTestUpdate polls the update until it ends, prints each test case as it finishes, and // returns the final state. It takes the steps of the bundle pipeline runner's wait loop -// (bundle/run/pipeline.go), which cannot be reused because it starts a regular update. -func waitForTestUpdate(ctx context.Context, w *databricks.WorkspaceClient, pipelineID, updateID string, run *testRun, printer *testProgressPrinter) (pipelines.UpdateInfoState, error) { +// (bundle/run/pipeline.go), which cannot be reused because it starts a regular update. It returns +// errTestWaitTimeout when timedOut fires; a nil timedOut never fires. +func waitForTestUpdate(ctx context.Context, w *databricks.WorkspaceClient, pipelineID, updateID string, run *testRun, printer *testProgressPrinter, timedOut <-chan time.Time) (pipelines.UpdateInfoState, error) { lastTimestamp := "" for { response, err := w.Pipelines.GetUpdateByPipelineIdAndUpdateId(ctx, pipelineID, updateID) @@ -383,6 +460,8 @@ func waitForTestUpdate(ctx context.Context, w *databricks.WorkspaceClient, pipel select { case <-ctx.Done(): return "", ctx.Err() + case <-timedOut: + return "", errTestWaitTimeout case <-time.After(testPollInterval): } } @@ -474,17 +553,24 @@ example: The command waits for the results. It exits with code 0 only if at least one test was collected and none failed or errored, so a run whose tests were all skipped succeeds; otherwise it exits with code 1. --no-wait returns after the -update starts. To stop the update, run 'pipelines stop'.`, +update starts. + +--timeout and Ctrl+C stop waiting but not the update; pass its ID to --resume +to read the results later. To stop the update, run 'pipelines stop'.`, } var pipelineID string var pytestArgs []string var noWait bool + var timeout time.Duration + var resumeUpdateID string testGroup := cmdgroup.NewFlagGroup("Pipeline Test") testGroup.FlagSet().StringVar(&pipelineID, "pipeline-id", "", "ID of the pipeline to test, instead of a KEY.") testGroup.FlagSet().StringArrayVar(&pytestArgs, "pytest-arg", nil, "Argument to pass to pytest (repeatable). Only -k and -m are supported.") testGroup.FlagSet().BoolVar(&noWait, "no-wait", false, "Don't wait for the tests to complete.") + testGroup.FlagSet().DurationVar(&timeout, "timeout", 0, "Stop waiting for results after this duration, for example 30m. The update keeps running.") + testGroup.FlagSet().StringVar(&resumeUpdateID, "resume", "", "Read the results of an existing test update instead of starting one.") wrappedCmd := cmdgroup.NewCommandWithGroupFlag(cmd) wrappedCmd.AddFlagGroup(testGroup) @@ -504,6 +590,31 @@ update starts. To stop the update, run 'pipelines stop'.`, if err := validatePytestArgs(pytestArgs); err != nil { return err } + if timeout < 0 { + return errors.New("--timeout must not be negative") + } + if noWait && timeout > 0 { + return errors.New("--timeout cannot be used with --no-wait") + } + // An empty --resume, such as from an unset shell variable, must not start a new run. + resume := cmd.Flags().Changed("resume") + if resume { + selectionCount := len(args) + if pipelineID == "" { + selectionCount = max(len(args)-1, 0) + } + switch { + case noWait: + return errors.New("--resume cannot be used with --no-wait") + case len(pytestArgs) > 0: + return errors.New("--resume cannot be used with --pytest-arg") + case selectionCount > 0: + return errors.New("--resume cannot be used with test selections") + } + if _, err := uuid.Parse(resumeUpdateID); err != nil { + return fmt.Errorf("--resume %q is not an update ID; pass the update ID that pipelines test printed", resumeUpdateID) + } + } var target testTarget if pipelineID != "" { id, err := uuid.Parse(pipelineID) @@ -517,16 +628,28 @@ update starts. To stop the update, run 'pipelines stop'.`, } else { cmd.SetContext(logdiag.InitContext(cmd.Context())) var err error - target, err = resolveTestTargetInBundle(cmd, args) + target, err = resolveTestTargetInBundle(cmd, args, resume) if err != nil { return err } } ctx := cmd.Context() - req := newStartTestUpdateRequest(target.selections, pytestArgs) - updateID, err := startTestUpdate(ctx, target.w, target.pipelineID, req) - if err != nil { + // The handler is installed before the update starts, but only the wait listens to it: a + // Ctrl+C during Start Update must not cancel a request that the service may accept, or the + // update would run without its ID being printed. + waitCtx, stop := notifyTestInterrupt(ctx) + defer stop() + + updateID := resumeUpdateID + if !resume { + req := newStartTestUpdateRequest(target.selections, pytestArgs) + var err error + updateID, err = startTestUpdate(ctx, target.w, target.pipelineID, req) + if err != nil { + return err + } + } else if err := checkTestUpdate(ctx, target.w, target.pipelineID, updateID); err != nil { return err } out := cmd.OutOrStdout() @@ -537,9 +660,23 @@ update starts. To stop the update, run 'pipelines stop'.`, run := newTestRun() printer := &testProgressPrinter{out: out, shortID: target.shortTestID} - state, err := waitForTestUpdate(ctx, target.w, target.pipelineID, updateID, run, printer) + // The timeout is not a context deadline: the SDK rate limiter fails a request that would wait + // past the deadline before the context expires, with an error that does not name the timeout. + var timedOut <-chan time.Time + if timeout > 0 { + timedOut = time.After(timeout) + } + state, err := waitForTestUpdate(waitCtx, target.w, target.pipelineID, updateID, run, printer, timedOut) if err != nil { - return err + isTimeout := errors.Is(err, errTestWaitTimeout) + if !isTimeout && waitCtx.Err() == nil { + return err + } + resumeCmd := target.resumeCommand(cmd, updateID) + if isTimeout { + return fmt.Errorf("stopped waiting for update %s after %s; the update keeps running, to get its results run: %s", updateID, timeout, resumeCmd) + } + return fmt.Errorf("stopped waiting for update %s; the update keeps running, to get its results run: %s", updateID, resumeCmd) } incompleteReason := run.incompleteReason(state) renderTestResult(out, run, incompleteReason, target.shortTestID) diff --git a/cmd/pipelines/test_signal_test.go b/cmd/pipelines/test_signal_test.go new file mode 100644 index 00000000000..78c6c731e3e --- /dev/null +++ b/cmd/pipelines/test_signal_test.go @@ -0,0 +1,39 @@ +//go:build linux || darwin + +package pipelines + +import ( + "syscall" + "testing" + "time" + + "github.com/databricks/cli/libs/testserver" + databricks "github.com/databricks/databricks-sdk-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A Ctrl+C while Start Update is in flight stops the wait, but not the request: the service may +// have accepted the update, so the command must still receive and print its ID. +func TestInterruptDuringStartUpdateKeepsTheUpdateID(t *testing.T) { + server := testserver.New(t) + server.Handle("POST", "/api/2.0/pipelines/{pipeline_id}/updates", func(req testserver.Request) any { + require.NoError(t, syscall.Kill(syscall.Getpid(), syscall.SIGINT)) + time.Sleep(200 * time.Millisecond) + return map[string]string{"update_id": "u1"} + }) + w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "token"}) + require.NoError(t, err) + + waitCtx, stop := notifyTestInterrupt(t.Context()) + defer stop() + updateID, err := startTestUpdate(t.Context(), w, "p1", newStartTestUpdateRequest(nil, nil)) + require.NoError(t, err) + assert.Equal(t, "u1", updateID) + + select { + case <-waitCtx.Done(): + case <-time.After(5 * time.Second): + t.Fatal("the interrupt did not stop the wait") + } +} diff --git a/cmd/pipelines/test_test.go b/cmd/pipelines/test_test.go index 7d160a50ec5..d01259b56f7 100644 --- a/cmd/pipelines/test_test.go +++ b/cmd/pipelines/test_test.go @@ -3,10 +3,13 @@ package pipelines import ( "bytes" "errors" + "net/http" + "net/http/httptest" "os" "path/filepath" "strings" "testing" + "time" "github.com/databricks/cli/bundle" "github.com/databricks/cli/bundle/config" @@ -280,7 +283,7 @@ func TestWaitForTestUpdate(t *testing.T) { var out bytes.Buffer run := newTestRun() printer := &testProgressPrinter{out: &out, shortID: func(id string) string { return relativeTestID("/Workspace/p", id) }} - state, err := waitForTestUpdate(t.Context(), w, "p1", "u1", run, printer) + state, err := waitForTestUpdate(t.Context(), w, "p1", "u1", run, printer, nil) require.NoError(t, err) assert.Equal(t, pipelines.UpdateInfoStateCompleted, state) @@ -293,3 +296,84 @@ func TestWaitForTestUpdate(t *testing.T) { assert.Equal(t, testRunSummary{Total: 2, Passed: 2}, run.counts()) assert.Contains(t, out.String(), "2 of 2 PASSED tests/test_a.py::test_b") } + +// The timeout ends the wait between polls of an update that is still running. +func TestWaitForTestUpdateTimeout(t *testing.T) { + server := testserver.New(t) + server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/updates/{update_id}", func(req testserver.Request) any { + return map[string]any{"update": map[string]any{"update_id": "u1", "state": "RUNNING"}} + }) + server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req testserver.Request) any { + return map[string]any{"events": []any{}} + }) + + w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "token"}) + require.NoError(t, err) + timedOut := make(chan time.Time, 1) + timedOut <- time.Time{} + printer := &testProgressPrinter{out: &bytes.Buffer{}, shortID: func(id string) string { return id }} + _, err = waitForTestUpdate(t.Context(), w, "p1", "u1", newTestRun(), printer, timedOut) + assert.ErrorIs(t, err, errTestWaitTimeout) +} + +func TestTestTargetResumeCommand(t *testing.T) { + tests := []struct { + name string + target testTarget + flags map[string]string + want string + }{ + { + name: "names that need quoting", + target: testTarget{key: "my_pipeline", pipelineID: "p1"}, + flags: map[string]string{"target": "dev west", "profile": "CI West"}, + want: "databricks pipelines test my_pipeline --target 'dev west' --profile 'CI West' --resume u1", + }, + { + name: "bundle variables, which --resume needs to load the bundle again", + target: testTarget{key: "my_pipeline", pipelineID: "p1"}, + flags: map[string]string{"target": "dev", "var": "catalog=orders,schema=dev team"}, + want: "databricks pipelines test my_pipeline --target dev --var 'catalog=orders' --var 'schema=dev team' --resume u1", + }, + { + name: "pipeline ID with profile", + target: testTarget{pipelineID: "p1"}, + flags: map[string]string{"profile": "staging"}, + want: "databricks pipelines test --pipeline-id p1 --profile staging --resume u1", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + rootCmd := &cobra.Command{Use: "databricks"} + rootCmd.PersistentFlags().StringP("profile", "p", "", "") + rootCmd.PersistentFlags().StringP("target", "t", "", "") + rootCmd.PersistentFlags().StringSlice("var", nil, "") + pipelinesCmd := &cobra.Command{Use: "pipelines"} + cmd := &cobra.Command{Use: "test"} + rootCmd.AddCommand(pipelinesCmd) + pipelinesCmd.AddCommand(cmd) + for name, value := range tc.flags { + require.NoError(t, cmd.Flag(name).Value.Set(value)) + } + assert.Equal(t, tc.want, tc.target.resumeCommand(cmd, "u1")) + }) + } +} + +// Where the service cannot report test_only, as before pipeline testing is enabled, the update +// may be a test update, so the command reads its results instead of rejecting it. +func TestCheckTestUpdateWithoutTestOnly(t *testing.T) { + const updateID = "11111111-1111-1111-1111-111111111111" + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/api/2.0/pipelines/p1/updates/"+updateID { + _, _ = w.Write([]byte(`{"update": {"state": "COMPLETED"}}`)) + return + } + _, _ = w.Write([]byte(`{}`)) + })) + t.Cleanup(srv.Close) + w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: srv.URL, Token: "token"}) + require.NoError(t, err) + + assert.NoError(t, checkTestUpdate(t.Context(), w, "p1", updateID)) +}