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.flag-error.xml b/acceptance/pipelines/test/arguments/out.flag-error.xml new file mode 100644 index 00000000000..475a397f5fe --- /dev/null +++ b/acceptance/pipelines/test/arguments/out.flag-error.xml @@ -0,0 +1,8 @@ + + + + + unsupported pytest argument "-x": only -k and -m are supported + + + diff --git a/acceptance/pipelines/test/arguments/out.gate-off.xml b/acceptance/pipelines/test/arguments/out.gate-off.xml new file mode 100644 index 00000000000..efe909a6fa0 --- /dev/null +++ b/acceptance/pipelines/test/arguments/out.gate-off.xml @@ -0,0 +1,8 @@ + + + + + pipelines test is experimental. Set DATABRICKS_CLI_PIPELINES_TEST=true to enable it + + + diff --git a/acceptance/pipelines/test/arguments/out.no-wait.xml b/acceptance/pipelines/test/arguments/out.no-wait.xml new file mode 100644 index 00000000000..5f025801a52 --- /dev/null +++ b/acceptance/pipelines/test/arguments/out.no-wait.xml @@ -0,0 +1,8 @@ + + + + + --junit-xml cannot be used with --no-wait because no results are available when the command returns + + + 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..d03f8618d22 --- /dev/null +++ b/acceptance/pipelines/test/arguments/output.txt @@ -0,0 +1,116 @@ + +>>> [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. + +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' + +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. + +--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'. + +--junit-xml writes one entry per test with its outcome and duration. An +incomplete run, a run without tests, or a timeout adds a failing entry that +names the cause. The report replaces the file on every exit except a flag that +cannot be parsed. With workspace.sync_mode immutable, test names keep the path +of the deployed snapshot, which changes with each deployment. + +Usage: + databricks pipelines test [flags] [KEY] [TEST_SELECTION...] + +Pipeline Test Flags: + --junit-xml string Write the results as a JUnit XML report to this file, replacing it. + --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 + +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 + +=== With the command off, an older report is still replaced +>>> musterr [CLI] pipelines test --pipeline-id 1234 --junit-xml out.gate-off.xml +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 + +>>> musterr [CLI] pipelines test --pipeline-id 1234 --pytest-arg= +Error: --pytest-arg requires a non-empty value + +=== The pipeline ID must be a pipeline ID +>>> musterr [CLI] pipelines test --pipeline-id 1234 +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 + +=== --junit-xml needs the results +>>> musterr [CLI] pipelines test --pipeline-id 1234 --no-wait --junit-xml out.no-wait.xml +Error: --junit-xml cannot be used with --no-wait because no results are available when the command returns + +=== An invalid flag replaces an older report +>>> musterr [CLI] pipelines test --pipeline-id 1234 --pytest-arg=-x --junit-xml out.flag-error.xml +Error: unsupported pytest argument "-x": only -k and -m are supported diff --git a/acceptance/pipelines/test/arguments/script b/acceptance/pipelines/test/arguments/script new file mode 100644 index 00000000000..ceb54ec432e --- /dev/null +++ b/acceptance/pipelines/test/arguments/script @@ -0,0 +1,37 @@ +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 "With the command off, an older report is still replaced" +echo '' > out.gate-off.xml +(unset DATABRICKS_CLI_PIPELINES_TEST; trace musterr $CLI pipelines test --pipeline-id 1234 --junit-xml out.gate-off.xml) + +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 +trace musterr $CLI pipelines test --pipeline-id 1234 --pytest-arg="" + +title "The pipeline ID must be a pipeline ID" +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 "" + +title "--junit-xml needs the results" +trace musterr $CLI pipelines test --pipeline-id 1234 --no-wait --junit-xml out.no-wait.xml + +title "An invalid flag replaces an older report" +echo '' > out.flag-error.xml +trace musterr $CLI pipelines test --pipeline-id 1234 --pytest-arg=-x --junit-xml out.flag-error.xml diff --git a/acceptance/pipelines/test/immutable/databricks.yml b/acceptance/pipelines/test/immutable/databricks.yml new file mode 100644 index 00000000000..7f9a598cea2 --- /dev/null +++ b/acceptance/pipelines/test/immutable/databricks.yml @@ -0,0 +1,17 @@ +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 + testing: + sources: + - glob: + include: ./tests/** 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..04fcaaaeda7 --- /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][0]?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 --no-wait +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][0]) +Update: [UUID][1] + +>>> [CLI] pipelines test my_pipeline --no-wait +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][0]) +Update: [UUID][2] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID][0]/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][0]/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..e78d8a60a9c --- /dev/null +++ b/acceptance/pipelines/test/immutable/script @@ -0,0 +1,10 @@ +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 --no-wait +trace $CLI pipelines test my_pipeline --no-wait +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/invalid-project/databricks.yml b/acceptance/pipelines/test/invalid-project/databricks.yml new file mode 100644 index 00000000000..b23fa23f1a1 --- /dev/null +++ b/acceptance/pipelines/test/invalid-project/databricks.yml @@ -0,0 +1,7 @@ +bundle: + name: test-pipeline-test-invalid-project + +resources: + pipelines: + my_pipeline: + name: ${var.missing} diff --git a/acceptance/pipelines/test/invalid-project/out.invalid.xml b/acceptance/pipelines/test/invalid-project/out.invalid.xml new file mode 100644 index 00000000000..ff48bd9dac9 --- /dev/null +++ b/acceptance/pipelines/test/invalid-project/out.invalid.xml @@ -0,0 +1,8 @@ + + + + + reference does not exist: ${var.missing} + + + diff --git a/acceptance/pipelines/test/invalid-project/out.test.toml b/acceptance/pipelines/test/invalid-project/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/invalid-project/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/invalid-project/output.txt b/acceptance/pipelines/test/invalid-project/output.txt new file mode 100644 index 00000000000..5b8058bab75 --- /dev/null +++ b/acceptance/pipelines/test/invalid-project/output.txt @@ -0,0 +1,5 @@ + +=== A project with errors names its first error in the report +>>> musterr [CLI] pipelines test my_pipeline --junit-xml out.invalid.xml +Error: reference does not exist: ${var.missing} + diff --git a/acceptance/pipelines/test/invalid-project/script b/acceptance/pipelines/test/invalid-project/script new file mode 100644 index 00000000000..95bce3496ca --- /dev/null +++ b/acceptance/pipelines/test/invalid-project/script @@ -0,0 +1,2 @@ +title "A project with errors names its first error in the report" +trace musterr $CLI pipelines test my_pipeline --junit-xml out.invalid.xml diff --git a/acceptance/pipelines/test/live/databricks.yml.tmpl b/acceptance/pipelines/test/live/databricks.yml.tmpl new file mode 100644 index 00000000000..d9bc81ca0ca --- /dev/null +++ b/acceptance/pipelines/test/live/databricks.yml.tmpl @@ -0,0 +1,22 @@ +bundle: + name: test-pipelines-test-live-$UNIQUE_NAME + +resources: + pipelines: + orders: + # The Pipelines API rejects duplicate names, so the name is unique per run. + name: test-pipelines-test-live-$UNIQUE_NAME + catalog: main + schema: default + serverless: true + root_path: . + libraries: + - file: + path: ./transformations/orders.py + testing: + sources: + - glob: + include: ./tests/** + environment: + dependencies: + - -e ./helpers diff --git a/acceptance/pipelines/test/live/helpers/order_helpers.py b/acceptance/pipelines/test/live/helpers/order_helpers.py new file mode 100644 index 00000000000..8c963d8c00f --- /dev/null +++ b/acceptance/pipelines/test/live/helpers/order_helpers.py @@ -0,0 +1,2 @@ +def order_total(amounts): + return sum(amounts) diff --git a/acceptance/pipelines/test/live/helpers/pyproject.toml b/acceptance/pipelines/test/live/helpers/pyproject.toml new file mode 100644 index 00000000000..8062e6015d7 --- /dev/null +++ b/acceptance/pipelines/test/live/helpers/pyproject.toml @@ -0,0 +1,10 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "order-helpers" +version = "0.1.0" + +[tool.setuptools] +py-modules = ["order_helpers"] diff --git a/acceptance/pipelines/test/live/out.test.toml b/acceptance/pipelines/test/live/out.test.toml new file mode 100644 index 00000000000..314df1c2386 --- /dev/null +++ b/acceptance/pipelines/test/live/out.test.toml @@ -0,0 +1,2 @@ +Cloud = true +CloudSlow = true diff --git a/acceptance/pipelines/test/live/output.txt b/acceptance/pipelines/test/live/output.txt new file mode 100644 index 00000000000..8366f167fd2 --- /dev/null +++ b/acceptance/pipelines/test/live/output.txt @@ -0,0 +1,22 @@ + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-pipelines-test-live-[UNIQUE_NAME]/default/files... +Created pipelines.orders +Files: 6 uploaded, 0 deleted +Resources: 1 created, 0 changed, 0 deleted, 0 unchanged + +=== Run the tests and wait for the results +Done. PASSED=1 FAILED=0 SKIPPED=0 ERROR=0 TOTAL=1 + name="test_order_total" + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.pipelines.orders + +This action will result in the deletion of the following Lakeflow Spark Declarative Pipelines along with the +Streaming Tables (STs) and Materialized Views (MVs) managed by them. Set 'cascade_on_destroy: false' on a pipeline to retain datasets on pipeline deletion: + delete resources.pipelines.orders + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/test-pipelines-test-live-[UNIQUE_NAME]/default + +Destroy: 1 deleted diff --git a/acceptance/pipelines/test/live/script b/acceptance/pipelines/test/live/script new file mode 100644 index 00000000000..e9ceb2eff33 --- /dev/null +++ b/acceptance/pipelines/test/live/script @@ -0,0 +1,20 @@ +if [ -n "$CLOUD_ENV" ] && [ -z "${TEST_PIPELINES_TESTING_ENABLED:-}" ]; then + echo "SKIP_TEST set TEST_PIPELINES_TESTING_ENABLED for a workspace with pipeline testing enabled" + exit 0 +fi + +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve +} +trap cleanup EXIT + +trace $CLI bundle deploy + +title "Run the tests and wait for the results\n" +# Durations and test IDs differ between runs, so only the summary line and the case names are +# compared. The script stops if the command fails. +$CLI pipelines test --junit-xml junit.xml &> LOG.test +tail -1 LOG.test +grep -o ' name="test_[^"]*"' junit.xml diff --git a/acceptance/pipelines/test/live/test.toml b/acceptance/pipelines/test/live/test.toml new file mode 100644 index 00000000000..7d60b7bc94f --- /dev/null +++ b/acceptance/pipelines/test/live/test.toml @@ -0,0 +1,20 @@ +# Also runs against a real workspace, when that workspace has pipeline testing enabled. +# The script skips the cloud run unless TEST_PIPELINES_TESTING_ENABLED is set. +Cloud = true +CloudSlow = true +RecordRequests = false +Ignore = [".databricks", "databricks.yml", "junit.xml"] + +# Locally, the test server stands in for the service's test run. Start Update is the test server's. +[[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:01.000Z", "origin": {"test_case_id": "/Workspace/tests/test_orders.py::test_order_total"}, "details": {"test_case_progress": {"status": "PASSED"}}}, + {"id": "e2", "event_type": "test_run_summary", "timestamp": "2026-01-01T00:00:02.000Z", "details": {"test_run_summary": {"total": 1, "passed": 1, "failed": 0, "skipped": 0, "errored": 0}}} +]} +''' diff --git a/acceptance/pipelines/test/live/tests/test_orders.py b/acceptance/pipelines/test/live/tests/test_orders.py new file mode 100644 index 00000000000..89d502a0f45 --- /dev/null +++ b/acceptance/pipelines/test/live/tests/test_orders.py @@ -0,0 +1,6 @@ +# order_helpers is not on the import path; only its installation as a test dependency provides it. +from order_helpers import order_total + + +def test_order_total(): + assert order_total([1, 2, 3]) == 6 diff --git a/acceptance/pipelines/test/live/transformations/orders.py b/acceptance/pipelines/test/live/transformations/orders.py new file mode 100644 index 00000000000..c0ec524e2ee --- /dev/null +++ b/acceptance/pipelines/test/live/transformations/orders.py @@ -0,0 +1,6 @@ +from pyspark import pipelines as dp + + +@dp.materialized_view +def orders(): + return spark.range(3) diff --git a/acceptance/pipelines/test/no-test-sources/databricks.yml b/acceptance/pipelines/test/no-test-sources/databricks.yml new file mode 100644 index 00000000000..708fd010d78 --- /dev/null +++ b/acceptance/pipelines/test/no-test-sources/databricks.yml @@ -0,0 +1,10 @@ +bundle: + name: test-pipeline-test-no-test-sources + +resources: + pipelines: + my_pipeline: + name: test-pipeline-test-no-test-sources + libraries: + - file: + path: pipeline_file.py diff --git a/acceptance/pipelines/test/no-test-sources/out.test.toml b/acceptance/pipelines/test/no-test-sources/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/no-test-sources/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/no-test-sources/output.txt b/acceptance/pipelines/test/no-test-sources/output.txt new file mode 100644 index 00000000000..a5932edae92 --- /dev/null +++ b/acceptance/pipelines/test/no-test-sources/output.txt @@ -0,0 +1,17 @@ + +>>> [CLI] pipelines deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/test-pipeline-test-no-test-sources/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] + +=== The bundle now sets testing, so the service error comes with a hint to deploy + +>>> musterr [CLI] pipelines test --no-wait +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: [PIPELINE_TEST_NO_TEST_SOURCES] The pipeline has no test sources. Set `testing.sources` to the test files or folders to run. +The bundle sets testing for my_pipeline, but the deployed pipeline has none. Run 'databricks pipelines deploy' first diff --git a/acceptance/pipelines/test/no-test-sources/pipeline_file.py b/acceptance/pipelines/test/no-test-sources/pipeline_file.py new file mode 100644 index 00000000000..4a7a3a1878e --- /dev/null +++ b/acceptance/pipelines/test/no-test-sources/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/no-test-sources/script b/acceptance/pipelines/test/no-test-sources/script new file mode 100644 index 00000000000..901fe04b3e9 --- /dev/null +++ b/acceptance/pipelines/test/no-test-sources/script @@ -0,0 +1,11 @@ +# The pipeline is deployed before the bundle sets testing. +trace $CLI pipelines deploy +rm "$OUT_REQUESTS" +update_file.py databricks.yml " path: pipeline_file.py" " path: pipeline_file.py + testing: + sources: + - glob: + include: ./tests/**" + +title "The bundle now sets testing, so the service error comes with a hint to deploy\n" +trace musterr $CLI pipelines test --no-wait diff --git a/acceptance/pipelines/test/no-test-sources/test.toml b/acceptance/pipelines/test/no-test-sources/test.toml new file mode 100644 index 00000000000..c62a3b8ab00 --- /dev/null +++ b/acceptance/pipelines/test/no-test-sources/test.toml @@ -0,0 +1,10 @@ +# The deployed pipeline has no test sources. +[[Server]] +Pattern = "POST /api/2.0/pipelines/{pipeline_id}/updates" +Response.StatusCode = 400 +Response.Body = ''' +{ + "error_code": "INVALID_PARAMETER_VALUE", + "message": "[PIPELINE_TEST_NO_TEST_SOURCES] The pipeline has no test sources. Set `testing.sources` to the test files or folders to run." +} +''' diff --git a/acceptance/pipelines/test/no-test-sources/tests/test_orders.py b/acceptance/pipelines/test/no-test-sources/tests/test_orders.py new file mode 100644 index 00000000000..6058cdd0bba --- /dev/null +++ b/acceptance/pipelines/test/no-test-sources/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/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..46c03562359 --- /dev/null +++ b/acceptance/pipelines/test/pipeline-id/output.txt @@ -0,0 +1,50 @@ + +=== Relative test selections resolve against the pipeline root_path +>>> MSYS_NO_PATHCONV=1 [CLI] pipelines test --pipeline-id [UUID][0] tests/test_orders.py::test_backfill /Workspace/Users/someone@example.com/other/tests --no-wait +Pipeline: [UUID][0] +Update: [UUID][1] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID][0]/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][2] 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][2] /Workspace/Users/someone@example.com/tests/test_other.py::test_x --no-wait +Pipeline: [UUID][2] +Update: [UUID][3] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID][2]/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: Unable to locate the bundle root: databricks.yml not found. + +Run this command from a directory that contains a bundle, or set the DATABRICKS_BUNDLE_ROOT environment variable to the bundle root directory. +Alternatively, create a new bundle with 'databricks bundle init'. + diff --git a/acceptance/pipelines/test/pipeline-id/script b/acceptance/pipelines/test/pipeline-id/script new file mode 100644 index 00000000000..bf8ae479ce7 --- /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 --no-wait +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 --no-wait +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/results/failed/out.report.xml b/acceptance/pipelines/test/results/failed/out.report.xml new file mode 100644 index 00000000000..ec831adc1a1 --- /dev/null +++ b/acceptance/pipelines/test/results/failed/out.report.xml @@ -0,0 +1,15 @@ + + + + + + AssertionError: assert 100 == 150 + + + + + + RuntimeError: fixture orders setup failed + + + diff --git a/acceptance/pipelines/test/results/failed/out.test.toml b/acceptance/pipelines/test/results/failed/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/results/failed/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/results/failed/output.txt b/acceptance/pipelines/test/results/failed/output.txt new file mode 100644 index 00000000000..2a9924f3a89 --- /dev/null +++ b/acceptance/pipelines/test/results/failed/output.txt @@ -0,0 +1,21 @@ + +=== Failed, errored and skipped tests +>>> musterr [CLI] pipelines test --pipeline-id [UUID][0] --junit-xml out.report.xml +Pipeline: [UUID][0] +Update: [UUID][1] +Running tests... +1 of 4 PASSED tests/test_orders.py::test_valid_order ........................... [PASSED] +2 of 4 FAILED tests/test_orders.py::test_backfill .............................. [FAILED in 2.14s] +3 of 4 SKIPPED tests/test_orders.py::test_legacy_format ........................ [SKIPPED] +4 of 4 ERROR tests/test_orders.py::test_fixture ................................ [ERROR] + +Completed with 1 failed and 1 errored tests: + +Failure in tests/test_orders.py::test_backfill + AssertionError: assert 100 == 150 + +Error in tests/test_orders.py::test_fixture + RuntimeError: fixture orders setup failed + +JUnit report: out.report.xml +Done. PASSED=1 FAILED=1 SKIPPED=1 ERROR=1 TOTAL=4 diff --git a/acceptance/pipelines/test/results/failed/script b/acceptance/pipelines/test/results/failed/script new file mode 100644 index 00000000000..4dd938350c0 --- /dev/null +++ b/acceptance/pipelines/test/results/failed/script @@ -0,0 +1,2 @@ +title "Failed, errored and skipped tests" +trace musterr $CLI pipelines test --pipeline-id $pipeline_id --junit-xml out.report.xml diff --git a/acceptance/pipelines/test/results/failed/test.toml b/acceptance/pipelines/test/results/failed/test.toml new file mode 100644 index 00000000000..54707dfe879 --- /dev/null +++ b/acceptance/pipelines/test/results/failed/test.toml @@ -0,0 +1,25 @@ +[[Server]] +Pattern = "POST /api/2.0/pipelines/{pipeline_id}/updates" +Response.Body = '{"update_id": "00000000-0000-0000-0000-0000000000a1"}' + +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/updates/{update_id}" +Response.Body = '{"update": {"update_id": "00000000-0000-0000-0000-0000000000a1", "state": "FAILED"}}' + +[[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:00.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_backfill"}, "details": {"test_case_progress": {"status": "PENDING"}}}, + {"id": "e3", "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_legacy_format"}, "details": {"test_case_progress": {"status": "PENDING"}}}, + {"id": "e4", "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_fixture"}, "details": {"test_case_progress": {"status": "PENDING"}}}, + {"id": "e5", "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": "PASSED"}}}, + {"id": "e6", "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_backfill"}, "details": {"test_case_progress": {"status": "RUNNING"}}}, + {"id": "e7", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:03.140Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_backfill"}, "details": {"test_case_progress": {"status": "FAILED"}}, "level": "ERROR", "error": {"fatal": false, "exceptions": [{"class_name": "AssertionError", "message": "assert 100 == 150"}]}}, + {"id": "e8", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:04.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_legacy_format"}, "details": {"test_case_progress": {"status": "SKIPPED"}}}, + {"id": "e9", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:05.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_fixture"}, "details": {"test_case_progress": {"status": "ERROR"}}, "level": "ERROR", "error": {"fatal": false, "exceptions": [{"class_name": "RuntimeError", "message": "fixture orders setup failed"}]}}, + {"id": "e10", "event_type": "test_run_summary", "timestamp": "2026-01-01T00:00:06.000Z", "details": {"test_run_summary": {"total": 4, "passed": 1, "failed": 1, "skipped": 1, "errored": 1}}}, + {"id": "e11", "event_type": "update_progress", "timestamp": "2026-01-01T00:00:07.000Z", "level": "ERROR", "error": {"fatal": true, "exceptions": [{"class_name": "TestFailure", "message": "[PIPELINE_TEST_FAILED] 2 of the pipeline's tests failed (1 passed, 1 skipped)."}]}} +]} +''' diff --git a/acceptance/pipelines/test/results/incomplete/out.report.xml b/acceptance/pipelines/test/results/incomplete/out.report.xml new file mode 100644 index 00000000000..8bd8097922b --- /dev/null +++ b/acceptance/pipelines/test/results/incomplete/out.report.xml @@ -0,0 +1,12 @@ + + + + + + + + + RUN INCOMPLETE: the update failed before all test results were received. update [UUID][1] is FAILED: ComputeTerminatedException: Compute terminated unexpectedly. + + + diff --git a/acceptance/pipelines/test/results/incomplete/out.test.toml b/acceptance/pipelines/test/results/incomplete/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/results/incomplete/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/results/incomplete/output.txt b/acceptance/pipelines/test/results/incomplete/output.txt new file mode 100644 index 00000000000..ce9fddb75f9 --- /dev/null +++ b/acceptance/pipelines/test/results/incomplete/output.txt @@ -0,0 +1,12 @@ + +=== Compute ends before all results are received +>>> musterr [CLI] pipelines test --pipeline-id [UUID][0] --junit-xml out.report.xml +Pipeline: [UUID][0] +Update: [UUID][1] +Running tests... +1 of 2 PASSED tests/test_compute.py::test_small_order .......................... [PASSED] + +RUN INCOMPLETE: the update failed before all test results were received. +JUnit report: out.report.xml +Done. PASSED=1 FAILED=0 SKIPPED=0 ERROR=0 TOTAL=2 +Error: update [UUID][1] is FAILED: ComputeTerminatedException: Compute terminated unexpectedly. diff --git a/acceptance/pipelines/test/results/incomplete/script b/acceptance/pipelines/test/results/incomplete/script new file mode 100644 index 00000000000..b28d2bdee5e --- /dev/null +++ b/acceptance/pipelines/test/results/incomplete/script @@ -0,0 +1,2 @@ +title "Compute ends before all results are received" +trace musterr $CLI pipelines test --pipeline-id $pipeline_id --junit-xml out.report.xml diff --git a/acceptance/pipelines/test/results/incomplete/test.toml b/acceptance/pipelines/test/results/incomplete/test.toml new file mode 100644 index 00000000000..d78f4db704d --- /dev/null +++ b/acceptance/pipelines/test/results/incomplete/test.toml @@ -0,0 +1,19 @@ +[[Server]] +Pattern = "POST /api/2.0/pipelines/{pipeline_id}/updates" +Response.Body = '{"update_id": "00000000-0000-0000-0000-0000000000a1"}' + +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/updates/{update_id}" +Response.Body = '{"update": {"update_id": "00000000-0000-0000-0000-0000000000a1", "state": "FAILED"}}' + +[[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_compute.py::test_small_order"}, "details": {"test_case_progress": {"status": "PENDING"}}}, + {"id": "e2", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:00.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_compute.py::test_large_order"}, "details": {"test_case_progress": {"status": "PENDING"}}}, + {"id": "e3", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:01.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_compute.py::test_small_order"}, "details": {"test_case_progress": {"status": "PASSED"}}}, + {"id": "e4", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:01.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_compute.py::test_large_order"}, "details": {"test_case_progress": {"status": "RUNNING"}}}, + {"id": "e5", "event_type": "update_progress", "timestamp": "2026-01-01T00:00:02.000Z", "level": "ERROR", "error": {"fatal": true, "exceptions": [{"class_name": "ComputeTerminatedException", "message": "Compute terminated unexpectedly."}]}} +]} +''' diff --git a/acceptance/pipelines/test/results/passed/out.report.xml b/acceptance/pipelines/test/results/passed/out.report.xml new file mode 100644 index 00000000000..2ece15eb5ea --- /dev/null +++ b/acceptance/pipelines/test/results/passed/out.report.xml @@ -0,0 +1,7 @@ + + + + + + + diff --git a/acceptance/pipelines/test/results/passed/out.test.toml b/acceptance/pipelines/test/results/passed/out.test.toml new file mode 100644 index 00000000000..18b1a88417e --- /dev/null +++ b/acceptance/pipelines/test/results/passed/out.test.toml @@ -0,0 +1 @@ +Cloud = false diff --git a/acceptance/pipelines/test/results/passed/output.txt b/acceptance/pipelines/test/results/passed/output.txt new file mode 100644 index 00000000000..9a5bafe8474 --- /dev/null +++ b/acceptance/pipelines/test/results/passed/output.txt @@ -0,0 +1,31 @@ + +=== All tests pass +>>> [CLI] pipelines test --pipeline-id [UUID][0] --junit-xml out.report.xml +Pipeline: [UUID][0] +Update: [UUID][1] +Running tests... +1 of 2 PASSED tests/test_orders.py::test_valid_order ........................... [PASSED in 1.84s] +2 of 2 PASSED tests/test_orders.py::test_backfill .............................. [PASSED in 1.00s] + +JUnit report: out.report.xml +Done. PASSED=2 FAILED=0 SKIPPED=0 ERROR=0 TOTAL=2 + +>>> print_requests.py --get //events +{ + "method": "GET", + "path": "/api/2.0/pipelines/[UUID][0]/events", + "q": { + "filter": "update_id = '[UUID][1]'", + "max_results": "250", + "order_by": "timestamp asc" + } +} +{ + "method": "GET", + "path": "/api/2.0/pipelines/[UUID][0]/events", + "q": { + "filter": "update_id = '[UUID][1]'", + "max_results": "250", + "order_by": "timestamp asc" + } +} diff --git a/acceptance/pipelines/test/results/passed/script b/acceptance/pipelines/test/results/passed/script new file mode 100644 index 00000000000..d7132c4dbc3 --- /dev/null +++ b/acceptance/pipelines/test/results/passed/script @@ -0,0 +1,3 @@ +title "All tests pass" +trace $CLI pipelines test --pipeline-id $pipeline_id --junit-xml out.report.xml +trace print_requests.py --get //events diff --git a/acceptance/pipelines/test/results/passed/test.toml b/acceptance/pipelines/test/results/passed/test.toml new file mode 100644 index 00000000000..f821ca318eb --- /dev/null +++ b/acceptance/pipelines/test/results/passed/test.toml @@ -0,0 +1,21 @@ +[[Server]] +Pattern = "POST /api/2.0/pipelines/{pipeline_id}/updates" +Response.Body = '{"update_id": "00000000-0000-0000-0000-0000000000a1"}' + +[[Server]] +Pattern = "GET /api/2.0/pipelines/{pipeline_id}/updates/{update_id}" +Response.Body = '{"update": {"update_id": "00000000-0000-0000-0000-0000000000a1", "state": "COMPLETED"}}' + +[[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:00.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_backfill"}, "details": {"test_case_progress": {"status": "PENDING"}}}, + {"id": "e3", "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": "e4", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:02.840Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_valid_order"}, "details": {"test_case_progress": {"status": "PASSED"}}}, + {"id": "e5", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:03.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_backfill"}, "details": {"test_case_progress": {"status": "RUNNING"}}}, + {"id": "e6", "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:04.000Z", "origin": {"test_case_id": "/Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_backfill"}, "details": {"test_case_progress": {"status": "PASSED"}}}, + {"id": "e7", "event_type": "test_run_summary", "timestamp": "2026-01-01T00:00:05.000Z", "details": {"test_run_summary": {"total": 2, "passed": 2, "failed": 0, "skipped": 0, "errored": 0}}} +]} +''' diff --git a/acceptance/pipelines/test/results/script.prepare b/acceptance/pipelines/test/results/script.prepare new file mode 100644 index 00000000000..caaa90ba318 --- /dev/null +++ b/acceptance/pipelines/test/results/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/canceled/out.report.xml b/acceptance/pipelines/test/run-control/canceled/out.report.xml new file mode 100644 index 00000000000..3743b6239a3 --- /dev/null +++ b/acceptance/pipelines/test/run-control/canceled/out.report.xml @@ -0,0 +1,11 @@ + + + + + + + + RUN INCOMPLETE: the update was canceled before all test results were received. update [UUID][1] is CANCELED + + + 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..a21b056a361 --- /dev/null +++ b/acceptance/pipelines/test/run-control/canceled/output.txt @@ -0,0 +1,10 @@ + +=== A canceled update is an incomplete run +>>> musterr [CLI] pipelines test --pipeline-id [UUID][0] --resume [UUID][1] --junit-xml out.report.xml +Pipeline: [UUID][0] +Update: [UUID][1] + +RUN INCOMPLETE: the update was canceled before all test results were received. +JUnit report: out.report.xml +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..beb99f5b049 --- /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 --junit-xml out.report.xml 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.report.xml b/acceptance/pipelines/test/run-control/timeout/out.report.xml new file mode 100644 index 00000000000..ac0d802ec6f --- /dev/null +++ b/acceptance/pipelines/test/run-control/timeout/out.report.xml @@ -0,0 +1,11 @@ + + + + + + + + stopped waiting for update [UUID][1] after 1s; the update keeps running, to get its results run: databricks pipelines test --pipeline-id [UUID][0] --junit-xml out.report.xml --resume [UUID][1] + + + 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..2be2f36818e --- /dev/null +++ b/acceptance/pipelines/test/run-control/timeout/output.txt @@ -0,0 +1,8 @@ + +=== --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 --junit-xml out.report.xml +Pipeline: [UUID][0] +Update: [UUID][1] +Running tests... +JUnit report: out.report.xml +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] --junit-xml out.report.xml --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..b95c8689fdd --- /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 --junit-xml out.report.xml 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/acceptance/pipelines/test/selection/databricks.yml b/acceptance/pipelines/test/selection/databricks.yml new file mode 100644 index 00000000000..ea0662b1043 --- /dev/null +++ b/acceptance/pipelines/test/selection/databricks.yml @@ -0,0 +1,14 @@ +bundle: + name: test-pipeline-test-selection + +resources: + pipelines: + my_pipeline: + name: test-pipeline-test-selection + libraries: + - file: + path: pipeline_file.py + testing: + sources: + - glob: + include: ./tests/** 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..ae6afcb4754 --- /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][0]?w=[WORKSPACE_ID] + +=== Run all tests of the only pipeline +>>> [CLI] pipelines test --no-wait +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][0]) +Update: [UUID][1] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID][0]/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 --no-wait +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][0]) +Update: [UUID][2] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID][0]/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 --no-wait +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][0]) +Update: [UUID][3] + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID][0]/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: "tests/test_orders.py" is not a pipeline in this bundle (pipelines: my_pipeline). Usage: databricks pipelines test my_pipeline tests/test_orders.py 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..3344ef8a37d --- /dev/null +++ b/acceptance/pipelines/test/selection/script @@ -0,0 +1,19 @@ +trace $CLI pipelines deploy +rm "$OUT_REQUESTS" + +title "Run all tests of the only pipeline" +trace $CLI pipelines test --no-wait +trace print_requests.py //updates + +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 --no-wait +trace print_requests.py //updates + +title "Relative paths resolve from the working directory" +cd tests +trace $CLI pipelines test my_pipeline test_orders.py::test_valid_order --no-wait +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.start-failed.xml b/acceptance/pipelines/test/submit-error/out.start-failed.xml new file mode 100644 index 00000000000..be5603014cf --- /dev/null +++ b/acceptance/pipelines/test/submit-error/out.start-failed.xml @@ -0,0 +1,8 @@ + + + + + An active update '[UUID][1]' already exists for pipeline '[UUID][2]'. + + + 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..909f46a871b --- /dev/null +++ b/acceptance/pipelines/test/submit-error/output.txt @@ -0,0 +1,13 @@ + +=== A submit error from the service is shown as is, no update ID is printed, and an older report is replaced +>>> musterr [CLI] pipelines test --pipeline-id [UUID][0] --junit-xml out.start-failed.xml +Error: An active update '[UUID][1]' already exists for pipeline '[UUID][2]'. + +>>> print_requests.py //updates +{ + "method": "POST", + "path": "/api/2.0/pipelines/[UUID][0]/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..1035f977d33 --- /dev/null +++ b/acceptance/pipelines/test/submit-error/script @@ -0,0 +1,7 @@ +pipeline_id=$($CLI pipelines create --json '{"name": "orders", "root_path": "/Workspace/Users/someone@example.com/orders"}' | jq -r .pipeline_id) +rm "$OUT_REQUESTS" + +title "A submit error from the service is shown as is, no update ID is printed, and an older report is replaced" +echo '' > out.start-failed.xml +trace musterr $CLI pipelines test --pipeline-id $pipeline_id --junit-xml out.start-failed.xml +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..8932d1cfc26 --- /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 '11111111-2222-3333-4444-555555555555'." +} +''' +Response.StatusCode = 409 diff --git a/acceptance/pipelines/test/test.toml b/acceptance/pipelines/test/test.toml new file mode 100644 index 00000000000..4b2b5e84b92 --- /dev/null +++ b/acceptance/pipelines/test/test.toml @@ -0,0 +1,6 @@ +RecordRequests = true +Ignore = ['.databricks'] +# The command is experimental; the gate case in arguments turns it off. +Env.DATABRICKS_CLI_PIPELINES_TEST = "true" +# The bundles declare the testing property of their pipeline. +Env.DATABRICKS_BUNDLE_ENABLE_EXPERIMENTAL_PIPELINE_TESTING = "true" 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..83fdf0b11ee --- /dev/null +++ b/cmd/pipelines/test.go @@ -0,0 +1,773 @@ +package pipelines + +import ( + "context" + "errors" + "fmt" + "io" + "io/fs" + "maps" + "net/http" + "os" + "os/signal" + "path" + "path/filepath" + "slices" + "strings" + "syscall" + "time" + "uuid" + + "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/apierr" + "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 strings.TrimSpace(arg) == "": + return errors.New("--pytest-arg requires a non-empty value") + 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 and are +// passed as typed: the service checks them and reports errors about what the user wrote. +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, "/") { + 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 appends a relative path to the pipeline's root_path without cleaning it, +// so that the service checks the path as the user typed it. +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 strings.TrimSuffix(rootPath, "/") + "/" + filepath.ToSlash(relPath), nil +} + +// relativeTestID shows a test ID whose file is under folder relative to that folder, as pytest names +// tests from the project root. Other IDs are returned unchanged. The service reports IDs with the +// /Workspace prefix, which a folder may lack, so both are compared without it. +func relativeTestID(folder, id string) string { + if folder == "" { + return id + } + prefix := strings.TrimSuffix(strings.TrimPrefix(folder, "/Workspace"), "/") + "/" + if rel, ok := strings.CutPrefix(strings.TrimPrefix(id, "/Workspace"), prefix); ok { + return rel + } + return id +} + +// 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 +} + +// 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) + 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 +} + +// pipelineTestNoTestSources is the error class of Start Update for a pipeline without testing.sources. +const pipelineTestNoTestSources = "PIPELINE_TEST_NO_TEST_SOURCES" + +// withDeployHint adds a hint to deploy when Start Update says the pipeline has no test sources but +// the bundle sets testing: the bundle was changed after its last deploy. +func withDeployHint(cmd *cobra.Command, t testTarget, err error) error { + apiErr, ok := errors.AsType[*apierr.APIError](err) + if !t.bundleHasTesting || !ok { + return err + } + // The class is the error code or, for errors with a generic code, the [CLASS] prefix of the message. + if apiErr.ErrorCode != pipelineTestNoTestSources && !strings.Contains(apiErr.Message, pipelineTestNoTestSources) { + return err + } + return fmt.Errorf("%w\nThe bundle sets testing for %s, but the deployed pipeline has none. Run '%s deploy' first", err, t.key, cmd.Parent().CommandPath()) +} + +// testTarget is the pipeline a test run is submitted to. +type testTarget struct { + key string + pipelineID string + selections []string + w *databricks.WorkspaceClient + + // bundleHasTesting is true when the bundle sets testing for the pipeline. + bundleHasTesting bool + + // projectFolder is the deployed folder that test IDs are shown relative to: the bundle's + // synchronized files, or root_path. It is "" when unknown. + projectFolder string +} + +func (t testTarget) shortTestID(id string) string { + return relativeTestID(t.projectFolder, id) +} + +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, wait bool) (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) + + // root_path resolves relative selections and shortens the test IDs printed while waiting. + var rootPath string + if wait || 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, projectFolder: rootPath}, nil +} + +// resolveTestTargetInBundle identifies the pipeline by its bundle KEY, which must precede any selections. +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 { + 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() + + pipelineID, err := resolvePipelineIdFromKey(ctx, b, key) + if err != nil { + if _, lookupErr := resources.Lookup(b, key, isPipeline); lookupErr != nil { + return testTarget{}, notPipelineKeyError(cmd, b, key, args) + } + return testTarget{}, err + } + if !resume { + 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 + } + pipeline := b.Config.Resources.Pipelines[key] + return testTarget{ + key: key, + pipelineID: pipelineID, + selections: selections, + w: b.WorkspaceClient(ctx), + bundleHasTesting: pipeline != nil && pipeline.Testing != nil, + 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)) + } + // Reading the results again replaces the partial report. + if flag := cmd.Flag("junit-xml"); flag != nil && flag.Value.String() != "" { + parts = append(parts, "--junit-xml", shellquote.BashArg(flag.Value.String())) + } + // --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. 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) + if err != nil { + return "", err + } + if response.Update == nil { + return "", fmt.Errorf("no update found with id %s for pipeline %s", updateID, pipelineID) + } + state := response.Update.State + printer.updateState(state) + + // Reading from the last timestamp returns its events again; the run ignores duplicates. + events, err := readTestEvents(ctx, w, pipelineID, updateID, lastTimestamp) + if err != nil { + return "", err + } + if len(events) > 0 { + lastTimestamp = events[len(events)-1].Timestamp + } + addTestEvents(run, printer, events) + + switch state { + case pipelines.UpdateInfoStateCompleted, pipelines.UpdateInfoStateFailed, pipelines.UpdateInfoStateCanceled: + // Events can become readable after later ones, so read all of them once more. + events, err := readTestEvents(ctx, w, pipelineID, updateID, "") + if err != nil { + return "", err + } + addTestEvents(run, printer, events) + return state, nil + default: + } + + select { + case <-ctx.Done(): + return "", ctx.Err() + case <-timedOut: + return "", errTestWaitTimeout + case <-time.After(testPollInterval): + } + } +} + +// addTestEvents adds events to the run and prints each case that gets its final outcome. +func addTestEvents(run *testRun, printer *testProgressPrinter, events []testPipelineEvent) { + for _, e := range events { + if c := run.add(e); c != nil { + printer.caseFinished(c, len(run.cases)) + } + } +} + +// updateError returns the error that the update reported: its own, or else the test verdict. +func (r *testRun) updateError() string { + if r.fatalError != "" { + return r.fatalError + } + return r.verdictError +} + +// testRunError returns the error the command exits with after the update ended, or nil if at +// least one test was collected and every collected test passed or was skipped. +func testRunError(updateID string, state pipelines.UpdateInfoState, run *testRun, incompleteReason string) error { + if incompleteReason == "" { + summary := run.summaries[0] + testsFailed := summary.Total == 0 || summary.Failed > 0 || summary.Errored > 0 + if !testsFailed && state == pipelines.UpdateInfoStateCompleted && run.fatalError == "" { + return nil + } + // A test update fails when its tests fail; the printed result explains it, unless the + // update ended in another way or reports an error of its own. The service's own verdict, + // such as "[PIPELINE_TEST_FAILED] 1 of the pipeline's tests failed", is not. + ended := state == pipelines.UpdateInfoStateFailed || state == pipelines.UpdateInfoStateCompleted + if testsFailed && ended && run.fatalError == "" { + return root.ErrAlreadyPrinted + } + } else if state == pipelines.UpdateInfoStateCompleted { + return root.ErrAlreadyPrinted + } + if updateError := run.updateError(); updateError != "" { + return fmt.Errorf("update %s is %s: %s", updateID, state, updateError) + } + return fmt.Errorf("update %s is %s", updateID, state) +} + +// testRunProblem describes why a run that ended did not pass for a reason other than +// failed tests, or returns "" otherwise. +func testRunProblem(run *testRun, incompleteReason string, runErr error) string { + var lines []string + switch { + case incompleteReason != "": + lines = append(lines, fmt.Sprintf("RUN INCOMPLETE: %s.", incompleteReason)) + case run.summaries[0].Total == 0: + lines = append(lines, "No tests were collected.") + } + if runErr != nil && !errors.Is(runErr, root.ErrAlreadyPrinted) { + lines = append(lines, runErr.Error()) + } + return strings.Join(lines, "\n") +} + +// testFlags holds the flags of pipelines test. +type testFlags struct { + pipelineID string + pytestArgs []string + noWait bool + timeout time.Duration + resumeUpdateID string + junitPath string +} + +// checkTestFlags rejects the flag combinations that pipelines test does not support, before any API call. +func checkTestFlags(cmd *cobra.Command, args []string, f testFlags) 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(f.pytestArgs); err != nil { + return err + } + if f.timeout < 0 { + return errors.New("--timeout must not be negative") + } + if f.noWait && f.timeout > 0 { + return errors.New("--timeout cannot be used with --no-wait") + } + if f.noWait && f.junitPath != "" { + return errors.New("--junit-xml cannot be used with --no-wait because no results are available when the command returns") + } + // An empty --resume, such as from an unset shell variable, must not start a new run. + if cmd.Flags().Changed("resume") { + selectionCount := len(args) + if f.pipelineID == "" { + selectionCount = max(len(args)-1, 0) + } + switch { + case f.noWait: + return errors.New("--resume cannot be used with --no-wait") + case len(f.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(f.resumeUpdateID); err != nil { + return fmt.Errorf("--resume %q is not an update ID; pass the update ID that pipelines test printed", f.resumeUpdateID) + } + } + return nil +} + +// resolveTestTarget identifies the pipeline by --pipeline-id or, in a bundle, by its KEY. +func resolveTestTarget(cmd *cobra.Command, args []string, f testFlags, resume bool) (testTarget, error) { + if f.pipelineID != "" { + id, err := uuid.Parse(f.pipelineID) + if err != nil { + return testTarget{}, fmt.Errorf("--pipeline-id %q is not a pipeline ID", f.pipelineID) + } + return resolveTestTargetByID(cmd, args, id.String(), !f.noWait) + } + cmd.SetContext(logdiag.InitContext(cmd.Context())) + return resolveTestTargetInBundle(cmd, args, resume) +} + +// notPipelineKeyError explains that the first argument is not a pipeline KEY, names the bundle's +// pipelines, and shows the command that passes the arguments as test selections of the first one. +func notPipelineKeyError(cmd *cobra.Command, b *bundle.Bundle, key string, args []string) error { + keys := slices.Sorted(maps.Keys(resources.Completions(b, isPipeline))) + if len(keys) == 0 { + return fmt.Errorf("%q is not a pipeline in this bundle, and the bundle has no pipelines", key) + } + usage := []string{cmd.CommandPath(), shellquote.BashArg(keys[0])} + for _, arg := range args { + usage = append(usage, shellquote.BashArg(arg)) + } + return fmt.Errorf("%q is not a pipeline in this bundle (pipelines: %s). Usage: %s", key, strings.Join(keys, ", "), strings.Join(usage, " ")) +} + +func testCommand() *cobra.Command { + cmd := &cobra.Command{ + Use: "test [flags] [KEY] [TEST_SELECTION...]", + 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. + +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' + +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. + +--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'. + +--junit-xml writes one entry per test with its outcome and duration. An +incomplete run, a run without tests, or a timeout adds a failing entry that +names the cause. The report replaces the file on every exit except a flag that +cannot be parsed. With workspace.sync_mode immutable, test names keep the path +of the deployed snapshot, which changes with each deployment.`, + } + + var f testFlags + + testGroup := cmdgroup.NewFlagGroup("Pipeline Test") + testGroup.FlagSet().StringVar(&f.pipelineID, "pipeline-id", "", "ID of the pipeline to test, instead of a KEY.") + testGroup.FlagSet().StringArrayVar(&f.pytestArgs, "pytest-arg", nil, "Argument to pass to pytest (repeatable). Only -k and -m are supported.") + testGroup.FlagSet().BoolVar(&f.noWait, "no-wait", false, "Don't wait for the tests to complete.") + testGroup.FlagSet().DurationVar(&f.timeout, "timeout", 0, "Stop waiting for results after this duration, for example 30m. The update keeps running.") + testGroup.FlagSet().StringVar(&f.resumeUpdateID, "resume", "", "Read the results of an existing test update instead of starting one.") + testGroup.FlagSet().StringVar(&f.junitPath, "junit-xml", "", "Write the results as a JUnit XML report to this file, replacing it.") + + wrappedCmd := cmdgroup.NewCommandWithGroupFlag(cmd) + wrappedCmd.AddFlagGroup(testGroup) + + cmd.SetFlagErrorFunc(testFlagErrorFunc) + + cmd.RunE = func(cmd *cobra.Command, args []string) error { + // Every exit writes the report, so that an older report never survives. + report := &testReport{path: f.junitPath, out: cmd.OutOrStdout(), suiteName: f.pipelineID} + if report.suiteName == "" && len(args) > 0 { + report.suiteName = args[0] + } + if err := checkTestFlags(cmd, args, f); err != nil { + return report.failed(cmd.Context(), err) + } + resume := cmd.Flags().Changed("resume") + target, err := resolveTestTarget(cmd, args, f, resume) + if err != nil { + return report.failed(cmd.Context(), err) + } + report.suiteName = target.key + if report.suiteName == "" { + report.suiteName = target.pipelineID + } + report.projectFolder = target.projectFolder + ctx := cmd.Context() + + // 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 := f.resumeUpdateID + if !resume { + req := newStartTestUpdateRequest(target.selections, f.pytestArgs) + updateID, err = startTestUpdate(ctx, target.w, target.pipelineID, req) + if err != nil { + return report.failed(ctx, withDeployHint(cmd, target, err)) + } + } else if err := checkTestUpdate(ctx, target.w, target.pipelineID, updateID); err != nil { + return report.failed(ctx, err) + } + out := cmd.OutOrStdout() + target.printHeader(out, updateID) + if f.noWait { + return nil + } + + run := newTestRun() + printer := &testProgressPrinter{out: out, shortID: target.shortTestID} + // 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 f.timeout > 0 { + timedOut = time.After(f.timeout) + } + state, err := waitForTestUpdate(waitCtx, target.w, target.pipelineID, updateID, run, printer, timedOut) + if err != nil { + isTimeout := errors.Is(err, errTestWaitTimeout) + if !isTimeout && waitCtx.Err() == nil { + return report.finish(updateID, run, err.Error(), err) + } + resumeCmd := target.resumeCommand(cmd, updateID) + stopErr := fmt.Errorf("stopped waiting for update %s; the update keeps running, to get its results run: %s", updateID, resumeCmd) + if isTimeout { + stopErr = fmt.Errorf("stopped waiting for update %s after %s; the update keeps running, to get its results run: %s", updateID, f.timeout, resumeCmd) + } + return report.finish(updateID, run, stopErr.Error(), stopErr) + } + incompleteReason := run.incompleteReason(state) + renderTestResult(out, run, incompleteReason, target.shortTestID) + runErr := testRunError(updateID, state, run, incompleteReason) + // The report line comes before the summary, so that the summary stays the last line. + err = report.finish(updateID, run, testRunProblem(run, incompleteReason, runErr), runErr) + renderTestSummaryLine(out, run, incompleteReason) + return err + } + + return cmd +} diff --git a/cmd/pipelines/test_events.go b/cmd/pipelines/test_events.go new file mode 100644 index 00000000000..e9dd88aff70 --- /dev/null +++ b/cmd/pipelines/test_events.go @@ -0,0 +1,317 @@ +package pipelines + +import ( + "context" + "fmt" + "net/http" + "strings" + "time" + + "github.com/databricks/cli/libs/auth" + databricks "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/client" + "github.com/databricks/databricks-sdk-go/service/pipelines" +) + +const ( + testCaseProgressEventType = "test_case_progress" + testRunSummaryEventType = "test_run_summary" +) + +type testCaseStatus string + +const ( + testCasePending testCaseStatus = "PENDING" + testCaseRunning testCaseStatus = "RUNNING" + testCasePassed testCaseStatus = "PASSED" + testCaseFailed testCaseStatus = "FAILED" + testCaseSkipped testCaseStatus = "SKIPPED" + testCaseError testCaseStatus = "ERROR" +) + +// parseTestCaseStatus accepts both the short status and the proto enum value name. +func parseTestCaseStatus(s string) testCaseStatus { + return testCaseStatus(strings.TrimPrefix(s, "PIPELINE_TEST_CASE_STATUS_")) +} + +func (s testCaseStatus) isFinal() bool { + switch s { + case testCasePassed, testCaseFailed, testCaseSkipped, testCaseError: + return true + default: + return false + } +} + +type testRunSummary struct { + Total int `json:"total"` + Passed int `json:"passed"` + Failed int `json:"failed"` + Skipped int `json:"skipped"` + Errored int `json:"errored"` +} + +// testPipelineEvent is an entry of List Pipeline Events. The Go SDK PipelineEvent has no +// details or origin.test_case_id, so test events are decoded from the raw response. +type testPipelineEvent struct { + Id string `json:"id"` + EventType string `json:"event_type"` + Timestamp string `json:"timestamp"` + Origin struct { + TestCaseId string `json:"test_case_id"` + } `json:"origin"` + Error *pipelines.ErrorDetail `json:"error"` + Truncation *pipelines.Truncation `json:"truncation"` + Details struct { + TestCaseProgress *struct { + Status string `json:"status"` + } `json:"test_case_progress"` + TestRunSummary *testRunSummary `json:"test_run_summary"` + } `json:"details"` +} + +func (e testPipelineEvent) isTruncated(field string) bool { + if e.Truncation == nil { + return false + } + for _, f := range e.Truncation.TruncatedFields { + if f.FieldName == field { + return true + } + } + return false +} + +// testEventsPageSize is the largest page List Pipeline Events returns. +const testEventsPageSize = 250 + +// readTestEvents returns the events of one update in timestamp order, following every page. With +// sinceTimestamp it returns only the events at or after that timestamp, which must be a timestamp +// string exactly as the API returned it. +func readTestEvents(ctx context.Context, w *databricks.WorkspaceClient, pipelineID, updateID, sinceTimestamp string) ([]testPipelineEvent, error) { + apiClient, err := client.New(w.Config) + if err != nil { + return nil, fmt.Errorf("failed to create API client: %w", err) + } + path := fmt.Sprintf("/api/2.0/pipelines/%s/events", pipelineID) + request := pipelines.ListPipelineEventsRequest{ + Filter: buildPipelineEventFilter(updateID, nil, nil, sinceTimestamp, ""), + OrderBy: []string{"timestamp asc"}, + MaxResults: testEventsPageSize, + } + var events []testPipelineEvent + for { + var page struct { + Events []testPipelineEvent `json:"events"` + NextPageToken string `json:"next_page_token"` + } + err := apiClient.Do(ctx, http.MethodGet, path, auth.WorkspaceIDHeaders(w.Config), nil, request, &page) + if err != nil { + return nil, fmt.Errorf("failed to fetch pipeline events: %w", err) + } + events = append(events, page.Events...) + if page.NextPageToken == "" || len(page.Events) == 0 { + return events, nil + } + // A request with page_token must not repeat filter or order_by: the token carries them, + // and the service rejects the request otherwise. + request = pipelines.ListPipelineEventsRequest{PageToken: page.NextPageToken, MaxResults: testEventsPageSize} + } +} + +// testCaseResult is what the events report about one test case. +type testCaseResult struct { + // id is origin.test_case_id: the absolute workspace path of the test file followed by the pytest + // test ID, such as /Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_backfill. + id string + status testCaseStatus + // finals counts final outcomes; a complete result has exactly one per case. + finals int + started time.Time + finished time.Time + errorDetail *pipelines.ErrorDetail + detailsTruncated bool +} + +// duration returns the time between the RUNNING and the final event, if both are known. +func (c *testCaseResult) duration() (time.Duration, bool) { + if c.started.IsZero() || c.finished.IsZero() || c.finished.Before(c.started) { + return 0, false + } + return c.finished.Sub(c.started), true +} + +// testRun accumulates the events of a test update, ignoring events it has already seen. +type testRun struct { + seen map[string]bool + cases []*testCaseResult + byID map[string]*testCaseResult + summaries []testRunSummary + // unreadable counts test events whose outcome is missing. + unreadable int + // fatalError is the first fatal error that the update reported, other than the test verdict. + fatalError string + // verdictError is the service's verdict that tests failed, if the update reported it. + verdictError string +} + +// testVerdictErrorPrefix starts the message of the error by which the service reports that tests +// failed, such as "[PIPELINE_TEST_FAILED] 1 of the pipeline's tests failed (1 passed, 0 skipped).". +// The message is matched because ErrorDetail has no error class yet; only the output depends on it. +const testVerdictErrorPrefix = "[PIPELINE_TEST_FAILED]" + +// isTestVerdict reports whether an update error is the service's verdict that tests failed. +func isTestVerdict(e *pipelines.ErrorDetail) bool { + for _, exception := range e.Exceptions { + if strings.HasPrefix(exception.Message, testVerdictErrorPrefix) { + return true + } + } + return false +} + +func newTestRun() *testRun { + return &testRun{seen: map[string]bool{}, byID: map[string]*testCaseResult{}} +} + +// add records an event and returns the case it gave a final outcome to, if any. +func (r *testRun) add(e testPipelineEvent) *testCaseResult { + if r.seen[e.Id] { + return nil + } + r.seen[e.Id] = true + + if e.Error != nil && e.Error.Fatal { + switch { + case isTestVerdict(e.Error): + r.verdictError = errorDetailText(e.Error) + case r.fatalError == "": + r.fatalError = errorDetailText(e.Error) + } + } + + switch e.EventType { + case testRunSummaryEventType: + if e.Details.TestRunSummary == nil { + r.unreadable++ + return nil + } + r.summaries = append(r.summaries, *e.Details.TestRunSummary) + case testCaseProgressEventType: + progress := e.Details.TestCaseProgress + if progress == nil || e.Origin.TestCaseId == "" { + r.unreadable++ + return nil + } + status := parseTestCaseStatus(progress.Status) + if !status.isFinal() && status != testCasePending && status != testCaseRunning { + r.unreadable++ + return nil + } + + c, ok := r.byID[e.Origin.TestCaseId] + if !ok { + c = &testCaseResult{id: e.Origin.TestCaseId, status: testCasePending} + r.byID[c.id] = c + r.cases = append(r.cases, c) + } + timestamp, _ := time.Parse(time.RFC3339Nano, e.Timestamp) + switch { + case status == testCaseRunning: + c.started = timestamp + if !c.status.isFinal() { + c.status = testCaseRunning + } + case status.isFinal(): + c.finals++ + c.status = status + c.finished = timestamp + c.errorDetail = e.Error + c.detailsTruncated = e.isTruncated("error") + if c.finals == 1 { + return c + } + } + } + return nil +} + +// counts returns the number of cases with each final outcome. +func (r *testRun) counts() testRunSummary { + var s testRunSummary + for _, c := range r.cases { + if !c.status.isFinal() { + continue + } + s.Total++ + switch c.status { + case testCasePassed: + s.Passed++ + case testCaseFailed: + s.Failed++ + case testCaseSkipped: + s.Skipped++ + case testCaseError: + s.Errored++ + default: + } + } + return s +} + +// incompleteReason explains why the events of an ended update are not a complete result, +// or returns "" if they are: one summary and one final outcome per case, matching its counts. +func (r *testRun) incompleteReason(state pipelines.UpdateInfoState) string { + if r.unreadable > 0 { + return "some test results could not be read" + } + switch len(r.summaries) { + case 0: + switch state { + case pipelines.UpdateInfoStateFailed: + return "the update failed before all test results were received" + case pipelines.UpdateInfoStateCanceled: + return "the update was canceled before all test results were received" + default: + return "the update ended without a test summary" + } + case 1: + default: + return fmt.Sprintf("received %d test summaries instead of one", len(r.summaries)) + } + + for _, c := range r.cases { + if c.finals > 1 { + return fmt.Sprintf("test %s has more than one result", c.id) + } + } + summary := r.summaries[0] + received := r.counts() + if received.Total != summary.Total { + return fmt.Sprintf("received %d of %d test results", received.Total, summary.Total) + } + if received != summary { + return "the test results do not match the test summary" + } + for _, c := range r.cases { + if c.finals == 0 { + return fmt.Sprintf("test %s has no result", c.id) + } + } + return "" +} + +func errorDetailText(detail *pipelines.ErrorDetail) string { + var parts []string + for _, e := range detail.Exceptions { + switch { + case e.ClassName != "" && e.Message != "": + parts = append(parts, e.ClassName+": "+e.Message) + case e.Message != "": + parts = append(parts, e.Message) + case e.ClassName != "": + parts = append(parts, e.ClassName) + } + } + return strings.Join(parts, "\n") +} diff --git a/cmd/pipelines/test_events_test.go b/cmd/pipelines/test_events_test.go new file mode 100644 index 00000000000..fb0288b26c8 --- /dev/null +++ b/cmd/pipelines/test_events_test.go @@ -0,0 +1,254 @@ +package pipelines + +import ( + "encoding/json" + "fmt" + "net/url" + "strconv" + "strings" + "testing" + "time" + + "github.com/databricks/cli/libs/testserver" + databricks "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/service/pipelines" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// testEvents builds the events of one test, each with its own event ID. +type testEvents struct { + t *testing.T + n int +} + +func newTestEvents(t *testing.T) *testEvents { + return &testEvents{t: t} +} + +func (b *testEvents) testCase(id, status, timestamp string, extra ...string) testPipelineEvent { + b.n++ + return decodeTestEvent(b.t, fmt.Sprintf(`{ + "id": "e%d", "event_type": "test_case_progress", "timestamp": %q, + "origin": {"test_case_id": %q}, + "details": {"test_case_progress": {"status": %q}}%s + }`, b.n, timestamp, id, status, strings.Join(append([]string{""}, extra...), ","))) +} + +func (b *testEvents) summary(total, passed, failed, skipped, errored int) testPipelineEvent { + b.n++ + return decodeTestEvent(b.t, fmt.Sprintf(`{ + "id": "e%d", "event_type": "test_run_summary", + "details": {"test_run_summary": {"total": %d, "passed": %d, "failed": %d, "skipped": %d, "errored": %d}} + }`, b.n, total, passed, failed, skipped, errored)) +} + +func decodeTestEvent(t *testing.T, data string) testPipelineEvent { + var e testPipelineEvent + require.NoError(t, json.Unmarshal([]byte(data), &e)) + return e +} + +func TestTestRunAdd(t *testing.T) { + ev := newTestEvents(t) + run := newTestRun() + events := []testPipelineEvent{ + ev.testCase("tests/test_a.py::test_ok", "PENDING", "2026-01-01T00:00:00.000Z"), + ev.testCase("tests/test_a.py::test_ok", "RUNNING", "2026-01-01T00:00:01.000Z"), + ev.testCase("tests/test_a.py::test_ok", "PASSED", "2026-01-01T00:00:02.840Z"), + ev.testCase("tests/test_a.py::test_bad", "PIPELINE_TEST_CASE_STATUS_FAILED", "2026-01-01T00:00:03.000Z", + `"error": {"exceptions": [{"class_name": "AssertionError", "message": "assert 1 == 2"}]}`, + `"truncation": {"truncated_fields": [{"field_name": "error"}]}`), + // A late RUNNING event does not undo the final outcome. + ev.testCase("tests/test_a.py::test_bad", "RUNNING", "2026-01-01T00:00:02.000Z"), + ev.summary(2, 1, 1, 0, 0), + } + + var finished []string + for _, e := range events { + if c := run.add(e); c != nil { + finished = append(finished, c.id) + } + } + // Reading every page again adds nothing. + for _, e := range events { + assert.Nil(t, run.add(e)) + } + + assert.Equal(t, []string{"tests/test_a.py::test_ok", "tests/test_a.py::test_bad"}, finished) + require.Len(t, run.cases, 2) + + ok := run.cases[0] + assert.Equal(t, testCasePassed, ok.status) + d, known := ok.duration() + assert.True(t, known) + assert.Equal(t, 1840*time.Millisecond, d) + + bad := run.cases[1] + assert.Equal(t, testCaseFailed, bad.status) + assert.True(t, bad.detailsTruncated) + assert.Equal(t, "AssertionError: assert 1 == 2", errorDetailText(bad.errorDetail)) + d, known = bad.duration() + assert.True(t, known) + assert.Equal(t, time.Second, d) + + assert.Empty(t, run.incompleteReason(pipelines.UpdateInfoStateFailed)) +} + +func TestTestRunIncompleteReason(t *testing.T) { + ev := newTestEvents(t) + passed := func() testPipelineEvent { + return ev.testCase("t::a", "PASSED", "") + } + tests := []struct { + name string + events []testPipelineEvent + state pipelines.UpdateInfoState + want string + }{ + { + name: "no summary after completion", + events: []testPipelineEvent{passed()}, + state: pipelines.UpdateInfoStateCompleted, + want: "the update ended without a test summary", + }, + { + name: "two summaries", + events: []testPipelineEvent{passed(), ev.summary(1, 1, 0, 0, 0), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "received 2 test summaries instead of one", + }, + { + name: "missing outcome", + events: []testPipelineEvent{passed(), ev.testCase("t::b", "RUNNING", ""), ev.summary(2, 2, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "received 1 of 2 test results", + }, + { + name: "case without outcome that the summary leaves out", + events: []testPipelineEvent{passed(), ev.testCase("t::b", "RUNNING", ""), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "test t::b has no result", + }, + { + name: "two outcomes for one case", + events: []testPipelineEvent{passed(), ev.testCase("t::a", "FAILED", ""), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "test t::a has more than one result", + }, + { + name: "outcomes differ from summary", + events: []testPipelineEvent{ev.testCase("t::a", "FAILED", ""), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "the test results do not match the test summary", + }, + { + name: "outcome without details", + events: []testPipelineEvent{ + passed(), + decodeTestEvent(t, `{"id": "truncated", "event_type": "test_case_progress", "origin": {"test_case_id": "t::b"}}`), + ev.summary(1, 1, 0, 0, 0), + }, + state: pipelines.UpdateInfoStateCompleted, + want: "some test results could not be read", + }, + { + name: "unknown status", + events: []testPipelineEvent{passed(), ev.testCase("t::b", "PIPELINE_TEST_CASE_STATUS_UNSPECIFIED", ""), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "some test results could not be read", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + run := newTestRun() + for _, e := range tc.events { + run.add(e) + } + assert.Equal(t, tc.want, run.incompleteReason(tc.state)) + }) + } +} + +func TestReadTestEventsFollowsPages(t *testing.T) { + server := testserver.New(t) + var queries []url.Values + server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req testserver.Request) any { + queries = append(queries, req.URL.Query()) + if req.URL.Query().Get("page_token") == "" { + return map[string]any{"events": []map[string]string{{"id": "e1"}}, "next_page_token": "page2"} + } + return map[string]any{"events": []map[string]string{{"id": "e2"}}} + }) + + w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "token"}) + require.NoError(t, err) + events, err := readTestEvents(t.Context(), w, "p1", "u1", "2026-01-01T00:00:01.000Z") + require.NoError(t, err) + + var ids []string + for _, e := range events { + ids = append(ids, e.Id) + } + assert.Equal(t, []string{"e1", "e2"}, ids) + assert.Equal(t, []url.Values{ + { + "filter": {"update_id = 'u1' AND timestamp >= '2026-01-01T00:00:01.000Z'"}, + "order_by": {"timestamp asc"}, + "max_results": {"250"}, + }, + { + "page_token": {"page2"}, + "max_results": {"250"}, + }, + }, queries) +} + +// NFR3: a run of 200 cases spans several pages, and each case is counted once. +func TestReadTestEvents200Cases(t *testing.T) { + const cases = 200 + var all []map[string]any + for i := range cases { + id := fmt.Sprintf("/Workspace/p/tests/test_many.py::test_case_%03d", i) + for j, status := range []string{"PENDING", "RUNNING", "PASSED"} { + all = append(all, map[string]any{ + "id": fmt.Sprintf("c%d-%d", i, j), "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:00.000Z", + "origin": map[string]any{"test_case_id": id}, "details": map[string]any{"test_case_progress": map[string]any{"status": status}}, + }) + } + } + all = append(all, map[string]any{ + "id": "summary", "event_type": "test_run_summary", + "details": map[string]any{"test_run_summary": map[string]any{"total": cases, "passed": cases}}, + }) + + server := testserver.New(t) + pages := 0 + server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req testserver.Request) any { + pages++ + offset := 0 + if token := req.URL.Query().Get("page_token"); token != "" { + _, err := fmt.Sscan(token, &offset) + require.NoError(t, err) + } + end := min(offset+testEventsPageSize, len(all)) + page := map[string]any{"events": all[offset:end]} + if end < len(all) { + page["next_page_token"] = strconv.Itoa(end) + } + return page + }) + + w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "token"}) + require.NoError(t, err) + events, err := readTestEvents(t.Context(), w, "p1", "u1", "") + require.NoError(t, err) + assert.Equal(t, 3, pages) + + run := newTestRun() + for _, e := range append(events, events...) { + run.add(e) + } + assert.Empty(t, run.incompleteReason(pipelines.UpdateInfoStateCompleted)) + assert.Equal(t, testRunSummary{Total: cases, Passed: cases}, run.counts()) +} diff --git a/cmd/pipelines/test_junit.go b/cmd/pipelines/test_junit.go new file mode 100644 index 00000000000..d452281f355 --- /dev/null +++ b/cmd/pipelines/test_junit.go @@ -0,0 +1,193 @@ +package pipelines + +import ( + "context" + "encoding/xml" + "errors" + "fmt" + "io" + "path/filepath" + "strings" + + "github.com/databricks/cli/cmd/root" + "github.com/databricks/cli/libs/atomicfile" + "github.com/databricks/cli/libs/logdiag" +) + +type junitTestSuites struct { + XMLName xml.Name `xml:"testsuites"` + Suites []junitTestSuite `xml:"testsuite"` +} + +type junitTestSuite struct { + Name string `xml:"name,attr"` + Tests int `xml:"tests,attr"` + Failures int `xml:"failures,attr"` + Errors int `xml:"errors,attr"` + Skipped int `xml:"skipped,attr"` + Cases []junitTestCase `xml:"testcase"` +} + +type junitTestCase struct { + ClassName string `xml:"classname,attr,omitempty"` + Name string `xml:"name,attr"` + Time string `xml:"time,attr,omitempty"` + Failure *junitResult `xml:"failure"` + Error *junitResult `xml:"error"` + Skipped *junitResult `xml:"skipped"` +} + +type junitResult struct { + Message string `xml:"message,attr,omitempty"` + Text string `xml:",chardata"` +} + +// junitCaseName splits a pytest test ID the way pytest's JUnit report does: the classname +// is the module path plus any class names, joined by dots. See +// https://docs.pytest.org/en/stable/how-to/output.html#creating-junitxml-format-files. +// A parameter ID in brackets, such as test_cast[a::int], stays with the test name. Brackets are +// looked for only after the file path, which can contain them too. +func junitCaseName(testID string) (string, string) { + file, node, hasNode := strings.Cut(testID, "::") + module := strings.ReplaceAll(strings.TrimSuffix(strings.TrimPrefix(file, "/"), ".py"), "/", ".") + if !hasNode { + return "", module + } + params := "" + if i := strings.Index(node, "["); i >= 0 { + node, params = node[:i], node[i:] + } + parts := strings.Split(node, "::") + parts[len(parts)-1] += params + classes := append([]string{module}, parts[:len(parts)-1]...) + return strings.Join(classes, "."), parts[len(parts)-1] +} + +func newJUnitResult(text string) *junitResult { + message, _, _ := strings.Cut(text, "\n") + return &junitResult{Message: message, Text: text} +} + +// junitNoResult is the error message of a collected case that has no final outcome. +const junitNoResult = "no result received" + +// newJUnitReport has one entry per collected case; a case without a final outcome is an error, +// because CI tools treat skipped entries as passing. +// A non-empty problem, such as an incomplete run, adds a failing entry for the update so the +// report never passes by omission. +// Cases are named relative to projectFolder, as pytest names them, so that the names do not depend on +// the working directory. +func newJUnitReport(suiteName, updateID, projectFolder string, run *testRun, problem string) junitTestSuites { + suite := junitTestSuite{Name: suiteName} + for _, c := range run.cases { + entry := junitTestCase{} + entry.ClassName, entry.Name = junitCaseName(relativeTestID(projectFolder, c.id)) + if d, ok := c.duration(); ok { + entry.Time = fmt.Sprintf("%.3f", d.Seconds()) + } + + details := "" + if c.errorDetail != nil { + details = errorDetailText(c.errorDetail) + } + if c.detailsTruncated { + details = strings.TrimPrefix(details+"\n(details truncated)", "\n") + } + switch c.status { + case testCasePassed: + case testCaseFailed: + entry.Failure = newJUnitResult(details) + suite.Failures++ + case testCaseError: + entry.Error = newJUnitResult(details) + suite.Errors++ + case testCaseSkipped: + entry.Skipped = &junitResult{} + suite.Skipped++ + default: + entry.Error = &junitResult{Message: junitNoResult} + suite.Errors++ + } + suite.Cases = append(suite.Cases, entry) + } + + if problem != "" { + suite.Cases = append(suite.Cases, junitTestCase{ + Name: "update " + updateID, + Error: newJUnitResult(problem), + }) + suite.Errors++ + } + suite.Tests = len(suite.Cases) + return junitTestSuites{Suites: []junitTestSuite{suite}} +} + +// testReport writes the JUnit report of a test run to path, if --junit-xml is set. +type testReport struct { + path string + out io.Writer + suiteName string + projectFolder string +} + +// finish writes the report of a run that ended with runErr. A write error is joined with runErr, +// so that neither is lost. +func (r *testReport) finish(updateID string, run *testRun, problem string, runErr error) error { + if r.path == "" { + return runErr + } + if err := writeJUnitReport(r.path, newJUnitReport(r.suiteName, updateID, r.projectFolder, run, problem)); err != nil { + if runErr == nil || errors.Is(runErr, root.ErrAlreadyPrinted) { + return err + } + return errors.Join(runErr, err) + } + fmt.Fprintf(r.out, "JUnit report: %s\n", filepath.ToSlash(r.path)) + return runErr +} + +// failed writes a report with one failing entry for an error that ended the command before the +// update could be read, so that an older report does not survive. A write error is joined with +// err, as in finish. +func (r *testReport) failed(ctx context.Context, err error) error { + if r.path == "" { + return err + } + message := err.Error() + printed := errors.Is(err, root.ErrAlreadyPrinted) + if printed { + // The error was printed as a diagnostic. Name it instead of the sentinel's text. + message = "the project has errors, see the command output" + if logdiag.IsSetup(ctx) && logdiag.GetFirstErrorSummary(ctx) != "" { + message = logdiag.GetFirstErrorSummary(ctx) + } + } + report := junitTestSuites{Suites: []junitTestSuite{{ + Name: r.suiteName, + Tests: 1, + Errors: 1, + Cases: []junitTestCase{{Name: "pipelines test", Error: newJUnitResult(message)}}, + }}} + if writeErr := writeJUnitReport(r.path, report); writeErr != nil { + if printed { + return writeErr + } + return errors.Join(err, writeErr) + } + return err +} + +// writeJUnitReport replaces the file at path with the report. +func writeJUnitReport(path string, report junitTestSuites) error { + data, err := xml.MarshalIndent(report, "", " ") + if err != nil { + return err + } + data = append([]byte(xml.Header), data...) + data = append(data, '\n') + // A partly written file would be read as a broken report, so the file is replaced at once. + if err := atomicfile.Write(path, data, 0o644); err != nil { + return fmt.Errorf("failed to write JUnit report: %w", err) + } + return nil +} diff --git a/cmd/pipelines/test_junit_test.go b/cmd/pipelines/test_junit_test.go new file mode 100644 index 00000000000..11e4ad51ca8 --- /dev/null +++ b/cmd/pipelines/test_junit_test.go @@ -0,0 +1,76 @@ +package pipelines + +import ( + "errors" + "io" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestJUnitCaseName(t *testing.T) { + tests := []struct { + testID, className, name string + }{ + {"tests/test_orders.py::test_backfill", "tests.test_orders", "test_backfill"}, + {"tests/test_orders.py::TestOrders::test_tax[a-b]", "tests.test_orders.TestOrders", "test_tax[a-b]"}, + // A parameter ID may contain "::". + {"tests/test_a.py::test_cast[a::int]", "tests.test_a", "test_cast[a::int]"}, + {"tests/test_a.py::TestCast::test_cast[a::int]", "tests.test_a.TestCast", "test_cast[a::int]"}, + {"tests/test_orders.py", "", "tests.test_orders"}, + {"/Workspace/Users/someone@example.com/tests/test_x.py::test_x", "Workspace.Users.someone@example.com.tests.test_x", "test_x"}, + // A bracket in a folder name is part of the path, not a parameter ID. + {"tests/[weird]/test_a.py::test_x", "tests.[weird].test_a", "test_x"}, + } + for _, tc := range tests { + className, name := junitCaseName(tc.testID) + assert.Equal(t, tc.className, className, tc.testID) + assert.Equal(t, tc.name, name, tc.testID) + } +} + +func TestWriteJUnitReport(t *testing.T) { + ev := newTestEvents(t) + // Acceptance tests cover each outcome. This covers what the test server does not send: + // multi-line and truncated details, and text that XML must escape. + const project = "/Workspace/Users/someone@example.com/.bundle/orders/dev/files" + run := newTestRun() + for _, e := range []testPipelineEvent{ + ev.testCase(project+"/tests/test_a.py::test_bad", "FAILED", "", + `"error": {"exceptions": [{"class_name": "AssertionError", "message": "assert 1 == 2\nwhere 1 = f()"}]}`, + `"truncation": {"truncated_fields": [{"field_name": "error"}]}`), + ev.testCase(project+"/tests/test_a.py::test_setup", "ERROR", "", + `"error": {"exceptions": [{"class_name": "RuntimeError", "message": "fixture failed"}]}`), + } { + run.add(e) + } + + path := filepath.Join(t.TempDir(), "report.xml") + require.NoError(t, writeJUnitReport(path, newJUnitReport("orders", "u1", project, run, ""))) + + data, err := os.ReadFile(path) + require.NoError(t, err) + assert.Equal(t, ` + + + + AssertionError: assert 1 == 2 where 1 = f() (details truncated) + + + RuntimeError: fixture <db> failed + + + +`, string(data)) +} + +func TestTestReportJoinsWriteErrors(t *testing.T) { + report := &testReport{path: filepath.Join(t.TempDir(), "missing", "report.xml"), out: io.Discard, suiteName: "orders"} + runErr := errors.New("update u1 is FAILED") + err := report.finish("u1", newTestRun(), runErr.Error(), runErr) + require.ErrorIs(t, err, runErr) + assert.ErrorContains(t, err, "failed to write JUnit report") +} diff --git a/cmd/pipelines/test_render.go b/cmd/pipelines/test_render.go new file mode 100644 index 00000000000..7d5b5c8df67 --- /dev/null +++ b/cmd/pipelines/test_render.go @@ -0,0 +1,118 @@ +package pipelines + +import ( + "fmt" + "io" + "strings" + "unicode/utf8" + + "github.com/databricks/databricks-sdk-go/service/pipelines" +) + +// testLineWidth is the width to which a case line is padded with dots before its status. The +// fixed width lines up the statuses while each case prints as it finishes. +const testLineWidth = 80 + +// testProgressPrinter prints the phase of a test update and each case as it finishes. +type testProgressPrinter struct { + out io.Writer + shortID func(id string) string + finished int + waiting bool + running bool +} + +func (p *testProgressPrinter) updateState(state pipelines.UpdateInfoState) { + switch state { + case pipelines.UpdateInfoStateQueued, + pipelines.UpdateInfoStateCreated, + pipelines.UpdateInfoStateWaitingForResources, + pipelines.UpdateInfoStateInitializing, + pipelines.UpdateInfoStateResetting, + pipelines.UpdateInfoStateSettingUpTables: + if !p.waiting && !p.running { + fmt.Fprintln(p.out, "Waiting for compute...") + p.waiting = true + } + case pipelines.UpdateInfoStateRunning: + p.startRunning() + default: + } +} + +func (p *testProgressPrinter) startRunning() { + if !p.running { + fmt.Fprintln(p.out, "Running tests...") + p.running = true + } +} + +// caseFinished prints a line such as +// "2 of 3 FAILED tests/test_orders.py::test_backfill ...... [FAILED in 2.14s]". total is the +// number of known cases; the service reports every case as PENDING before the first one runs. +func (p *testProgressPrinter) caseFinished(c *testCaseResult, total int) { + p.startRunning() + p.finished++ + prefix := fmt.Sprintf("%d of %d %s %s ", p.finished, total, c.status, p.shortID(c.id)) + status := string(c.status) + if d, ok := c.duration(); ok { + status += fmt.Sprintf(" in %.2fs", d.Seconds()) + } + dots := strings.Repeat(".", max(testLineWidth-utf8.RuneCountInString(prefix), 0)) + fmt.Fprintf(p.out, "%s%s [%s]\n", prefix, dots, status) +} + +// renderTestResult prints the end of a test run, before its summary line: the details of every +// failed or errored case, then why the result is incomplete or that no tests were collected. +func renderTestResult(out io.Writer, run *testRun, incompleteReason string, shortID func(id string) string) { + fmt.Fprintln(out) + counts := run.counts() + if counts.Failed > 0 || counts.Errored > 0 { + fmt.Fprintf(out, "Completed with %d failed and %d errored tests:\n", counts.Failed, counts.Errored) + for _, c := range run.cases { + var label string + switch c.status { + case testCaseFailed: + label = "Failure in" + case testCaseError: + label = "Error in" + default: + continue + } + fmt.Fprintf(out, "\n%s %s\n", label, shortID(c.id)) + details := "" + if c.errorDetail != nil { + details = errorDetailText(c.errorDetail) + } + if details == "" { + details = "No details were reported." + } + for line := range strings.SplitSeq(details, "\n") { + fmt.Fprintf(out, " %s\n", line) + } + if c.detailsTruncated { + fmt.Fprintln(out, " (details truncated)") + } + } + fmt.Fprintln(out) + } + + switch { + case incompleteReason != "": + fmt.Fprintf(out, "RUN INCOMPLETE: %s.\n", incompleteReason) + case run.summaries[0].Total == 0: + fmt.Fprintln(out, "No tests were collected.") + } +} + +// renderTestSummaryLine prints the last line of a test run, "Done. PASSED=… TOTAL=…". +// A complete run shows the service's summary. An incomplete run shows the results received, and +// its total counts every case the events named. +func renderTestSummaryLine(out io.Writer, run *testRun, incompleteReason string) { + s := run.counts() + s.Total = len(run.cases) + if incompleteReason == "" { + s = run.summaries[0] + } + fmt.Fprintf(out, "Done. PASSED=%d FAILED=%d SKIPPED=%d ERROR=%d TOTAL=%d\n", s.Passed, s.Failed, s.Skipped, s.Errored, s.Total) +} diff --git a/cmd/pipelines/test_render_test.go b/cmd/pipelines/test_render_test.go new file mode 100644 index 00000000000..874a841a719 --- /dev/null +++ b/cmd/pipelines/test_render_test.go @@ -0,0 +1,80 @@ +package pipelines + +import ( + "bytes" + "strings" + "testing" + + "github.com/databricks/databricks-sdk-go/service/pipelines" + "github.com/stretchr/testify/assert" +) + +func shortTestID(id string) string { + return strings.TrimPrefix(id, "/Workspace/p/") +} + +func TestTestProgressPrinter(t *testing.T) { + var out bytes.Buffer + p := &testProgressPrinter{out: &out, shortID: shortTestID} + p.updateState(pipelines.UpdateInfoStateWaitingForResources) + p.updateState(pipelines.UpdateInfoStateInitializing) + p.updateState(pipelines.UpdateInfoStateRunning) + + ev := newTestEvents(t) + run := newTestRun() + run.add(ev.testCase("/Workspace/p/tests/test_orders.py::test_backfill", "RUNNING", "2026-01-01T00:00:00.000Z")) + p.caseFinished(run.add(ev.testCase("/Workspace/p/tests/test_orders.py::test_backfill", "FAILED", "2026-01-01T00:00:02.140Z")), 2) + long := "/Workspace/p/tests/" + strings.Repeat("x", 70) + ".py::test_y" + p.caseFinished(run.add(ev.testCase(long, "SKIPPED", "")), 2) + + assert.Equal(t, `Waiting for compute... +Running tests... +1 of 2 FAILED tests/test_orders.py::test_backfill .............................. [FAILED in 2.14s] +2 of 2 SKIPPED `+shortTestID(long)+` [SKIPPED] +`, out.String()) +} + +func TestRenderTestResult(t *testing.T) { + ev := newTestEvents(t) + render := func(run *testRun, incompleteReason string) string { + var out bytes.Buffer + renderTestResult(&out, run, incompleteReason, shortTestID) + renderTestSummaryLine(&out, run, incompleteReason) + return out.String() + } + + failed := newTestRun() + failed.add(ev.testCase("/Workspace/p/tests/test_a.py::test_ok", "PASSED", "")) + failed.add(ev.testCase("/Workspace/p/tests/test_a.py::test_bad", "FAILED", "", + `"error": {"exceptions": [{"class_name": "AssertionError", "message": "assert 1 == 2"}]}`, + `"truncation": {"truncated_fields": [{"field_name": "error"}]}`)) + failed.add(ev.testCase("/Workspace/p/tests/test_a.py::test_fixture", "ERROR", "")) + failed.add(ev.summary(3, 1, 1, 0, 1)) + assert.Equal(t, ` +Completed with 1 failed and 1 errored tests: + +Failure in tests/test_a.py::test_bad + AssertionError: assert 1 == 2 + (details truncated) + +Error in tests/test_a.py::test_fixture + No details were reported. + +Done. PASSED=1 FAILED=1 SKIPPED=0 ERROR=1 TOTAL=3 +`, render(failed, "")) + + incomplete := newTestRun() + incomplete.add(ev.testCase("/Workspace/p/tests/test_b.py::test_one", "PASSED", "")) + incomplete.add(ev.testCase("/Workspace/p/tests/test_b.py::test_two", "PENDING", "")) + assert.Equal(t, ` +RUN INCOMPLETE: the update failed before all test results were received. +Done. PASSED=1 FAILED=0 SKIPPED=0 ERROR=0 TOTAL=2 +`, render(incomplete, "the update failed before all test results were received")) + + empty := newTestRun() + empty.add(ev.summary(0, 0, 0, 0, 0)) + assert.Equal(t, ` +No tests were collected. +Done. PASSED=0 FAILED=0 SKIPPED=0 ERROR=0 TOTAL=0 +`, render(empty, "")) +} 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 new file mode 100644 index 00000000000..fe4cf86a137 --- /dev/null +++ b/cmd/pipelines/test_test.go @@ -0,0 +1,414 @@ +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" + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/cli/cmd/root" + "github.com/databricks/cli/libs/testserver" + databricks "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/apierr" + "github.com/databricks/databricks-sdk-go/service/pipelines" + "github.com/spf13/cobra" + + "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{""}, wantErr: "--pytest-arg requires a non-empty value"}, + {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/../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`) +} + +func TestNewStartTestUpdateRequest(t *testing.T) { + assert.Equal(t, startTestUpdateRequest{TestOnly: true}, newStartTestUpdateRequest(nil, nil)) + assert.Equal(t, startTestUpdateRequest{ + TestOnly: true, + TestOptions: &testOptions{TestSelection: []string{"/Workspace/tests"}, PytestArgs: []string{"-k", "orders"}}, + }, newStartTestUpdateRequest([]string{"/Workspace/tests"}, []string{"-k", "orders"})) +} + +func TestRootPathSelection(t *testing.T) { + got, err := rootPathSelection("/Workspace/Users/someone/orders/", "../other/tests") + require.NoError(t, err) + // The path is not cleaned, so the service rejects ".." with its own error. + assert.Equal(t, "/Workspace/Users/someone/orders/../other/tests", got) + + _, err = rootPathSelection("", "tests") + assert.ErrorContains(t, err, "the pipeline has no root_path") +} + +// 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)) +} + +func TestNotPipelineKeyError(t *testing.T) { + cmd := &cobra.Command{Use: "test"} + b := &bundle.Bundle{} + b.Config.Resources.Pipelines = map[string]*resources.Pipeline{"orders": {}, "customers": {}} + err := notPipelineKeyError(cmd, b, "tests/test_orders.py", []string{"tests/test_orders.py", "tests/test_x.py::test_a b"}) + assert.EqualError(t, err, `"tests/test_orders.py" is not a pipeline in this bundle (pipelines: customers, orders). Usage: test customers tests/test_orders.py 'tests/test_x.py::test_a b'`) + + err = notPipelineKeyError(cmd, &bundle.Bundle{}, "tests", []string{"tests"}) + assert.EqualError(t, err, `"tests" is not a pipeline in this bundle, and the bundle has no pipelines`) +} + +func TestWithDeployHint(t *testing.T) { + parent := &cobra.Command{Use: "pipelines"} + cmd := &cobra.Command{Use: "test"} + parent.AddCommand(cmd) + noSources := &apierr.APIError{ErrorCode: "INVALID_PARAMETER_VALUE", Message: "[PIPELINE_TEST_NO_TEST_SOURCES] The pipeline has no test sources."} + withTesting := testTarget{key: "orders", bundleHasTesting: true} + + err := withDeployHint(cmd, withTesting, noSources) + assert.ErrorIs(t, err, noSources) + assert.EqualError(t, err, "[PIPELINE_TEST_NO_TEST_SOURCES] The pipeline has no test sources.\nThe bundle sets testing for orders, but the deployed pipeline has none. Run 'pipelines deploy' first") + + // No hint without testing in the bundle, or for another error. + assert.Same(t, error(noSources), withDeployHint(cmd, testTarget{key: "orders"}, noSources)) + conflict := &apierr.APIError{ErrorCode: "RESOURCE_CONFLICT", Message: "An active update already exists."} + assert.Same(t, error(conflict), withDeployHint(cmd, withTesting, conflict)) +} + +func TestTestRunError(t *testing.T) { + ev := newTestEvents(t) + tests := []struct { + name string + events func() []testPipelineEvent + state pipelines.UpdateInfoState + wantErr string + }{ + { + name: "incomplete after completion", + events: func() []testPipelineEvent { + return []testPipelineEvent{ev.testCase("t::a", "PASSED", "")} + }, + state: pipelines.UpdateInfoStateCompleted, + wantErr: root.ErrAlreadyPrinted.Error(), + }, + { + name: "update failed with the service's verdict although no test failed", + events: func() []testPipelineEvent { + return []testPipelineEvent{ + ev.testCase("t::a", "PASSED", ""), + ev.summary(1, 1, 0, 0, 0), + decodeTestEvent(t, `{"id": "fatal", "error": {"fatal": true, "exceptions": [{"message": "[PIPELINE_TEST_FAILED] 1 of the pipeline's tests failed."}]}}`), + } + }, + state: pipelines.UpdateInfoStateFailed, + wantErr: "update u1 is FAILED: [PIPELINE_TEST_FAILED] 1 of the pipeline's tests failed.", + }, + { + name: "tests failed although the update completed", + events: func() []testPipelineEvent { + return []testPipelineEvent{ev.testCase("t::a", "FAILED", ""), ev.summary(1, 0, 1, 0, 0)} + }, + state: pipelines.UpdateInfoStateCompleted, + wantErr: root.ErrAlreadyPrinted.Error(), + }, + { + name: "no tests collected although the update completed", + events: func() []testPipelineEvent { + return []testPipelineEvent{ev.summary(0, 0, 0, 0, 0)} + }, + state: pipelines.UpdateInfoStateCompleted, + wantErr: root.ErrAlreadyPrinted.Error(), + }, + { + name: "all tests passed but the completed update reported a fatal error", + events: func() []testPipelineEvent { + return []testPipelineEvent{ + ev.testCase("t::a", "PASSED", ""), + ev.summary(1, 1, 0, 0, 0), + decodeTestEvent(t, `{"id": "fatal", "error": {"fatal": true, "exceptions": [{"message": "result delivery failed"}]}}`), + } + }, + state: pipelines.UpdateInfoStateCompleted, + wantErr: "update u1 is COMPLETED: result delivery failed", + }, + { + name: "all tests were skipped", + events: func() []testPipelineEvent { + return []testPipelineEvent{ev.testCase("t::a", "SKIPPED", ""), ev.summary(1, 0, 0, 1, 0)} + }, + state: pipelines.UpdateInfoStateCompleted, + }, + { + name: "canceled after tests failed", + events: func() []testPipelineEvent { + return []testPipelineEvent{ev.testCase("t::a", "FAILED", ""), ev.summary(1, 0, 1, 0, 0)} + }, + state: pipelines.UpdateInfoStateCanceled, + wantErr: "update u1 is CANCELED", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + run := newTestRun() + for _, e := range tc.events() { + run.add(e) + } + err := testRunError("u1", tc.state, run, run.incompleteReason(tc.state)) + if tc.wantErr == "" { + assert.NoError(t, err) + } else { + assert.EqualError(t, err, tc.wantErr) + } + }) + } +} + +func TestRelativeTestID(t *testing.T) { + id := "/Workspace/Users/someone/orders/tests/test_a.py::test_x" + assert.Equal(t, "tests/test_a.py::test_x", relativeTestID("/Workspace/Users/someone/orders", id)) + // root_path may lack the /Workspace prefix that the service adds to test IDs. + assert.Equal(t, "tests/test_a.py::test_x", relativeTestID("/Users/someone/orders/", id)) + assert.Equal(t, id, relativeTestID("/Workspace/Users/someone/other", id)) + assert.Equal(t, id, relativeTestID("", id)) +} + +// The wait reads events from the last timestamp, where a later event can share that timestamp, +// and reads all events again after the update ends, where an event can appear late. +func TestWaitForTestUpdate(t *testing.T) { + event := func(id, caseID, status, timestamp string) map[string]any { + return map[string]any{ + "id": id, "event_type": "test_case_progress", "timestamp": timestamp, + "origin": map[string]any{"test_case_id": caseID}, "details": map[string]any{"test_case_progress": map[string]any{"status": status}}, + } + } + a, b := "/Workspace/p/tests/test_a.py::test_a", "/Workspace/p/tests/test_a.py::test_b" + first := []map[string]any{ + event("e1", a, "PENDING", "2026-01-01T00:00:01.000Z"), + event("e2", b, "PENDING", "2026-01-01T00:00:01.000Z"), + event("e3", a, "PASSED", "2026-01-01T00:00:02.000Z"), + } + // e4 has the timestamp of the cursor; e5 and the summary are readable only after the update ends. + tie := append(first[2:], event("e4", b, "RUNNING", "2026-01-01T00:00:02.000Z")) + late := []map[string]any{event("e5", b, "PASSED", "2026-01-01T00:00:03.000Z"), { + "id": "e6", "event_type": "test_run_summary", "timestamp": "2026-01-01T00:00:04.000Z", + "details": map[string]any{"test_run_summary": map[string]any{"total": 2, "passed": 2}}, + }} + + server := testserver.New(t) + polls := 0 + server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/updates/{update_id}", func(req testserver.Request) any { + polls++ + state := "RUNNING" + if polls == 2 { + state = "COMPLETED" + } + return map[string]any{"update": map[string]any{"update_id": "u1", "state": state}} + }) + var filters []string + server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req testserver.Request) any { + filter := req.URL.Query().Get("filter") + filters = append(filters, filter) + switch { + case !strings.Contains(filter, "timestamp") && polls == 1: + return map[string]any{"events": first} + case strings.Contains(filter, "timestamp"): + return map[string]any{"events": tie} + default: + return map[string]any{"events": append(append(first, tie...), late...)} + } + }) + + w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "token"}) + require.NoError(t, err) + 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, nil) + require.NoError(t, err) + + assert.Equal(t, pipelines.UpdateInfoStateCompleted, state) + assert.Equal(t, []string{ + "update_id = 'u1'", + "update_id = 'u1' AND timestamp >= '2026-01-01T00:00:02.000Z'", + "update_id = 'u1'", + }, filters) + assert.Empty(t, run.incompleteReason(state)) + 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)) +} + +func TestTestRunProblem(t *testing.T) { + ev := newTestEvents(t) + passed := newTestRun() + passed.add(ev.testCase("t::a", "PASSED", "")) + passed.add(ev.summary(1, 1, 0, 0, 0)) + assert.Empty(t, testRunProblem(passed, "", nil)) + assert.Empty(t, testRunProblem(passed, "", root.ErrAlreadyPrinted)) + assert.Equal(t, "update u1 is COMPLETED: result delivery failed", testRunProblem(passed, "", errors.New("update u1 is COMPLETED: result delivery failed"))) + + empty := newTestRun() + empty.add(ev.summary(0, 0, 0, 0, 0)) + assert.Equal(t, "No tests were collected.", testRunProblem(empty, "", root.ErrAlreadyPrinted)) + + assert.Equal(t, "RUN INCOMPLETE: the update was canceled.\nupdate u1 is CANCELED", + testRunProblem(newTestRun(), "the update was canceled", errors.New("update u1 is CANCELED"))) +}