diff --git a/.nextchanges/cli/pipelines-test.md b/.nextchanges/cli/pipelines-test.md new file mode 100644 index 00000000000..675b38ccc11 --- /dev/null +++ b/.nextchanges/cli/pipelines-test.md @@ -0,0 +1 @@ +* Add the `databricks pipelines test` command, which runs a pipeline's tests in a test-only update. ([#7297](https://github.com/databricks/cli/pull/7297)) diff --git a/acceptance/pipelines/databricks-cli-help/output.txt b/acceptance/pipelines/databricks-cli-help/output.txt index 8514fa33b2b..adc729f4f5c 100644 --- a/acceptance/pipelines/databricks-cli-help/output.txt +++ b/acceptance/pipelines/databricks-cli-help/output.txt @@ -30,6 +30,7 @@ Main Commands run Run a pipeline show Preview a table's columns and sample rows stop Stop a pipeline + test *Preview* Run the tests of a pipeline Management Commands apply-environment *Public Preview* Apply the latest environment to the pipeline. diff --git a/acceptance/pipelines/test/arguments/out.test.toml b/acceptance/pipelines/test/arguments/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/arguments/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/arguments/output.txt b/acceptance/pipelines/test/arguments/output.txt new file mode 100644 index 00000000000..ed845172e52 --- /dev/null +++ b/acceptance/pipelines/test/arguments/output.txt @@ -0,0 +1,60 @@ + +>>> [CLI] pipelines test --help +Run the tests of a pipeline in a test-only update. + +This command is experimental. Set DATABRICKS_CLI_PIPELINES_TEST=true to enable it. + +PIPELINE_KEY is the unique name of the pipeline, as defined in its YAML file. +It is required when test selections are given. If there is only one pipeline +in the project and no arguments are given, it is selected automatically. +Outside a project, identify the pipeline with --pipeline-id; all positional +arguments are then test selections. + +TEST_SELECTION is a test directory, file, or pytest test ID, for example +tests/test_orders.py::test_backfill. In a project, a relative path is a local +path, relative to the working directory, and is converted to its deployed +workspace path. With --pipeline-id, a relative path is relative to the +pipeline's root_path. Paths that start with / are workspace paths. Without a +selection, all of the pipeline's tests run. + +Each --pytest-arg passes one argument to pytest, in the order given. Only -k +and -m are supported, each followed by its value in the next --pytest-arg, for +example: + + databricks pipelines test my_pipeline --pytest-arg=-k --pytest-arg='orders and not slow' + +Usage: + databricks pipelines test [PIPELINE_KEY] [TEST_SELECTION...] [flags] [--pytest-arg=ARG ...] + +Pipeline Test Flags: + --pipeline-id string ID of the pipeline to test, instead of a PIPELINE_KEY. + --pytest-arg stringArray Argument to pass to pytest (repeatable). Only -k and -m are supported. + +Flags: + -h, --help help for test + +Global Flags: + --debug enable debug logging + -o, --output type output type: text or json (default text) + -p, --profile string ~/.databrickscfg profile + -t, --target string bundle target to use (if applicable) + --var strings set values for variables defined in project config. Example: --var="foo=bar" + +=== The command is experimental and off by default +>>> musterr [CLI] pipelines test --pipeline-id 1234 +Error: pipelines test is experimental. Set DATABRICKS_CLI_PIPELINES_TEST=true to enable it + +=== pytest arguments must be passed with --pytest-arg +>>> musterr [CLI] pipelines test --pipeline-id 1234 -k orders +Error: -k is a pytest argument: pass pytest arguments with --pytest-arg, for example --pytest-arg=-k --pytest-arg='orders and not slow' + +>>> musterr [CLI] pipelines test --pipeline-id 1234 -- -k orders +Error: the -- separator is not supported: pass pytest arguments with --pytest-arg, for example --pytest-arg=-k --pytest-arg='orders and not slow' + +=== Invalid --pytest-arg values fail before any request +>>> musterr [CLI] pipelines test --pipeline-id 1234 --pytest-arg=-x +Error: unsupported pytest argument "-x": only -k and -m are supported + +=== JSON output is not supported +>>> musterr [CLI] pipelines test --pipeline-id 1234 -o json +Error: --output json is not supported by pipelines test diff --git a/acceptance/pipelines/test/arguments/script b/acceptance/pipelines/test/arguments/script new file mode 100644 index 00000000000..d027c13dcb8 --- /dev/null +++ b/acceptance/pipelines/test/arguments/script @@ -0,0 +1,14 @@ +trace $CLI pipelines test --help + +title "The command is experimental and off by default" +(unset DATABRICKS_CLI_PIPELINES_TEST; trace musterr $CLI pipelines test --pipeline-id 1234) + +title "pytest arguments must be passed with --pytest-arg" +trace musterr $CLI pipelines test --pipeline-id 1234 -k orders +trace musterr $CLI pipelines test --pipeline-id 1234 -- -k orders + +title "Invalid --pytest-arg values fail before any request" +trace musterr $CLI pipelines test --pipeline-id 1234 --pytest-arg=-x + +title "JSON output is not supported" +trace musterr $CLI pipelines test --pipeline-id 1234 -o json diff --git a/acceptance/pipelines/test/immutable/databricks.yml b/acceptance/pipelines/test/immutable/databricks.yml new file mode 100644 index 00000000000..483234d0dd4 --- /dev/null +++ b/acceptance/pipelines/test/immutable/databricks.yml @@ -0,0 +1,13 @@ +bundle: + name: test-pipeline-test-immutable + +workspace: + sync_mode: immutable + +resources: + pipelines: + my_pipeline: + name: test-pipeline-test-immutable + libraries: + - file: + path: pipeline_file.py diff --git a/acceptance/pipelines/test/immutable/out.test.toml b/acceptance/pipelines/test/immutable/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/immutable/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/immutable/output.txt b/acceptance/pipelines/test/immutable/output.txt new file mode 100644 index 00000000000..f6c4a58a45b --- /dev/null +++ b/acceptance/pipelines/test/immutable/output.txt @@ -0,0 +1,53 @@ + +>>> [CLI] pipelines deploy +Created internal_immutable_snapshots.immutable +Created pipelines.my_pipeline +Files: 0 uploaded, 0 deleted +Resources: 2 created, 0 changed, 0 deleted, 0 unchanged +View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?w=[WORKSPACE_ID] + +=== Local test paths are not supported with workspace.sync_mode immutable +>>> musterr [CLI] pipelines test my_pipeline tests/test_orders.py::test_backfill +Recommendation: This command runs the last deployed version of the code + +If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included. + +Error: cannot resolve test selection "tests/test_orders.py::test_backfill": local test paths are not supported with workspace.sync_mode immutable; pass an absolute workspace path + +=== Absolute workspace paths and running all tests still work +>>> MSYS_NO_PATHCONV=1 [CLI] pipelines test my_pipeline /Workspace/Users/someone@example.com/tests/test_orders.py::test_backfill +Recommendation: This command runs the last deployed version of the code + +If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included. + +Pipeline: my_pipeline ([UUID]) +Update: [UUID] + +>>> [CLI] pipelines test my_pipeline +Recommendation: This command runs the last deployed version of the code + +If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included. + +Pipeline: my_pipeline ([UUID]) +Update: [UUID] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "test_only": true, + "test_options": { + "test_selection": [ + "/Workspace/Users/someone@example.com/tests/test_orders.py::test_backfill" + ] + } + } +} +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "test_only": true + } +} diff --git a/acceptance/pipelines/test/immutable/pipeline_file.py b/acceptance/pipelines/test/immutable/pipeline_file.py new file mode 100644 index 00000000000..4a7a3a1878e --- /dev/null +++ b/acceptance/pipelines/test/immutable/pipeline_file.py @@ -0,0 +1,6 @@ +from pyspark import pipelines as dp + + +@dp.table +def orders(): + return spark.range(10) diff --git a/acceptance/pipelines/test/immutable/script b/acceptance/pipelines/test/immutable/script new file mode 100644 index 00000000000..8cd43738c35 --- /dev/null +++ b/acceptance/pipelines/test/immutable/script @@ -0,0 +1,12 @@ +trace $CLI pipelines deploy +rm "$OUT_REQUESTS" + +title "Local test paths are not supported with workspace.sync_mode immutable" +trace musterr $CLI pipelines test my_pipeline tests/test_orders.py::test_backfill + +title "Absolute workspace paths and running all tests still work" +trace MSYS_NO_PATHCONV=1 $CLI pipelines test my_pipeline /Workspace/Users/someone@example.com/tests/test_orders.py::test_backfill +# Like the service, the test server rejects a new update while one is active. +$CLI pipelines stop my_pipeline >> LOG.stop 2>&1 +trace $CLI pipelines test my_pipeline +trace print_requests.py //updates diff --git a/acceptance/pipelines/test/immutable/tests/test_orders.py b/acceptance/pipelines/test/immutable/tests/test_orders.py new file mode 100644 index 00000000000..ec31eaa1b01 --- /dev/null +++ b/acceptance/pipelines/test/immutable/tests/test_orders.py @@ -0,0 +1,2 @@ +def test_backfill(): + assert True diff --git a/acceptance/pipelines/test/pipeline-id/out.test.toml b/acceptance/pipelines/test/pipeline-id/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/pipeline-id/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/pipeline-id/output.txt b/acceptance/pipelines/test/pipeline-id/output.txt new file mode 100644 index 00000000000..36766e98fa8 --- /dev/null +++ b/acceptance/pipelines/test/pipeline-id/output.txt @@ -0,0 +1,46 @@ + +=== Relative test selections resolve against the pipeline root_path +>>> MSYS_NO_PATHCONV=1 [CLI] pipelines test --pipeline-id [UUID] tests/test_orders.py::test_backfill /Workspace/Users/someone@example.com/other/tests +Pipeline: [UUID] +Update: [UUID] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "test_only": true, + "test_options": { + "test_selection": [ + "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_backfill", + "/Workspace/Users/someone@example.com/other/tests" + ] + } + } +} + +=== Without a root_path, test selections must be workspace paths +>>> musterr [CLI] pipelines test --pipeline-id [UUID] tests +Error: cannot resolve test selection "tests": the pipeline has no root_path; use an absolute workspace path such as /Workspace/Users/someone@example.com/tests + +>>> MSYS_NO_PATHCONV=1 [CLI] pipelines test --pipeline-id [UUID] /Workspace/Users/someone@example.com/tests/test_other.py::test_x +Pipeline: [UUID] +Update: [UUID] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "test_only": true, + "test_options": { + "test_selection": [ + "/Workspace/Users/someone@example.com/tests/test_other.py::test_x" + ] + } + } +} + +=== Outside a project, --pipeline-id is required +>>> musterr [CLI] pipelines test tests +Error: outside a project, identify the pipeline with --pipeline-id diff --git a/acceptance/pipelines/test/pipeline-id/script b/acceptance/pipelines/test/pipeline-id/script new file mode 100644 index 00000000000..3745fe70c50 --- /dev/null +++ b/acceptance/pipelines/test/pipeline-id/script @@ -0,0 +1,15 @@ +pipeline_id=$($CLI pipelines create --json '{"name": "orders", "root_path": "/Workspace/Users/someone@example.com/orders"}' | jq -r .pipeline_id) +no_root_pipeline_id=$($CLI pipelines create --json '{"name": "no-root"}' | jq -r .pipeline_id) +rm "$OUT_REQUESTS" + +title "Relative test selections resolve against the pipeline root_path" +trace MSYS_NO_PATHCONV=1 $CLI pipelines test --pipeline-id $pipeline_id tests/test_orders.py::test_backfill /Workspace/Users/someone@example.com/other/tests +trace print_requests.py //updates + +title "Without a root_path, test selections must be workspace paths" +trace musterr $CLI pipelines test --pipeline-id $no_root_pipeline_id tests +trace MSYS_NO_PATHCONV=1 $CLI pipelines test --pipeline-id $no_root_pipeline_id /Workspace/Users/someone@example.com/tests/test_other.py::test_x +trace print_requests.py //updates + +title "Outside a project, --pipeline-id is required" +trace musterr $CLI pipelines test tests diff --git a/acceptance/pipelines/test/selection/databricks.yml b/acceptance/pipelines/test/selection/databricks.yml new file mode 100644 index 00000000000..0cf849a7919 --- /dev/null +++ b/acceptance/pipelines/test/selection/databricks.yml @@ -0,0 +1,10 @@ +bundle: + name: test-pipeline-test-selection + +resources: + pipelines: + my_pipeline: + name: test-pipeline-test-selection + libraries: + - file: + path: pipeline_file.py diff --git a/acceptance/pipelines/test/selection/out.test.toml b/acceptance/pipelines/test/selection/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/selection/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/selection/output.txt b/acceptance/pipelines/test/selection/output.txt new file mode 100644 index 00000000000..ce05fda9bb1 --- /dev/null +++ b/acceptance/pipelines/test/selection/output.txt @@ -0,0 +1,83 @@ + +>>> [CLI] pipelines deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-pipeline-test-selection/default/files... +Created pipelines.my_pipeline +Files: 3 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged +View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID]?w=[WORKSPACE_ID] + +=== Run all tests of the only pipeline +>>> [CLI] pipelines test +Recommendation: This command runs the last deployed version of the code + +If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included. + +Pipeline: my_pipeline ([UUID]) +Update: [UUID] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "test_only": true + } +} + +=== Select tests by relative path and test ID, with pytest arguments in both --pytest-arg forms +>>> MSYS_NO_PATHCONV=1 [CLI] pipelines test my_pipeline tests tests/test_orders.py::test_backfill /Workspace/Users/someone@example.com/tests/test_other.py --pytest-arg=-k --pytest-arg=orders and not slow --pytest-arg -m --pytest-arg smoke,fast +Recommendation: This command runs the last deployed version of the code + +If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included. + +Pipeline: my_pipeline ([UUID]) +Update: [UUID] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "test_only": true, + "test_options": { + "test_selection": [ + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-test-selection/default/files/tests", + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-test-selection/default/files/tests/test_orders.py::test_backfill", + "/Workspace/Users/someone@example.com/tests/test_other.py" + ], + "pytest_args": [ + "-k", + "orders and not slow", + "-m", + "smoke,fast" + ] + } + } +} + +=== Relative paths resolve from the working directory +>>> [CLI] pipelines test my_pipeline test_orders.py::test_valid_order +Recommendation: This command runs the last deployed version of the code + +If you've made local changes, run 'databricks pipelines deploy' first to ensure they are included. + +Pipeline: my_pipeline ([UUID]) +Update: [UUID] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID]/updates", + "body": { + "test_only": true, + "test_options": { + "test_selection": [ + "/Workspace/Users/[USERNAME]/.bundle/test-pipeline-test-selection/default/files/tests/test_orders.py::test_valid_order" + ] + } + } +} + +=== Test selections require the pipeline KEY first +>>> musterr [CLI] pipelines test tests/test_orders.py +Error: resource with key "tests/test_orders.py" not found; the first argument must be a pipeline KEY when test selections are given diff --git a/acceptance/pipelines/test/selection/pipeline_file.py b/acceptance/pipelines/test/selection/pipeline_file.py new file mode 100644 index 00000000000..4a7a3a1878e --- /dev/null +++ b/acceptance/pipelines/test/selection/pipeline_file.py @@ -0,0 +1,6 @@ +from pyspark import pipelines as dp + + +@dp.table +def orders(): + return spark.range(10) diff --git a/acceptance/pipelines/test/selection/script b/acceptance/pipelines/test/selection/script new file mode 100644 index 00000000000..18857609443 --- /dev/null +++ b/acceptance/pipelines/test/selection/script @@ -0,0 +1,24 @@ +trace $CLI pipelines deploy +rm "$OUT_REQUESTS" + +title "Run all tests of the only pipeline" +trace $CLI pipelines test +trace print_requests.py //updates +# Like the service, the test server rejects a new update while one is active. +$CLI pipelines stop my_pipeline >> LOG.stop 2>&1 +rm "$OUT_REQUESTS" + +title "Select tests by relative path and test ID, with pytest arguments in both --pytest-arg forms" +trace MSYS_NO_PATHCONV=1 $CLI pipelines test my_pipeline tests tests/test_orders.py::test_backfill /Workspace/Users/someone@example.com/tests/test_other.py --pytest-arg=-k --pytest-arg='orders and not slow' --pytest-arg -m --pytest-arg smoke,fast +trace print_requests.py //updates +$CLI pipelines stop my_pipeline >> LOG.stop 2>&1 +rm "$OUT_REQUESTS" + +title "Relative paths resolve from the working directory" +cd tests +trace $CLI pipelines test my_pipeline test_orders.py::test_valid_order +cd .. +trace print_requests.py //updates + +title "Test selections require the pipeline KEY first" +trace musterr $CLI pipelines test tests/test_orders.py diff --git a/acceptance/pipelines/test/selection/tests/test_orders.py b/acceptance/pipelines/test/selection/tests/test_orders.py new file mode 100644 index 00000000000..6058cdd0bba --- /dev/null +++ b/acceptance/pipelines/test/selection/tests/test_orders.py @@ -0,0 +1,6 @@ +def test_valid_order(): + assert True + + +def test_backfill(): + assert True diff --git a/acceptance/pipelines/test/submit-error/out.test.toml b/acceptance/pipelines/test/submit-error/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/submit-error/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/submit-error/output.txt b/acceptance/pipelines/test/submit-error/output.txt new file mode 100644 index 00000000000..524700892b2 --- /dev/null +++ b/acceptance/pipelines/test/submit-error/output.txt @@ -0,0 +1,13 @@ + +=== A submit error from the service is shown as is, and no update ID is printed +>>> musterr [CLI] pipelines test --pipeline-id 1234 +Error: An active update '[UUID]' already exists for pipeline '1234'. + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/1234/updates", + "body": { + "test_only": true + } +} diff --git a/acceptance/pipelines/test/submit-error/script b/acceptance/pipelines/test/submit-error/script new file mode 100644 index 00000000000..fe472e04fb5 --- /dev/null +++ b/acceptance/pipelines/test/submit-error/script @@ -0,0 +1,3 @@ +title "A submit error from the service is shown as is, and no update ID is printed" +trace musterr $CLI pipelines test --pipeline-id 1234 +trace print_requests.py //updates diff --git a/acceptance/pipelines/test/submit-error/test.toml b/acceptance/pipelines/test/submit-error/test.toml new file mode 100644 index 00000000000..166ee3c82b6 --- /dev/null +++ b/acceptance/pipelines/test/submit-error/test.toml @@ -0,0 +1,9 @@ +[[Server]] +Pattern = "POST /api/2.0/pipelines/{pipeline_id}/updates" +Response.Body = ''' +{ + "error_code": "RESOURCE_CONFLICT", + "message": "An active update '00000000-0000-0000-0000-000000000001' already exists for pipeline '1234'." +} +''' +Response.StatusCode = 409 diff --git a/acceptance/pipelines/test/test.toml b/acceptance/pipelines/test/test.toml new file mode 100644 index 00000000000..66f23523071 --- /dev/null +++ b/acceptance/pipelines/test/test.toml @@ -0,0 +1,4 @@ +RecordRequests = true +Ignore = ['.databricks'] +# The command is experimental; the gate case in arguments turns it off. +Env.DATABRICKS_CLI_PIPELINES_TEST = "true" diff --git a/bundle/root.go b/bundle/root.go index 5f7317a5e99..c41c52edfb7 100644 --- a/bundle/root.go +++ b/bundle/root.go @@ -70,6 +70,13 @@ func mustGetRoot(ctx context.Context) (string, diag.Diagnostics) { return getRootWithTraversal() } +// HasRoot reports whether a bundle root is configured or can be found without loading it. +// An invalid bundle root environment variable counts as a root so that loading reports the error. +func HasRoot(ctx context.Context) bool { + path, diags := tryGetRoot(ctx) + return path != "" || diags.HasError() +} + // tryGetRoot returns a bundle root or an empty string if one cannot be found. func tryGetRoot(ctx context.Context) (string, diag.Diagnostics) { // Note: an invalid value in the environment variable is still an error. diff --git a/bundle/root_test.go b/bundle/root_test.go index 27109f2aeff..20b160b8b40 100644 --- a/bundle/root_test.go +++ b/bundle/root_test.go @@ -110,3 +110,24 @@ func TestRootLookupError(t *testing.T) { require.True(t, diags.HasError()) require.Contains(t, diags[0].Summary, "Unable to locate the bundle root") } + +func TestHasRoot(t *testing.T) { + ctx := t.Context() + + // Have to set then unset to allow the testing package to revert it to its original value. + t.Setenv(env.RootVariable, "") + os.Unsetenv(env.RootVariable) + + dir := t.TempDir() + t.Chdir(dir) + require.False(t, HasRoot(ctx)) + + f, err := os.Create(filepath.Join(dir, config.FileNames[0])) + require.NoError(t, err) + f.Close() + require.True(t, HasRoot(ctx)) + + // An invalid environment variable is reported when the bundle is loaded. + t.Setenv(env.RootVariable, filepath.Join(dir, "doesntexist")) + require.True(t, HasRoot(ctx)) +} diff --git a/cmd/pipelines/commands.go b/cmd/pipelines/commands.go index 6a6b6931313..24882f97864 100644 --- a/cmd/pipelines/commands.go +++ b/cmd/pipelines/commands.go @@ -17,6 +17,7 @@ func Commands() []*cobra.Command { destroyCommand(), runCommand(), dryRunCommand(), + testCommand(), historyCommand(), logsCommand(), openCommand(), diff --git a/cmd/pipelines/test.go b/cmd/pipelines/test.go new file mode 100644 index 00000000000..96295a1d1b2 --- /dev/null +++ b/cmd/pipelines/test.go @@ -0,0 +1,395 @@ +package pipelines + +import ( + "context" + "errors" + "fmt" + "io" + "io/fs" + "net/http" + "os" + "path" + "path/filepath" + "strings" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/resources" + "github.com/databricks/cli/cmd/bundle/utils" + "github.com/databricks/cli/cmd/root" + "github.com/databricks/cli/libs/auth" + "github.com/databricks/cli/libs/cmdctx" + "github.com/databricks/cli/libs/cmdgroup" + "github.com/databricks/cli/libs/env" + "github.com/databricks/cli/libs/flags" + "github.com/databricks/cli/libs/logdiag" + "github.com/databricks/cli/libs/shellquote" + databricks "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/client" + "github.com/databricks/databricks-sdk-go/service/pipelines" + "github.com/spf13/cobra" + "github.com/spf13/pflag" +) + +// startTestUpdateRequest is the Start Update body of a test run. +// test_only and test_options are not in the Go SDK yet, so the request is sent directly. +type startTestUpdateRequest struct { + TestOnly bool `json:"test_only"` + TestOptions *testOptions `json:"test_options,omitempty"` +} + +type testOptions struct { + TestSelection []string `json:"test_selection,omitempty"` + PytestArgs []string `json:"pytest_args,omitempty"` +} + +func newStartTestUpdateRequest(selection, pytestArgs []string) startTestUpdateRequest { + req := startTestUpdateRequest{TestOnly: true} + if len(selection) > 0 || len(pytestArgs) > 0 { + req.TestOptions = &testOptions{TestSelection: selection, PytestArgs: pytestArgs} + } + return req +} + +// pipelinesTestEnvVar enables the experimental pipelines test command. +const pipelinesTestEnvVar = "DATABRICKS_CLI_PIPELINES_TEST" + +const pytestArgHint = "pass pytest arguments with --pytest-arg, for example --pytest-arg=-k --pytest-arg='orders and not slow'" + +// validatePytestArgs accepts only -k and -m, each followed by its value in the next argument, as the API +// requires. The value cannot be empty or start with "-", because pytest reads it as an option. +func validatePytestArgs(args []string) error { + for i := 0; i < len(args); i++ { + arg := args[i] + switch { + case arg == "-k" || arg == "-m": + if i == len(args)-1 { + return fmt.Errorf("pytest argument %s requires a value: pass it in the next --pytest-arg", arg) + } + next := args[i+1] + if strings.TrimSpace(next) == "" { + return fmt.Errorf("pytest argument %s requires a non-empty value", arg) + } + if strings.HasPrefix(next, "-") { + return fmt.Errorf("pytest argument %s requires a value, not %q: pass it in the next --pytest-arg", arg, next) + } + i++ + case strings.HasPrefix(arg, "-k") || strings.HasPrefix(arg, "-m"): + option, value := arg[:2], strings.TrimPrefix(arg[2:], "=") + return fmt.Errorf("pytest argument %q must be split: pass --pytest-arg=%s --pytest-arg=%s", arg, option, shellquote.BashArg(value)) + case !strings.HasPrefix(arg, "-"): + return fmt.Errorf("unsupported pytest argument %q: pass test paths and test IDs as TEST_SELECTION arguments", arg) + default: + return fmt.Errorf("unsupported pytest argument %q: only -k and -m are supported", arg) + } + } + return nil +} + +// testFlagErrorFunc adds guidance when pytest's -k or -m is passed as a CLI flag. +func testFlagErrorFunc(cmd *cobra.Command, err error) error { + if notExist, ok := errors.AsType[*pflag.NotExistError](err); ok { + if name := notExist.GetSpecifiedName(); name == "k" || name == "m" { + return fmt.Errorf("-%s is a pytest argument: %s", name, pytestArgHint) + } + } + return cmd.Parent().FlagErrorFunc()(cmd, err) +} + +// resolveTestSelections converts relative selection paths with toWorkspacePath and keeps +// the pytest test ID after "::" unchanged. Paths that start with "/" are workspace paths. +func resolveTestSelections(selections []string, toWorkspacePath func(string) (string, error)) ([]string, error) { + var resolved []string + for _, selection := range selections { + p, testID, _ := strings.Cut(selection, "::") + if p == "" { + return nil, fmt.Errorf("test selection %q has no path", selection) + } + if strings.HasPrefix(p, "/") { + p = path.Clean(p) + } else { + var err error + p, err = toWorkspacePath(p) + if err != nil { + return nil, fmt.Errorf("cannot resolve test selection %q: %w", selection, err) + } + } + if strings.Contains(selection, "::") { + p += "::" + testID + } + resolved = append(resolved, p) + } + return resolved, nil +} + +// bundleWorkspacePath maps a local path, relative to the working directory, to its deployed +// path under filesRoot, the deployed path of the bundle's synchronized files. +func bundleWorkspacePath(cwd, syncRoot, filesRoot, localPath string) (string, error) { + if !filepath.IsAbs(localPath) { + localPath = filepath.Join(cwd, localPath) + } + rel, err := filepath.Rel(syncRoot, localPath) + if err != nil || !filepath.IsLocal(rel) { + return "", fmt.Errorf("%s is outside the synchronized bundle files in %s", filepath.ToSlash(localPath), filepath.ToSlash(syncRoot)) + } + return path.Join(filesRoot, filepath.ToSlash(rel)), nil +} + +// bundleFilesRoot returns the deployed path of the bundle's synchronized files, as TranslatePaths +// maps resource paths, or "" for an immutable deployment, whose snapshot path the command does not know. +func bundleFilesRoot(b *bundle.Bundle) string { + switch { + case b.SyncMode() == config.SyncModeImmutable: + return "" + case b.IsSourceLinkedDeployment(): + return b.SyncRootPath + default: + return b.Config.Workspace.FilePath + } +} + +// bundleTestSelectionPath maps a local path, relative to the working directory, to its deployed path +// under filesRoot. The path must exist locally, so that a typo fails before compute starts. +func bundleTestSelectionPath(cwd, syncRoot, filesRoot, localPath string) (string, error) { + if filesRoot == "" { + return "", errors.New("local test paths are not supported with workspace.sync_mode immutable; pass an absolute workspace path") + } + workspacePath, err := bundleWorkspacePath(cwd, syncRoot, filesRoot, localPath) + if err != nil { + return "", err + } + if !filepath.IsAbs(localPath) { + localPath = filepath.Join(cwd, localPath) + } + if _, err := os.Stat(localPath); errors.Is(err, fs.ErrNotExist) { + return "", fmt.Errorf("%s does not exist; pass a local path or an absolute workspace path", filepath.ToSlash(localPath)) + } else if err != nil { + return "", err + } + return workspacePath, nil +} + +// hasRelativeSelection reports whether a selection is a relative path. +func hasRelativeSelection(selections []string) bool { + for _, selection := range selections { + if p, _, _ := strings.Cut(selection, "::"); !strings.HasPrefix(p, "/") { + return true + } + } + return false +} + +// rootPathSelection resolves a relative path against the pipeline's root_path. +func rootPathSelection(rootPath, relPath string) (string, error) { + if rootPath == "" { + return "", errors.New("the pipeline has no root_path; use an absolute workspace path such as /Workspace/Users/someone@example.com/tests") + } + return path.Join(rootPath, filepath.ToSlash(relPath)), nil +} + +// getRootPath reads the root_path of the deployed pipeline. +func getRootPath(ctx context.Context, w *databricks.WorkspaceClient, pipelineID string) (string, error) { + pipeline, err := w.Pipelines.Get(ctx, pipelines.GetPipelineRequest{PipelineId: pipelineID}) + if err != nil { + return "", err + } + if pipeline.Spec == nil { + return "", nil + } + return pipeline.Spec.RootPath, 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) + if err != nil { + return "", fmt.Errorf("failed to create API client: %w", err) + } + + var response pipelines.StartUpdateResponse + err = apiClient.Do( + ctx, + http.MethodPost, + fmt.Sprintf("/api/2.0/pipelines/%s/updates", pipelineID), + auth.WorkspaceIDHeaders(w.Config), + nil, + req, + &response, + ) + if err != nil { + return "", err + } + return response.UpdateId, nil +} + +// testTarget is the pipeline a test run is submitted to. +type testTarget struct { + key string + pipelineID string + selections []string + w *databricks.WorkspaceClient +} + +func (t testTarget) printHeader(out io.Writer, updateID string) { + if t.key != "" { + fmt.Fprintf(out, "Pipeline: %s (%s)\n", t.key, t.pipelineID) + } else { + fmt.Fprintf(out, "Pipeline: %s\n", t.pipelineID) + } + fmt.Fprintf(out, "Update: %s\n", updateID) +} + +// resolveTestTargetByID identifies the pipeline by --pipeline-id; all positional arguments are selections. +func resolveTestTargetByID(cmd *cobra.Command, args []string, pipelineID string) (testTarget, error) { + if cmd.Flag("target").Value.String() == "" { + cmd.SetContext(root.SkipLoadBundle(cmd.Context())) + } + if err := root.MustWorkspaceClient(cmd, args); err != nil { + return testTarget{}, err + } + ctx := cmd.Context() + w := cmdctx.WorkspaceClient(ctx) + + var rootPath string + if hasRelativeSelection(args) { + var err error + rootPath, err = getRootPath(ctx, w, pipelineID) + if err != nil { + return testTarget{}, err + } + } + selections, err := resolveTestSelections(args, func(p string) (string, error) { + return rootPathSelection(rootPath, p) + }) + if err != nil { + return testTarget{}, err + } + return testTarget{pipelineID: pipelineID, selections: selections, w: w}, nil +} + +// resolveTestTargetInBundle identifies the pipeline by its bundle KEY, which must precede any selections. +func resolveTestTargetInBundle(cmd *cobra.Command, args []string) (testTarget, error) { + var key string + b, err := utils.ProcessBundle(cmd, utils.ProcessOptions{ + PostInitFunc: func(ctx context.Context, b *bundle.Bundle) error { + var err error + key, err = resolvePipelineArgument(ctx, b, args[:min(len(args), 1)]) + return err + }, + ErrorOnEmptyState: true, + SkipInitContext: true, + }) + if err != nil { + return testTarget{}, err + } + ctx := cmd.Context() + + if _, err := resources.Lookup(b, key, isPipeline); err != nil { + return testTarget{}, fmt.Errorf("%w; the first argument must be a pipeline KEY when test selections are given", err) + } + pipelineID, err := resolvePipelineIdFromKey(ctx, b, key) + if err != nil { + return testTarget{}, err + } + suggestPipelineDeploy(ctx, cmd) + + cwd, err := os.Getwd() + if err != nil { + return testTarget{}, err + } + var selectionArgs []string + if len(args) > 1 { + selectionArgs = args[1:] + } + filesRoot := bundleFilesRoot(b) + selections, err := resolveTestSelections(selectionArgs, func(p string) (string, error) { + return bundleTestSelectionPath(cwd, b.SyncRootPath, filesRoot, p) + }) + if err != nil { + return testTarget{}, err + } + return testTarget{key: key, pipelineID: pipelineID, selections: selections, w: b.WorkspaceClient(ctx)}, nil +} + +func testCommand() *cobra.Command { + cmd := &cobra.Command{ + Use: "test [PIPELINE_KEY] [TEST_SELECTION...] [flags] [--pytest-arg=ARG ...]", + Short: "*Preview* Run the tests of a pipeline", + Long: `Run the tests of a pipeline in a test-only update. + +This command is experimental. Set DATABRICKS_CLI_PIPELINES_TEST=true to enable it. + +PIPELINE_KEY is the unique name of the pipeline, as defined in its YAML file. +It is required when test selections are given. If there is only one pipeline +in the project and no arguments are given, it is selected automatically. +Outside a project, identify the pipeline with --pipeline-id; all positional +arguments are then test selections. + +TEST_SELECTION is a test directory, file, or pytest test ID, for example +tests/test_orders.py::test_backfill. In a project, a relative path is a local +path, relative to the working directory, and is converted to its deployed +workspace path. With --pipeline-id, a relative path is relative to the +pipeline's root_path. Paths that start with / are workspace paths. Without a +selection, all of the pipeline's tests run. + +Each --pytest-arg passes one argument to pytest, in the order given. Only -k +and -m are supported, each followed by its value in the next --pytest-arg, for +example: + + databricks pipelines test my_pipeline --pytest-arg=-k --pytest-arg='orders and not slow'`, + } + + var pipelineID string + var pytestArgs []string + + testGroup := cmdgroup.NewFlagGroup("Pipeline Test") + testGroup.FlagSet().StringVar(&pipelineID, "pipeline-id", "", "ID of the pipeline to test, instead of a PIPELINE_KEY.") + testGroup.FlagSet().StringArrayVar(&pytestArgs, "pytest-arg", nil, "Argument to pass to pytest (repeatable). Only -k and -m are supported.") + + wrappedCmd := cmdgroup.NewCommandWithGroupFlag(cmd) + wrappedCmd.AddFlagGroup(testGroup) + + cmd.SetFlagErrorFunc(testFlagErrorFunc) + + cmd.RunE = func(cmd *cobra.Command, args []string) error { + if enabled, _ := env.GetBool(cmd.Context(), pipelinesTestEnvVar); !enabled { + return fmt.Errorf("pipelines test is experimental. Set %s=true to enable it", pipelinesTestEnvVar) + } + if cmd.ArgsLenAtDash() >= 0 { + return errors.New("the -- separator is not supported: " + pytestArgHint) + } + if root.OutputType(cmd) != flags.OutputText { + return fmt.Errorf("--output %s is not supported by pipelines test", root.OutputType(cmd)) + } + if err := validatePytestArgs(pytestArgs); err != nil { + return err + } + + var target testTarget + var err error + if pipelineID != "" { + target, err = resolveTestTargetByID(cmd, args, pipelineID) + } else { + ctx := logdiag.InitContext(cmd.Context()) + cmd.SetContext(ctx) + if !bundle.HasRoot(ctx) { + return errors.New("outside a project, identify the pipeline with --pipeline-id") + } + target, err = resolveTestTargetInBundle(cmd, args) + } + 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 { + return err + } + target.printHeader(cmd.OutOrStdout(), updateID) + return nil + } + + return cmd +} diff --git a/cmd/pipelines/test_test.go b/cmd/pipelines/test_test.go new file mode 100644 index 00000000000..dc4de92ed0f --- /dev/null +++ b/cmd/pipelines/test_test.go @@ -0,0 +1,95 @@ +package pipelines + +import ( + "errors" + "os" + "path/filepath" + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestValidatePytestArgs(t *testing.T) { + tests := []struct { + args []string + wantErr string + }{ + {args: nil}, + {args: []string{"-k", "orders and not slow", "-m", "smoke,fast"}}, + {args: []string{"-korders"}, wantErr: `pytest argument "-korders" must be split: pass --pytest-arg=-k --pytest-arg=orders`}, + {args: []string{"-k=orders and not slow"}, wantErr: `pytest argument "-k=orders and not slow" must be split: pass --pytest-arg=-k --pytest-arg='orders and not slow'`}, + {args: []string{"-k", " "}, wantErr: "pytest argument -k requires a non-empty value"}, + {args: []string{"-k", "-m"}, wantErr: `pytest argument -k requires a value, not "-m": pass it in the next --pytest-arg`}, + {args: []string{"-k"}, wantErr: "pytest argument -k requires a value: pass it in the next --pytest-arg"}, + {args: []string{"-x"}, wantErr: `unsupported pytest argument "-x": only -k and -m are supported`}, + {args: []string{"--collect-only"}, wantErr: `unsupported pytest argument "--collect-only": only -k and -m are supported`}, + {args: []string{"tests/test_orders.py"}, wantErr: `unsupported pytest argument "tests/test_orders.py": pass test paths and test IDs as TEST_SELECTION arguments`}, + } + for _, tc := range tests { + err := validatePytestArgs(tc.args) + if tc.wantErr == "" { + assert.NoError(t, err, tc.args) + } else { + assert.EqualError(t, err, tc.wantErr, tc.args) + } + } +} + +func TestResolveTestSelections(t *testing.T) { + toWorkspacePath := func(p string) (string, error) { + if p == "missing" { + return "", errors.New("not found") + } + return "/Workspace/root/" + p, nil + } + + resolved, err := resolveTestSelections([]string{ + "tests", + "tests/test_orders.py::TestOrders::test_backfill[a::b]", + "/Workspace/Users/someone/../someone/tests/test_x.py::test_x", + }, toWorkspacePath) + require.NoError(t, err) + assert.Equal(t, []string{ + "/Workspace/root/tests", + "/Workspace/root/tests/test_orders.py::TestOrders::test_backfill[a::b]", + "/Workspace/Users/someone/tests/test_x.py::test_x", + }, resolved) + + _, err = resolveTestSelections([]string{"::test_x"}, toWorkspacePath) + assert.EqualError(t, err, `test selection "::test_x" has no path`) + + _, err = resolveTestSelections([]string{"missing::test_x"}, toWorkspacePath) + assert.EqualError(t, err, `cannot resolve test selection "missing::test_x": not found`) +} + +// Source-linked deployments run the source files, as TranslatePaths maps them. +func TestBundleFilesRootSourceLinked(t *testing.T) { + b := &bundle.Bundle{SyncRootPath: "/Workspace/Users/someone/orders"} + b.Config.Workspace.FilePath = "/Workspace/Users/someone/.bundle/orders/dev/files" + b.Config.Workspace.SyncMode = config.SyncModeInPlace + assert.Equal(t, "/Workspace/Users/someone/orders", bundleFilesRoot(b)) +} + +func TestBundleTestSelectionPath(t *testing.T) { + syncRoot := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(syncRoot, "pipeline", "tests"), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(syncRoot, "conftest.py"), nil, 0o644)) + filesRoot := "/Workspace/Users/someone/.bundle/orders/dev/files" + + // A path may leave the working directory as long as it stays in the synchronized files. + got, err := bundleTestSelectionPath(filepath.Join(syncRoot, "pipeline", "tests"), syncRoot, filesRoot, filepath.Join("..", "..", "conftest.py")) + require.NoError(t, err) + assert.Equal(t, filesRoot+"/conftest.py", got) + + // A path that does not exist locally is an error, so that a typo fails before compute starts. + _, err = bundleTestSelectionPath(syncRoot, syncRoot, filesRoot, "tests/test_a.py") + assert.EqualError(t, err, filepath.ToSlash(filepath.Join(syncRoot, "tests", "test_a.py"))+" does not exist; pass a local path or an absolute workspace path") + + // A path outside the synchronized files is an error. The error shows forward slashes. + _, err = bundleTestSelectionPath(syncRoot, syncRoot, filesRoot, filepath.Join("..", "tests")) + assert.ErrorContains(t, err, "is outside the synchronized bundle files in "+filepath.ToSlash(syncRoot)) +} diff --git a/libs/testserver/fake_workspace.go b/libs/testserver/fake_workspace.go index c48ebb04df2..c9ea751f8f0 100644 --- a/libs/testserver/fake_workspace.go +++ b/libs/testserver/fake_workspace.go @@ -222,7 +222,9 @@ type FakeWorkspace struct { JobRunOutputs map[int64]jobs.RunOutput JobRunIdempotency map[string]int64 Pipelines map[string]pipelines.GetPipelineResponse - PipelineUpdates map[string]bool + PipelineUpdates map[string]*fakePipelineUpdate + pipelineEvents []fakePipelineEvent + pipelineEventTime time.Time Monitors map[string]catalog.MonitorInfo DataQualityMonitors map[string]dataquality.Monitor Apps map[string]apps.App @@ -531,7 +533,8 @@ func NewFakeWorkspace(url, token string) *FakeWorkspace { JobRunIdempotency: map[string]int64{}, Grants: map[string][]catalog.PrivilegeAssignment{}, Pipelines: map[string]pipelines.GetPipelineResponse{}, - PipelineUpdates: map[string]bool{}, + PipelineUpdates: map[string]*fakePipelineUpdate{}, + pipelineEventTime: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), Monitors: map[string]catalog.MonitorInfo{}, DataQualityMonitors: map[string]dataquality.Monitor{}, Apps: map[string]apps.App{}, diff --git a/libs/testserver/handlers.go b/libs/testserver/handlers.go index 1fa448f4743..a28f0f55961 100644 --- a/libs/testserver/handlers.go +++ b/libs/testserver/handlers.go @@ -674,11 +674,11 @@ func AddDefaultHandlers(server *Server) { }) server.Handle("POST", "/api/2.0/pipelines/{pipeline_id}/updates", func(req Request) any { - return req.Workspace.PipelineStartUpdate(req.Vars["pipeline_id"]) + return req.Workspace.PipelineStartUpdate(req.Vars["pipeline_id"], req.Body) }) server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req Request) any { - return req.Workspace.PipelineEvents(req.Vars["pipeline_id"]) + return req.Workspace.PipelineEvents(req.Vars["pipeline_id"], req.URL.Query()) }) server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/updates/{update_id}", func(req Request) any { diff --git a/libs/testserver/pipelines.go b/libs/testserver/pipelines.go index 0aedc44c5d3..48e0744acfb 100644 --- a/libs/testserver/pipelines.go +++ b/libs/testserver/pipelines.go @@ -143,10 +143,10 @@ func (s *FakeWorkspace) PipelineUpdate(req Request, pipelineId string) Response return Response{} } -func (s *FakeWorkspace) PipelineStartUpdate(pipelineId string) Response { +func (s *FakeWorkspace) PipelineStartUpdate(pipelineId string, body []byte) Response { defer s.LockUnlock()() - _, exists := s.Pipelines[pipelineId] + pipeline, exists := s.Pipelines[pipelineId] if !exists { return Response{ StatusCode: 404, @@ -154,30 +154,63 @@ func (s *FakeWorkspace) PipelineStartUpdate(pipelineId string) Response { } } - updateId := nextUUID() - s.PipelineUpdates[updateId] = true - - return Response{ - Body: pipelines.StartUpdateResponse{ - UpdateId: updateId, - }, + for id, update := range s.PipelineUpdates { + if update.pipelineID == pipelineId && update.active() { + return Response{ + StatusCode: 409, + Body: map[string]string{ + "error_code": "INVALID_STATE_TRANSITION", + "message": fmt.Sprintf("An active update '%s' already exists for pipeline '%s'.", id, pipelineId), + }, + } + } } -} -func (s *FakeWorkspace) PipelineEvents(pipelineId string) Response { - defer s.LockUnlock()() - - _, exists := s.Pipelines[pipelineId] - if !exists { + var test pipelineTestRequest + if len(body) > 0 { + if err := json.Unmarshal(body, &test); err != nil { + return Response{ + StatusCode: 400, + Body: fmt.Sprintf("cannot unmarshal request body: %s", err), + } + } + } + if test.TestOptions != nil && !test.TestOnly { return Response{ - StatusCode: 404, - Body: map[string]string{"message": fmt.Sprintf("The specified pipeline %s was not found.", pipelineId)}, + StatusCode: 400, + Body: map[string]string{"error_code": "INVALID_PARAMETER_VALUE", "message": "test_options requires test_only to be true."}, + } + } + if test.TestOptions != nil { + if message := pipelineTestOptionsError(test.TestOptions.PytestArgs, test.TestOptions.TestSelection); message != "" { + return Response{ + StatusCode: 400, + Body: map[string]string{"error_code": "INVALID_PARAMETER_VALUE", "message": message}, + } } } + updateId := nextUUID() + update := &fakePipelineUpdate{ + pipelineID: pipelineId, + states: []pipelines.UpdateInfoState{pipelines.UpdateInfoStateCompleted}, + } + if test.TestOnly { + var selection []string + if test.TestOptions != nil { + selection = test.TestOptions.TestSelection + } + var rootPath string + if pipeline.Spec != nil { + rootPath = pipeline.Spec.RootPath + } + update = s.startPipelineTestRun(pipelineId, updateId, s.collectPipelineTests(rootPath, selection)) + } + s.PipelineUpdates[updateId] = update + return Response{ - Body: map[string]any{ - "events": []pipelines.PipelineEvent{}, + Body: pipelines.StartUpdateResponse{ + UpdateId: updateId, }, } } @@ -193,25 +226,38 @@ func (s *FakeWorkspace) PipelineGetUpdate(pipelineId, updateId string) Response } } - // Check if the update exists - _, updateExists := s.PipelineUpdates[updateId] - if !updateExists { + update, updateExists := s.PipelineUpdates[updateId] + if !updateExists || update.pipelineID != pipelineId { return Response{ StatusCode: 404, Body: map[string]string{"message": fmt.Sprintf("The specified update %s was not found.", updateId)}, } } + update.polls++ return Response{ - Body: pipelines.GetUpdateResponse{ - Update: &pipelines.UpdateInfo{ + Body: fakeGetUpdateResponse{ + Update: fakeUpdateInfo{ UpdateId: updateId, - State: pipelines.UpdateInfoStateCompleted, + State: update.states[update.stateIndex()], + TestOnly: update.testOnly, }, }, } } +// fakeGetUpdateResponse is a Get Update response with UpdateInfo.test_only, which the Go SDK does not have +// yet. Like the service where pipeline testing is enabled, it reports test_only for every update. +type fakeGetUpdateResponse struct { + Update fakeUpdateInfo `json:"update"` +} + +type fakeUpdateInfo struct { + UpdateId string `json:"update_id"` + State pipelines.UpdateInfoState `json:"state"` + TestOnly bool `json:"test_only"` +} + func (s *FakeWorkspace) PipelineStop(pipelineId string) Response { defer s.LockUnlock()() @@ -223,6 +269,13 @@ func (s *FakeWorkspace) PipelineStop(pipelineId string) Response { } } + // Stop is pipeline-wide: it cancels whichever update is active. + for _, update := range s.PipelineUpdates { + if update.pipelineID == pipelineId && update.active() { + update.cancel() + } + } + return Response{ Body: pipelines.GetPipelineResponse{ PipelineId: pipelineId, diff --git a/libs/testserver/pipelines_test.go b/libs/testserver/pipelines_test.go index 39114714082..d8a9391c6d5 100644 --- a/libs/testserver/pipelines_test.go +++ b/libs/testserver/pipelines_test.go @@ -51,7 +51,7 @@ func TestPipelineCreate_AllowsSingleSegmentTargetSchemaName(t *testing.T) { func TestPipelineStartUpdate_HandlesNonExistentPipeline(t *testing.T) { workspace := NewFakeWorkspace("http://test", "dbapi123") - response := workspace.PipelineStartUpdate("non-existent-pipeline") + response := workspace.PipelineStartUpdate("non-existent-pipeline", nil) assert.Equal(t, 404, response.StatusCode) assert.Contains(t, response.Body.(map[string]string)["message"], "The specified pipeline non-existent-pipeline was not found") } @@ -74,7 +74,7 @@ func TestPipelineStop_AfterUpdate(t *testing.T) { pipelineId := createTestPipeline(t, workspace) - startResponse := workspace.PipelineStartUpdate(pipelineId) + startResponse := workspace.PipelineStartUpdate(pipelineId, nil) assert.Equal(t, 0, startResponse.StatusCode) stopResponse := workspace.PipelineStop(pipelineId) diff --git a/libs/testserver/pipelines_testing.go b/libs/testserver/pipelines_testing.go new file mode 100644 index 00000000000..23b8e498863 --- /dev/null +++ b/libs/testserver/pipelines_testing.go @@ -0,0 +1,400 @@ +package testserver + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "net/url" + "path" + "regexp" + "slices" + "strconv" + "strings" + "time" + + "github.com/databricks/databricks-sdk-go/service/pipelines" +) + +// Pipeline event types, which also name the event's details field. +const ( + pipelineEventUpdateProgress = "update_progress" + pipelineEventTestCaseProgress = "test_case_progress" + pipelineEventTestRunSummary = "test_run_summary" +) + +// fakePipelineUpdate is a pipeline update whose state advances by one step on each Get Update call. +type fakePipelineUpdate struct { + pipelineID string + testOnly bool + states []pipelines.UpdateInfoState + polls int + + // canceled is set by cancel. CANCELED then holds the index of a state the update never reached. + canceled bool +} + +// stateIndex returns the index of the state returned by the last Get Update call, or -1 before the +// first call. +func (u *fakePipelineUpdate) stateIndex() int { + return min(u.polls, len(u.states)) - 1 +} + +// eventIndex returns the index of the last state whose events are visible. Events emitted for a +// later state are not visible yet, and never are for a state that a canceled update did not reach. +func (u *fakePipelineUpdate) eventIndex() int { + if u.canceled { + return len(u.states) - 2 + } + return u.stateIndex() +} + +// active reports whether the update has not ended yet: the state of the last Get Update call, or +// the first state before that call, is not terminal. Like the service, an update that will end on +// its own is active until it does. +func (u *fakePipelineUpdate) active() bool { + return !isTerminalUpdateState(u.states[max(u.stateIndex(), 0)]) +} + +// cancel ends the update at its current state, as Stop does. CANCELED is the update's state from +// now on, even before the next Get Update call. +func (u *fakePipelineUpdate) cancel() { + u.states = append(u.states[:u.stateIndex()+1], pipelines.UpdateInfoStateCanceled) + u.polls = len(u.states) + u.canceled = true +} + +func isTerminalUpdateState(state pipelines.UpdateInfoState) bool { + switch state { + case pipelines.UpdateInfoStateCompleted, pipelines.UpdateInfoStateFailed, pipelines.UpdateInfoStateCanceled: + return true + default: + return false + } +} + +type fakePipelineEventOrigin struct { + PipelineId string `json:"pipeline_id"` + UpdateId string `json:"update_id"` + TestCaseId string `json:"test_case_id,omitempty"` +} + +// fakePipelineEvent is a pipeline event as returned by List Pipeline Events. The Go SDK +// PipelineEvent has no details or origin.test_case_id, so the fake defines its own shape. +type fakePipelineEvent struct { + Id string `json:"id"` + EventType string `json:"event_type"` + Timestamp string `json:"timestamp"` + Level pipelines.EventLevel `json:"level"` + Message string `json:"message"` + Origin fakePipelineEventOrigin `json:"origin"` + Error *pipelines.ErrorDetail `json:"error,omitempty"` + Details map[string]any `json:"details"` + + // visibleAt is the index of the update state from which the event can be read. + visibleAt int +} + +// pipelineTestRequest holds the Start Update fields of a test run, which are not in the Go SDK yet. +type pipelineTestRequest struct { + TestOnly bool `json:"test_only"` + TestOptions *struct { + TestSelection []string `json:"test_selection"` + PytestArgs []string `json:"pytest_args"` + } `json:"test_options"` +} + +var pipelineTestFunctionRegexp = regexp.MustCompile(`(?m)^def (test_\w+)\(`) + +// pipelineTestOptionsError returns the service's message for test options that it rejects, or "". Like +// PipelineTestSelection in the service, pytest_args holds -k and -m, each followed by a value that is not +// empty and does not start with "-", and each test selection is an absolute workspace path. +func pipelineTestOptionsError(pytestArgs, selection []string) string { + for i := 0; i < len(pytestArgs); i += 2 { + option := pytestArgs[i] + switch { + case option != "-k" && option != "-m": + return fmt.Sprintf("pytest_args accepts only -k and -m, each followed by its value. Got '%s'.", option) + case i+1 == len(pytestArgs): + return fmt.Sprintf("The pytest_args option '%s' must be followed by its value.", option) + case strings.TrimSpace(pytestArgs[i+1]) == "": + return fmt.Sprintf("The value of the pytest_args option '%s' must not be empty.", option) + case strings.HasPrefix(pytestArgs[i+1], "-"): + return fmt.Sprintf("The value of the pytest_args option '%s' must not start with '-'. Got '%s'.", option, pytestArgs[i+1]) + } + } + for _, entry := range selection { + if p, _, _ := strings.Cut(strings.TrimSpace(entry), "::"); !strings.HasPrefix(p, "/") { + return fmt.Sprintf("The test selection '%s' must be an absolute workspace path, such as /Workspace/Users/someone@example.com/orders/tests/test_orders.py.", entry) + } + } + return "" +} + +// collectPipelineTests stands in for pytest collection: it finds the top-level "def test_*" +// functions in the workspace test_*.py files under each selection, or under rootPath without one. +// Like origin.test_case_id, a test ID is the absolute workspace path of its file followed by the +// pytest test ID, such as /Workspace/Users/someone/orders/tests/test_orders.py::test_backfill. +func (s *FakeWorkspace) collectPipelineTests(rootPath string, selection []string) []string { + if len(selection) == 0 && rootPath != "" { + selection = []string{rootPath} + } + + var ids []string + for _, target := range selection { + targetPath, testName, _ := strings.Cut(target, "::") + var files []string + if _, ok := s.files[targetPath]; ok { + files = []string{targetPath} + } else { + for filePath := range s.files { + base := path.Base(filePath) + if strings.HasPrefix(filePath, targetPath+"/") && strings.HasPrefix(base, "test_") && strings.HasSuffix(base, ".py") { + files = append(files, filePath) + } + } + slices.Sort(files) + } + + for _, filePath := range files { + for _, match := range pipelineTestFunctionRegexp.FindAllSubmatch(s.files[filePath].Data, -1) { + name := string(match[1]) + id := filePath + "::" + name + if (testName == "" || testName == name) && !slices.Contains(ids, id) { + ids = append(ids, id) + } + } + } + } + return ids +} + +// startPipelineTestRun records the events of a test-only update. Tests do not run here; each +// test name selects its outcome: +// - "fail": FAILED, "error": ERROR (a fixture failed), "skip": SKIPPED, otherwise PASSED. +// - "crash": compute terminates when the case starts and the update fails without a summary. +// - "hang": the case never finishes and the update stays RUNNING until the pipeline is stopped. +// - "breaks_update": the case passes, but the update fails after the summary. +func (s *FakeWorkspace) startPipelineTestRun(pipelineID, updateID string, caseIDs []string) *fakePipelineUpdate { + const ( + running = 1 + ended = 2 + ) + update := &fakePipelineUpdate{ + pipelineID: pipelineID, + testOnly: true, + states: []pipelines.UpdateInfoState{pipelines.UpdateInfoStateWaitingForResources, pipelines.UpdateInfoStateRunning}, + } + + emit := func(visibleAt int, event fakePipelineEvent) { + s.pipelineEventTime = s.pipelineEventTime.Add(time.Second) + event.Id = nextUUID() + event.Timestamp = s.pipelineEventTime.UTC().Format("2006-01-02T15:04:05.000Z") + event.Origin.PipelineId = pipelineID + event.Origin.UpdateId = updateID + if event.Level == "" { + event.Level = pipelines.EventLevelInfo + } + event.visibleAt = visibleAt + s.pipelineEvents = append(s.pipelineEvents, event) + } + emitCase := func(id, status string, errorDetail *pipelines.ErrorDetail) { + event := fakePipelineEvent{ + EventType: pipelineEventTestCaseProgress, + Message: fmt.Sprintf("Test case %s is %s.", id, status), + Origin: fakePipelineEventOrigin{TestCaseId: id}, + Error: errorDetail, + Details: map[string]any{pipelineEventTestCaseProgress: map[string]any{"status": status}}, + } + if errorDetail != nil { + event.Level = pipelines.EventLevelError + } + emit(running, event) + } + exception := func(className, message string) *pipelines.ErrorDetail { + return &pipelines.ErrorDetail{Exceptions: []pipelines.SerializedException{{ClassName: className, Message: message}}} + } + + for _, id := range caseIDs { + emitCase(id, "PENDING", nil) + } + + counts := map[string]int{} + breaksUpdate := false + for _, id := range caseIDs { + emitCase(id, "RUNNING", nil) + _, name, _ := strings.Cut(id, "::") + switch { + case strings.Contains(name, "hang"): + return update + case strings.Contains(name, "crash"): + emit(ended, fakePipelineEvent{ + EventType: pipelineEventUpdateProgress, + Level: pipelines.EventLevelError, + Message: fmt.Sprintf("Update %s is FAILED.", updateID[:6]), + Error: &pipelines.ErrorDetail{Fatal: true, Exceptions: []pipelines.SerializedException{{ClassName: "ComputeTerminatedException", Message: "Compute terminated unexpectedly."}}}, + Details: map[string]any{pipelineEventUpdateProgress: map[string]any{"state": "FAILED"}}, + }) + update.states = append(update.states, pipelines.UpdateInfoStateFailed) + return update + case strings.Contains(name, "breaks_update"): + emitCase(id, "PASSED", nil) + counts["passed"]++ + breaksUpdate = true + case strings.Contains(name, "fail"): + emitCase(id, "FAILED", exception("AssertionError", "assert 100 == 150")) + counts["failed"]++ + case strings.Contains(name, "error"): + emitCase(id, "ERROR", exception("RuntimeError", "fixture 'orders' setup failed")) + counts["errors"]++ + case strings.Contains(name, "skip"): + emitCase(id, "SKIPPED", nil) + counts["skipped"]++ + default: + emitCase(id, "PASSED", nil) + counts["passed"]++ + } + } + + emit(ended, fakePipelineEvent{ + EventType: pipelineEventTestRunSummary, + Message: fmt.Sprintf("Collected %d tests.", len(caseIDs)), + Details: map[string]any{pipelineEventTestRunSummary: map[string]any{ + "total": len(caseIDs), + "passed": counts["passed"], + "failed": counts["failed"], + "skipped": counts["skipped"], + "errored": counts["errors"], + }}, + }) + // Like the service, a run whose tests failed or errored ends with the test verdict as the + // update's error. + if !breaksUpdate && (counts["failed"] > 0 || counts["errors"] > 0) { + failed := counts["failed"] + counts["errors"] + emit(ended, fakePipelineEvent{ + EventType: pipelineEventUpdateProgress, + Level: pipelines.EventLevelError, + Message: fmt.Sprintf("%d pipeline tests failed.", failed), + Error: &pipelines.ErrorDetail{Fatal: true, Exceptions: []pipelines.SerializedException{{ + ClassName: "TestFailure", + Message: fmt.Sprintf("[PIPELINE_TEST_FAILED] %d of the pipeline's tests failed (%d passed, %d skipped).", failed, counts["passed"], counts["skipped"]), + }}}, + Details: map[string]any{pipelineEventUpdateProgress: map[string]any{"state": "FAILED"}}, + }) + } + if breaksUpdate { + emit(ended, fakePipelineEvent{ + EventType: pipelineEventUpdateProgress, + Level: pipelines.EventLevelError, + Message: fmt.Sprintf("Update %s is FAILED.", updateID[:6]), + Error: &pipelines.ErrorDetail{Fatal: true, Exceptions: []pipelines.SerializedException{{ClassName: "ResultDeliveryException", Message: "Test results could not be verified."}}}, + Details: map[string]any{pipelineEventUpdateProgress: map[string]any{"state": "FAILED"}}, + }) + } + if len(caseIDs) == 0 || counts["failed"] > 0 || counts["errors"] > 0 || breaksUpdate { + update.states = append(update.states, pipelines.UpdateInfoStateFailed) + } else { + update.states = append(update.states, pipelines.UpdateInfoStateCompleted) + } + return update +} + +var ( + pipelineEventsUpdateIDFilter = regexp.MustCompile(`update_id\s*=\s*'([^']*)'`) + pipelineEventsTimestampFilter = regexp.MustCompile(`timestamp\s*>=\s*'([^']*)'`) +) + +// pipelineEventsPageToken is the query of the first page plus the offset of the next one. +type pipelineEventsPageToken struct { + Query url.Values `json:"query"` + Offset int `json:"offset"` +} + +// PipelineEvents lists the visible events of a pipeline. Only the update_id and timestamp >= +// conditions of the filter are applied. +func (s *FakeWorkspace) PipelineEvents(pipelineId string, query url.Values) 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)}, + } + } + + invalid := func(message string) Response { + return Response{ + StatusCode: 400, + Body: map[string]string{"error_code": "INVALID_PARAMETER_VALUE", "message": message}, + } + } + + maxResults := 25 + if value := query.Get("max_results"); value != "" { + n, err := strconv.Atoi(value) + if err != nil || n <= 0 { + return invalid("max_results must be a positive integer.") + } + maxResults = min(n, 250) + } + + offset := 0 + if token := query.Get("page_token"); token != "" { + for key := range query { + if key != "page_token" && key != "max_results" { + return invalid(fmt.Sprintf("page_token cannot be used with %s.", key)) + } + } + data, err := base64.URLEncoding.DecodeString(token) + var decoded pipelineEventsPageToken + if err == nil { + err = json.Unmarshal(data, &decoded) + } + if err != nil { + return invalid("Invalid page_token.") + } + query = decoded.Query + offset = decoded.Offset + } + + updateID := "" + if match := pipelineEventsUpdateIDFilter.FindStringSubmatch(query.Get("filter")); match != nil { + updateID = match[1] + } + var since time.Time + if match := pipelineEventsTimestampFilter.FindStringSubmatch(query.Get("filter")); match != nil { + var err error + since, err = time.Parse(time.RFC3339Nano, match[1]) + if err != nil { + return invalid(fmt.Sprintf("Invalid timestamp in filter: %s.", match[1])) + } + } + + var events []fakePipelineEvent + for _, event := range s.pipelineEvents { + update := s.PipelineUpdates[event.Origin.UpdateId] + if update.pipelineID != pipelineId || event.visibleAt > update.eventIndex() { + continue + } + if updateID != "" && event.Origin.UpdateId != updateID { + continue + } + if timestamp, _ := time.Parse(time.RFC3339Nano, event.Timestamp); timestamp.Before(since) { + continue + } + events = append(events, event) + } + + // Events are returned newest first unless ascending order is requested. + if query.Get("order_by") != "timestamp asc" { + slices.Reverse(events) + } + + page := events[min(offset, len(events)):] + body := map[string]any{"events": append([]fakePipelineEvent{}, page[:min(maxResults, len(page))]...)} + if len(page) > maxResults { + data, _ := json.Marshal(pipelineEventsPageToken{Query: query, Offset: offset + maxResults}) + body["next_page_token"] = base64.URLEncoding.EncodeToString(data) + } + return Response{Body: body} +} diff --git a/libs/testserver/pipelines_testing_test.go b/libs/testserver/pipelines_testing_test.go new file mode 100644 index 00000000000..c58bc074ba2 --- /dev/null +++ b/libs/testserver/pipelines_testing_test.go @@ -0,0 +1,243 @@ +package testserver + +import ( + "net/url" + "testing" + + "github.com/databricks/databricks-sdk-go/service/pipelines" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testRootPath = "/Workspace/Users/someone/orders" + +func createTestPipelineWithTests(t *testing.T, workspace *FakeWorkspace) string { + response := workspace.PipelineCreate(Request{Body: []byte(`{"name": "orders", "root_path": "` + testRootPath + `"}`)}) + require.Equal(t, 0, response.StatusCode) + workspace.files[testRootPath+"/tests/test_orders.py"] = FileEntry{Data: []byte("def test_ok():\n pass\n\ndef test_fails():\n pass\n\n def test_nested():\n pass\n")} + workspace.files[testRootPath+"/tests/test_slow.py"] = FileEntry{Data: []byte("def test_hang():\n pass\n")} + workspace.files[testRootPath+"/tests/helpers.py"] = FileEntry{Data: []byte("def test_helper():\n pass\n")} + return response.Body.(pipelines.CreatePipelineResponse).PipelineId +} + +func startTestUpdate(t *testing.T, workspace *FakeWorkspace, pipelineID, body string) string { + response := workspace.PipelineStartUpdate(pipelineID, []byte(body)) + require.Equal(t, 0, response.StatusCode, response.Body) + return response.Body.(pipelines.StartUpdateResponse).UpdateId +} + +func getUpdateState(workspace *FakeWorkspace, pipelineID, updateID string) pipelines.UpdateInfoState { + return workspace.PipelineGetUpdate(pipelineID, updateID).Body.(fakeGetUpdateResponse).Update.State +} + +func TestPipelineGetUpdateReportsTestOnly(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + + testUpdateID := startTestUpdate(t, workspace, pipelineID, `{"test_only": true}`) + assert.True(t, workspace.PipelineGetUpdate(pipelineID, testUpdateID).Body.(fakeGetUpdateResponse).Update.TestOnly) + + workspace.PipelineStop(pipelineID) + updateID := startTestUpdate(t, workspace, pipelineID, `{}`) + assert.False(t, workspace.PipelineGetUpdate(pipelineID, updateID).Body.(fakeGetUpdateResponse).Update.TestOnly) +} + +func listEvents(t *testing.T, workspace *FakeWorkspace, pipelineID string, query url.Values) ([]fakePipelineEvent, string) { + response := workspace.PipelineEvents(pipelineID, query) + require.Equal(t, 0, response.StatusCode, response.Body) + body := response.Body.(map[string]any) + token, _ := body["next_page_token"].(string) + return body["events"].([]fakePipelineEvent), token +} + +func TestCollectPipelineTests(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + createTestPipelineWithTests(t, workspace) + + // Like origin.test_case_id, an ID is the absolute workspace path of its file and the pytest test ID. + tests := testRootPath + "/tests/" + assert.Equal(t, []string{tests + "test_orders.py::test_ok", tests + "test_orders.py::test_fails", tests + "test_slow.py::test_hang"}, + workspace.collectPipelineTests(testRootPath, nil)) + assert.Equal(t, []string{tests + "test_orders.py::test_fails"}, + workspace.collectPipelineTests(testRootPath, []string{testRootPath + "/tests/test_orders.py::test_fails", testRootPath + "/tests/test_orders.py::test_fails"})) + assert.Equal(t, []string{tests + "helpers.py::test_helper"}, + workspace.collectPipelineTests(testRootPath, []string{testRootPath + "/tests/helpers.py"})) + assert.Equal(t, []string{testRootPath + "/tests/test_slow.py::test_hang"}, + workspace.collectPipelineTests("", []string{testRootPath + "/tests/test_slow.py"})) + assert.Empty(t, workspace.collectPipelineTests("", nil)) +} + +func TestPipelineTestUpdate(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + updateID := startTestUpdate(t, workspace, pipelineID, `{"test_only": true, "test_options": {"test_selection": ["`+testRootPath+`/tests/test_orders.py"]}}`) + query := url.Values{"filter": {"update_id='" + updateID + "'"}, "order_by": {"timestamp asc"}} + + events, _ := listEvents(t, workspace, pipelineID, query) + assert.Empty(t, events) + + assert.Equal(t, pipelines.UpdateInfoStateWaitingForResources, getUpdateState(workspace, pipelineID, updateID)) + assert.Equal(t, pipelines.UpdateInfoStateRunning, getUpdateState(workspace, pipelineID, updateID)) + events, _ = listEvents(t, workspace, pipelineID, query) + var statuses []string + for _, e := range events { + statuses = append(statuses, e.Origin.TestCaseId+" "+e.Details["test_case_progress"].(map[string]any)["status"].(string)) + } + file := testRootPath + "/tests/test_orders.py" + assert.Equal(t, []string{ + file + "::test_ok PENDING", + file + "::test_fails PENDING", + file + "::test_ok RUNNING", + file + "::test_ok PASSED", + file + "::test_fails RUNNING", + file + "::test_fails FAILED", + }, statuses) + + assert.Equal(t, pipelines.UpdateInfoStateFailed, getUpdateState(workspace, pipelineID, updateID)) + last := events[len(events)-1] + query.Set("filter", "update_id='"+updateID+"' AND timestamp >= '"+last.Timestamp+"'") + events, _ = listEvents(t, workspace, pipelineID, query) + require.Len(t, events, 3) + assert.Equal(t, last.Id, events[0].Id) + assert.Equal(t, map[string]any{"total": 2, "passed": 1, "failed": 1, "skipped": 0, "errored": 0}, events[1].Details["test_run_summary"]) + // Like the service, the update's error is the test verdict. + require.NotNil(t, events[2].Error) + assert.True(t, events[2].Error.Fatal) + assert.Equal(t, "[PIPELINE_TEST_FAILED] 1 of the pipeline's tests failed (1 passed, 0 skipped).", events[2].Error.Exceptions[0].Message) + + query.Set("filter", "timestamp >= 'yesterday'") + assert.Equal(t, 400, workspace.PipelineEvents(pipelineID, query).StatusCode) +} + +func TestPipelineEventsPages(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + updateID := startTestUpdate(t, workspace, pipelineID, `{"test_only": true}`) + for range 3 { + getUpdateState(workspace, pipelineID, updateID) + } + + all, token := listEvents(t, workspace, pipelineID, url.Values{"order_by": {"timestamp asc"}, "max_results": {"100"}}) + assert.Empty(t, token) + require.Len(t, all, 8) + + var paged []fakePipelineEvent + query := url.Values{"filter": {"update_id='" + updateID + "'"}, "order_by": {"timestamp asc"}, "max_results": {"3"}} + for { + events, token := listEvents(t, workspace, pipelineID, query) + paged = append(paged, events...) + if token == "" { + break + } + query = url.Values{"page_token": {token}, "max_results": {"3"}} + } + assert.Equal(t, all, paged) + + // Events are newest first by default. + newest, _ := listEvents(t, workspace, pipelineID, url.Values{"max_results": {"1"}}) + assert.Equal(t, all[len(all)-1].Id, newest[0].Id) + + response := workspace.PipelineEvents(pipelineID, url.Values{"page_token": {"x"}, "filter": {"level='INFO'"}}) + assert.Equal(t, 400, response.StatusCode) +} + +func TestPipelineTestUpdateConflictAndStop(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + updateID := startTestUpdate(t, workspace, pipelineID, `{"test_only": true, "test_options": {"test_selection": ["`+testRootPath+`/tests/test_slow.py"]}}`) + + response := workspace.PipelineStartUpdate(pipelineID, nil) + assert.Equal(t, 409, response.StatusCode) + assert.Equal(t, "An active update '"+updateID+"' already exists for pipeline '"+pipelineID+"'.", response.Body.(map[string]string)["message"]) + + workspace.PipelineStop(pipelineID) + for range 2 { + getUpdateState(workspace, pipelineID, updateID) + } + assert.Equal(t, pipelines.UpdateInfoStateCanceled, getUpdateState(workspace, pipelineID, updateID)) + startTestUpdate(t, workspace, pipelineID, `{}`) +} + +func TestPipelineStartUpdateRejectsTestOptionsWithoutTestOnly(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + + response := workspace.PipelineStartUpdate(pipelineID, []byte(`{"test_options": {"pytest_args": ["-k", "x"]}}`)) + assert.Equal(t, 400, response.StatusCode) +} + +func TestPipelineStartUpdateRejectsInvalidTestOptions(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + + // The messages are the service's, from PipelineTestSelection. + for options, message := range map[string]string{ + `{"pytest_args": ["-korders"]}`: "pytest_args accepts only -k and -m, each followed by its value. Got '-korders'.", + `{"pytest_args": ["-x"]}`: "pytest_args accepts only -k and -m, each followed by its value. Got '-x'.", + `{"pytest_args": ["-m"]}`: "The pytest_args option '-m' must be followed by its value.", + `{"pytest_args": ["-k", " "]}`: "The value of the pytest_args option '-k' must not be empty.", + `{"pytest_args": ["-k", "-m"]}`: "The value of the pytest_args option '-k' must not start with '-'. Got '-m'.", + `{"test_selection": ["tests/test_a.py::t"]}`: "The test selection 'tests/test_a.py::t' must be an absolute workspace path, such as /Workspace/Users/someone@example.com/orders/tests/test_orders.py.", + } { + response := workspace.PipelineStartUpdate(pipelineID, []byte(`{"test_only": true, "test_options": `+options+`}`)) + assert.Equal(t, 400, response.StatusCode, options) + assert.Equal(t, map[string]string{"error_code": "INVALID_PARAMETER_VALUE", "message": message}, response.Body, options) + } + + startTestUpdate(t, workspace, pipelineID, `{"test_only": true, "test_options": {"pytest_args": ["-k", "orders", "-m", "smoke"], "test_selection": ["`+testRootPath+`/tests"]}}`) +} + +func TestPipelineFiniteUpdateIsActiveUntilItEnds(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + selection := `{"test_only": true, "test_options": {"test_selection": ["` + testRootPath + `/tests/test_orders.py::test_ok"]}}` + updateID := startTestUpdate(t, workspace, pipelineID, selection) + + // Like the service, a second update is rejected until the first one ends, even if it will end on its own. + assert.Equal(t, 409, workspace.PipelineStartUpdate(pipelineID, []byte(selection)).StatusCode) + assert.Equal(t, pipelines.UpdateInfoStateWaitingForResources, getUpdateState(workspace, pipelineID, updateID)) + assert.Equal(t, pipelines.UpdateInfoStateRunning, getUpdateState(workspace, pipelineID, updateID)) + require.Equal(t, pipelines.UpdateInfoStateCompleted, getUpdateState(workspace, pipelineID, updateID)) + startTestUpdate(t, workspace, pipelineID, selection) +} + +func TestPipelineUpdateThatIsNeverPolledEndsWithItsFirstState(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + + // A plain update completes at once, so a second one can start without polling the first. + updateID := startTestUpdate(t, workspace, pipelineID, `{}`) + startTestUpdate(t, workspace, pipelineID, `{}`) + + // Stop leaves an update that already completed as it is. + require.Equal(t, 0, workspace.PipelineStop(pipelineID).StatusCode) + assert.Equal(t, pipelines.UpdateInfoStateCompleted, getUpdateState(workspace, pipelineID, updateID)) +} + +func TestPipelineStopHidesEventsOfStatesNotReached(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + updateID := startTestUpdate(t, workspace, pipelineID, `{"test_only": true, "test_options": {"test_selection": ["`+testRootPath+`/tests/test_orders.py"]}}`) + query := url.Values{"filter": {"update_id='" + updateID + "'"}} + + assert.Equal(t, pipelines.UpdateInfoStateWaitingForResources, getUpdateState(workspace, pipelineID, updateID)) + require.Equal(t, 0, workspace.PipelineStop(pipelineID).StatusCode) + assert.Equal(t, pipelines.UpdateInfoStateCanceled, getUpdateState(workspace, pipelineID, updateID)) + + // The update was stopped before RUNNING, so its case results and summary never show. + events, _ := listEvents(t, workspace, pipelineID, query) + assert.Empty(t, events) +} + +func TestPipelineStopCancelsAFiniteUpdate(t *testing.T) { + workspace := NewFakeWorkspace("http://test", "dbapi123") + pipelineID := createTestPipelineWithTests(t, workspace) + selection := `{"test_only": true, "test_options": {"test_selection": ["` + testRootPath + `/tests/test_orders.py::test_ok"]}}` + updateID := startTestUpdate(t, workspace, pipelineID, selection) + + require.Equal(t, 0, workspace.PipelineStop(pipelineID).StatusCode) + // The stopped update ends at once, so a new one can start before anyone reads the old one. + startTestUpdate(t, workspace, pipelineID, selection) + assert.Equal(t, pipelines.UpdateInfoStateCanceled, getUpdateState(workspace, pipelineID, updateID)) + assert.Equal(t, pipelines.UpdateInfoStateCanceled, getUpdateState(workspace, pipelineID, updateID)) +}