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/output.txt b/acceptance/pipelines/test/arguments/output.txt index 0e7d4dfb200..cf34209ecd6 100644 --- a/acceptance/pipelines/test/arguments/output.txt +++ b/acceptance/pipelines/test/arguments/output.txt @@ -23,12 +23,30 @@ 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 @@ -45,6 +63,11 @@ Global Flags: >>> 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 @@ -70,3 +93,33 @@ Error: --pipeline-id "1234" is not a pipeline ID >>> 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 index 0df8c20ce2d..ceb54ec432e 100644 --- a/acceptance/pipelines/test/arguments/script +++ b/acceptance/pipelines/test/arguments/script @@ -3,6 +3,10 @@ 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 @@ -16,3 +20,18 @@ 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 index 483234d0dd4..7f9a598cea2 100644 --- a/acceptance/pipelines/test/immutable/databricks.yml +++ b/acceptance/pipelines/test/immutable/databricks.yml @@ -11,3 +11,7 @@ resources: libraries: - file: path: pipeline_file.py + testing: + sources: + - glob: + include: ./tests/** diff --git a/acceptance/pipelines/test/immutable/output.txt b/acceptance/pipelines/test/immutable/output.txt index e18544efea3..02b529b6da8 100644 --- a/acceptance/pipelines/test/immutable/output.txt +++ b/acceptance/pipelines/test/immutable/output.txt @@ -17,7 +17,7 @@ Error: cannot resolve test selection "tests/test_orders.py::test_backfill": loca === 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 +>>> 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. @@ -25,7 +25,7 @@ If you've made local changes, run 'databricks pipelines deploy' first to ensure Pipeline: my_pipeline ([UUID][0]) Update: [UUID][1] ->>> [CLI] pipelines test my_pipeline +>>> [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. diff --git a/acceptance/pipelines/test/immutable/script b/acceptance/pipelines/test/immutable/script index 27025924735..e78d8a60a9c 100644 --- a/acceptance/pipelines/test/immutable/script +++ b/acceptance/pipelines/test/immutable/script @@ -5,6 +5,6 @@ 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 -trace $CLI pipelines test my_pipeline +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/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..57aef006f3b --- /dev/null +++ b/acceptance/pipelines/test/invalid-project/output.txt @@ -0,0 +1,6 @@ + +=== 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..d8424d76895 --- /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" +# 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..fa7ee30b275 --- /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" +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/output.txt b/acceptance/pipelines/test/pipeline-id/output.txt index 4664393739b..56c5565804d 100644 --- a/acceptance/pipelines/test/pipeline-id/output.txt +++ b/acceptance/pipelines/test/pipeline-id/output.txt @@ -1,7 +1,7 @@ === 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 +>>> 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] @@ -25,7 +25,7 @@ Update: [UUID][1] >>> 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 +>>> 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] diff --git a/acceptance/pipelines/test/pipeline-id/script b/acceptance/pipelines/test/pipeline-id/script index 3745fe70c50..bf8ae479ce7 100644 --- a/acceptance/pipelines/test/pipeline-id/script +++ b/acceptance/pipelines/test/pipeline-id/script @@ -3,12 +3,12 @@ no_root_pipeline_id=$($CLI pipelines create --json '{"name": "no-root"}' | jq -r rm "$OUT_REQUESTS" title "Relative test selections resolve against the pipeline root_path" -trace MSYS_NO_PATHCONV=1 $CLI pipelines test --pipeline-id $pipeline_id tests/test_orders.py::test_backfill /Workspace/Users/someone@example.com/other/tests +trace 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 +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" 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..8284a1e80aa --- /dev/null +++ b/acceptance/pipelines/test/results/failed/output.txt @@ -0,0 +1,22 @@ + +=== 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..38328d631af --- /dev/null +++ b/acceptance/pipelines/test/results/incomplete/output.txt @@ -0,0 +1,13 @@ + +=== 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..817b57ddabf --- /dev/null +++ b/acceptance/pipelines/test/results/passed/output.txt @@ -0,0 +1,32 @@ + +=== 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..b8671d24524 --- /dev/null +++ b/acceptance/pipelines/test/run-control/canceled/output.txt @@ -0,0 +1,11 @@ + +=== 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..fd236b1c474 --- /dev/null +++ b/acceptance/pipelines/test/run-control/not-test-update/output.txt @@ -0,0 +1,5 @@ + +=== --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..f0d33a156b0 --- /dev/null +++ b/acceptance/pipelines/test/run-control/resume/output.txt @@ -0,0 +1,10 @@ + +=== --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..b7eac150675 --- /dev/null +++ b/acceptance/pipelines/test/run-control/timeout/output.txt @@ -0,0 +1,9 @@ + +=== --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 index 0cf849a7919..ea0662b1043 100644 --- a/acceptance/pipelines/test/selection/databricks.yml +++ b/acceptance/pipelines/test/selection/databricks.yml @@ -8,3 +8,7 @@ resources: libraries: - file: path: pipeline_file.py + testing: + sources: + - glob: + include: ./tests/** diff --git a/acceptance/pipelines/test/selection/output.txt b/acceptance/pipelines/test/selection/output.txt index 47d75a6bb74..46ddf79e15c 100644 --- a/acceptance/pipelines/test/selection/output.txt +++ b/acceptance/pipelines/test/selection/output.txt @@ -8,7 +8,7 @@ View your pipeline my_pipeline here: [DATABRICKS_URL]/pipelines/[UUID][0]?w=[WOR === Run all tests of the only pipeline ->>> [CLI] pipelines test +>>> [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. @@ -27,7 +27,7 @@ Update: [UUID][1] === 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 +>>> 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. @@ -59,7 +59,7 @@ Update: [UUID][2] === Relative paths resolve from the working directory ->>> [CLI] pipelines test my_pipeline test_orders.py::test_valid_order +>>> [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. diff --git a/acceptance/pipelines/test/selection/script b/acceptance/pipelines/test/selection/script index 5e7efd415be..3344ef8a37d 100644 --- a/acceptance/pipelines/test/selection/script +++ b/acceptance/pipelines/test/selection/script @@ -2,16 +2,16 @@ trace $CLI pipelines deploy rm "$OUT_REQUESTS" title "Run all tests of the only pipeline" -trace $CLI pipelines test +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 +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 +trace $CLI pipelines test my_pipeline test_orders.py::test_valid_order --no-wait cd .. trace print_requests.py //updates 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/output.txt b/acceptance/pipelines/test/submit-error/output.txt index 32a27c892d3..975173bc843 100644 --- a/acceptance/pipelines/test/submit-error/output.txt +++ b/acceptance/pipelines/test/submit-error/output.txt @@ -1,8 +1,8 @@ -=== A submit error from the service is shown as is, and no update ID is printed +=== 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] -Error: An active update '[UUID][1]' already exists for pipeline '[UUID][0]'. +>>> 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 { diff --git a/acceptance/pipelines/test/submit-error/script b/acceptance/pipelines/test/submit-error/script index 0d0843028c6..1035f977d33 100644 --- a/acceptance/pipelines/test/submit-error/script +++ b/acceptance/pipelines/test/submit-error/script @@ -1,3 +1,7 @@ -title "A submit error from the service is shown as is, and no update ID is printed" -trace musterr $CLI pipelines test --pipeline-id 11111111-2222-3333-4444-555555555555 +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/test.toml b/acceptance/pipelines/test/test.toml index 66f23523071..4b2b5e84b92 100644 --- a/acceptance/pipelines/test/test.toml +++ b/acceptance/pipelines/test/test.toml @@ -2,3 +2,5 @@ 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/test.go b/cmd/pipelines/test.go index dbbc9e042e8..83fdf0b11ee 100644 --- a/cmd/pipelines/test.go +++ b/cmd/pipelines/test.go @@ -9,10 +9,13 @@ import ( "maps" "net/http" "os" + "os/signal" "path" "path/filepath" "slices" "strings" + "syscall" + "time" "uuid" "github.com/databricks/cli/bundle" @@ -28,6 +31,7 @@ import ( "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" @@ -192,6 +196,20 @@ func rootPathSelection(rootPath, relPath string) (string, error) { 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}) @@ -204,6 +222,31 @@ func getRootPath(ctx context.Context, w *databricks.WorkspaceClient, pipelineID return pipeline.Spec.RootPath, nil } +// updateTestOnly holds UpdateInfo.test_only from a Get Update response; the Go SDK does not have it yet. +// The service leaves it out where it cannot report it, such as before pipeline testing is enabled. +type updateTestOnly struct { + Update struct { + TestOnly *bool `json:"test_only"` + } `json:"update"` +} + +// checkTestUpdate returns an error if the service reports that the update is not a test-only update. +func checkTestUpdate(ctx context.Context, w *databricks.WorkspaceClient, pipelineID, updateID string) error { + apiClient, err := client.New(w.Config) + if err != nil { + return fmt.Errorf("failed to create API client: %w", err) + } + var response updateTestOnly + err = apiClient.Do(ctx, http.MethodGet, fmt.Sprintf("/api/2.0/pipelines/%s/updates/%s", pipelineID, updateID), auth.WorkspaceIDHeaders(w.Config), nil, nil, &response) + if err != nil { + return err + } + if testOnly := response.Update.TestOnly; testOnly != nil && !*testOnly { + return fmt.Errorf("update %s is not a test update; pass the update ID that pipelines test printed", updateID) + } + return nil +} + // startTestUpdate starts a test-only update and returns its ID. func startTestUpdate(ctx context.Context, w *databricks.WorkspaceClient, pipelineID string, req startTestUpdateRequest) (string, error) { apiClient, err := client.New(w.Config) @@ -227,12 +270,40 @@ func startTestUpdate(ctx context.Context, w *databricks.WorkspaceClient, pipelin 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) { @@ -245,7 +316,7 @@ func (t testTarget) printHeader(out io.Writer, updateID string) { } // resolveTestTargetByID identifies the pipeline by --pipeline-id; all positional arguments are selections. -func resolveTestTargetByID(cmd *cobra.Command, args []string, pipelineID string) (testTarget, error) { +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())) } @@ -255,8 +326,9 @@ func resolveTestTargetByID(cmd *cobra.Command, args []string, pipelineID string) ctx := cmd.Context() w := cmdctx.WorkspaceClient(ctx) + // root_path resolves relative selections and shortens the test IDs printed while waiting. var rootPath string - if hasRelativeSelection(args) { + if wait || hasRelativeSelection(args) { var err error rootPath, err = getRootPath(ctx, w, pipelineID) if err != nil { @@ -269,11 +341,11 @@ func resolveTestTargetByID(cmd *cobra.Command, args []string, pipelineID string) if err != nil { return testTarget{}, err } - return testTarget{pipelineID: pipelineID, selections: selections, w: w}, nil + 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) (testTarget, error) { +func resolveTestTargetInBundle(cmd *cobra.Command, args []string, resume bool) (testTarget, error) { var key string b, err := utils.ProcessBundle(cmd, utils.ProcessOptions{ PostInitFunc: func(ctx context.Context, b *bundle.Bundle) error { @@ -296,7 +368,9 @@ func resolveTestTargetInBundle(cmd *cobra.Command, args []string) (testTarget, e } return testTarget{}, err } - suggestPipelineDeploy(ctx, cmd) + if !resume { + suggestPipelineDeploy(ctx, cmd) + } cwd, err := os.Getwd() if err != nil { @@ -313,7 +387,242 @@ func resolveTestTargetInBundle(cmd *cobra.Command, args []string) (testTarget, e if err != nil { return testTarget{}, err } - return testTarget{key: key, pipelineID: pipelineID, selections: selections, w: b.WorkspaceClient(ctx)}, nil + 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 @@ -355,15 +664,32 @@ 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'`, + 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 pipelineID string - var pytestArgs []string + var f testFlags testGroup := cmdgroup.NewFlagGroup("Pipeline Test") - testGroup.FlagSet().StringVar(&pipelineID, "pipeline-id", "", "ID of the pipeline to test, instead of a KEY.") - testGroup.FlagSet().StringArrayVar(&pytestArgs, "pytest-arg", nil, "Argument to pass to pytest (repeatable). Only -k and -m are supported.") + testGroup.FlagSet().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) @@ -371,45 +697,76 @@ example: cmd.SetFlagErrorFunc(testFlagErrorFunc) cmd.RunE = func(cmd *cobra.Command, args []string) error { - if enabled, _ := env.GetBool(cmd.Context(), pipelinesTestEnvVar); !enabled { - return fmt.Errorf("pipelines test is experimental. Set %s=true to enable it", pipelinesTestEnvVar) + // 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 cmd.ArgsLenAtDash() >= 0 { - return errors.New("the -- separator is not supported: " + pytestArgHint) + if err := checkTestFlags(cmd, args, f); err != nil { + return report.failed(cmd.Context(), err) } - if root.OutputType(cmd) != flags.OutputText { - return fmt.Errorf("--output %s is not supported by pipelines test", root.OutputType(cmd)) + resume := cmd.Flags().Changed("resume") + target, err := resolveTestTarget(cmd, args, f, resume) + if err != nil { + return report.failed(cmd.Context(), err) } - if err := validatePytestArgs(pytestArgs); err != nil { - return err + report.suiteName = target.key + if report.suiteName == "" { + report.suiteName = target.pipelineID } - var target testTarget - if pipelineID != "" { - id, err := uuid.Parse(pipelineID) - if err != nil { - return fmt.Errorf("--pipeline-id %q is not a pipeline ID", pipelineID) - } - target, err = resolveTestTargetByID(cmd, args, id.String()) - if err != nil { - return err - } - } else { - cmd.SetContext(logdiag.InitContext(cmd.Context())) - var err error - target, err = resolveTestTargetInBundle(cmd, args) + 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 err + 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 } - ctx := cmd.Context() - req := newStartTestUpdateRequest(target.selections, pytestArgs) - updateID, err := startTestUpdate(ctx, target.w, target.pipelineID, req) + 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 { - return err + 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) } - target.printHeader(cmd.OutOrStdout(), updateID) - return nil + 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 index 682567d5b06..fe4cf86a137 100644 --- a/cmd/pipelines/test_test.go +++ b/cmd/pipelines/test_test.go @@ -1,14 +1,24 @@ 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" @@ -125,3 +135,280 @@ func TestNotPipelineKeyError(t *testing.T) { 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"))) +}