experimental/bundletest: add the cloud backend

experimental/bundletest/README.md

@@ -47,8 +47,9 @@ The same test runs against either backend, chosen by the `BUNDLETEST_BACKEND` en
   DuckDB's dialect, type system, and semantics differ, so a green local run means "the SQL
   is portable and structurally sound," not "this passes on Databricks." Genuinely
   dialect-dependent checks belong on cloud.
-- **`cloud`** — deco-provisioned real workspace. Real fidelity. *(Arrives as a stacked PR
-  on top of this base.)*
+- **`cloud`** — a real workspace. Real fidelity: `databricks bundle deploy` + real job runs
+  and SQL through the Databricks SDK. Slower (deploys take minutes) and costs real compute,
+  so it's the gated tier. The `cloud_only` assertions run here instead of skipping.
 
 ### How the local backend stays honest
 
@@ -75,3 +76,23 @@ uv venv --python 3.12
 uv pip install -e ".[dev]"
 uv run pytest -v
 ```
+
+### Run it on cloud
+
+The cloud backend deploys to a real workspace. Example fixture `examples/cloud_orders/` contains two SQL jobs, a managed volume, and a file_path dashboard under `main.bundletest_cloud`:
+
+```sh
+export BUNDLETEST_BACKEND=cloud
+export BUNDLETEST_PROFILE=<profile>              # from ~/.databrickscfg
+export BUNDLETEST_WAREHOUSE_ID=<sql-warehouse-id>  # used for seeding + assertion queries
+export BUNDLE_VAR_warehouse_id=<sql-warehouse-id>
+uv run --extra dev pytest examples/cloud_orders
+```
+
+(`examples/orders_bundle/` is local static-config only, not deployable to cloud.)
+
+Seeded tables and job runs are real and cost money, so unlike the local backend (a fresh
+in-memory DuckDB per test) the cloud backend persists state within a run. `teardown()` drops
+the tables it seeded and runs `bundle destroy`, but the tests still share a workspace — prefer
+a **module-scoped** `env` fixture (deploy once per module) and an isolated namespace per run
+over the local backend's function-scoped, throwaway one.

experimental/bundletest/examples/cloud_orders/dashboards/orders_overview.lvdash.json

@@ -0,0 +1,19 @@
+{
+  "datasets": [
+    {
+      "name": "summary",
+      "displayName": "Order summary",
+      "queryLines": [
+        "SELECT order_count, total_revenue\n",
+        "FROM main.bundletest_cloud.order_summary"
+      ]
+    }
+  ],
+  "pages": [
+    {
+      "name": "main",
+      "displayName": "Overview",
+      "layout": []
+    }
+  ]
+}

experimental/bundletest/examples/cloud_orders/databricks.yml

@@ -0,0 +1,51 @@
+# A deployable bundle for the cloud backend's own end-to-end validation.
+#
+# Unlike examples/orders_bundle (a static-config gallery the LOCAL backend only reads), this
+# one is actually `databricks bundle deploy`-ed to a real workspace, so it declares only
+# resources that deploy and run: two SQL jobs, a managed volume, and a file_path dashboard
+# (whose serialized form the cloud backend must hydrate from the workspace). Everything lives
+# under main.bundletest_cloud; the test fixture creates that schema and drops it (CASCADE)
+# after the run. The tests here are collected only when BUNDLETEST_BACKEND=cloud.
+bundle:
+  name: bundletest-cloud
+
+variables:
+  warehouse_id:
+    description: SQL warehouse the sql_task jobs and the dashboard run on
+
+resources:
+  volumes:
+    raw_data:
+      name: raw_data
+      catalog_name: main
+      schema_name: bundletest_cloud
+      volume_type: MANAGED
+
+  jobs:
+    # bronze -> silver: dedupe + pin price to a fixed decimal.
+    transform_orders:
+      name: transform_orders
+      tasks:
+        - task_key: transform
+          sql_task:
+            warehouse_id: ${var.warehouse_id}
+            file:
+              path: src/transform_orders.sql
+
+    # silver -> gold: one summary row.
+    aggregate_orders:
+      name: aggregate_orders
+      tasks:
+        - task_key: aggregate
+          sql_task:
+            warehouse_id: ${var.warehouse_id}
+            file:
+              path: src/aggregate_orders.sql
+
+  dashboards:
+    # Defined by file_path (not inline), so get_resource must read the rendered
+    # serialized_dashboard back from the deployed dashboard for source_tables() to work.
+    orders_overview:
+      display_name: Orders Overview (bundletest-cloud)
+      warehouse_id: ${var.warehouse_id}
+      file_path: dashboards/orders_overview.lvdash.json

experimental/bundletest/examples/cloud_orders/src/aggregate_orders.sql

@@ -0,0 +1,4 @@
+-- silver -> gold: one summary row over the cleaned orders.
+CREATE OR REPLACE TABLE main.bundletest_cloud.order_summary AS
+SELECT COUNT(*) AS order_count, SUM(total_price) AS total_revenue
+FROM main.bundletest_cloud.orders;

experimental/bundletest/examples/cloud_orders/src/transform_orders.sql

@@ -0,0 +1,5 @@
+-- bronze -> silver: drop duplicate/null-id rows and pin the price to a fixed decimal.
+CREATE OR REPLACE TABLE main.bundletest_cloud.orders AS
+SELECT DISTINCT order_id, CAST(total_price AS DECIMAL(10, 2)) AS total_price
+FROM main.bundletest_cloud.raw_orders
+WHERE order_id IS NOT NULL;

experimental/bundletest/examples/cloud_orders/tests/conftest.py

@@ -0,0 +1,63 @@
+"""Fixture for the cloud-backend end-to-end validation.
+
+These tests really deploy + run against a workspace, so they are collected only when
+BUNDLETEST_BACKEND=cloud. The env is module-scoped (deploy once — deploys take minutes and
+cost compute). Each run gets a UNIQUE schema (and bundle name) so two concurrent runs, or any
+shared use of the workspace, never collide; the schema is created up front and dropped CASCADE
+afterwards, so cleanup is unambiguous.
+"""
+
+import shutil
+import tempfile
+import uuid
+import warnings
+from pathlib import Path
+
+import pytest
+from bundletest import BundleEnv
+from bundletest.env import current_backend_kind, make_backend
+
+BUNDLE = str(Path(__file__).resolve().parent.parent)
+RUN_ID = uuid.uuid4().hex[:8]
+SCHEMA = f"main.bundletest_cloud_{RUN_ID}"
+
+# Nothing here runs on the local backend — skip collection entirely so CI stays local-only.
+if current_backend_kind() != "cloud":
+    collect_ignore_glob = ["test_*.py"]
+
+
+@pytest.fixture(scope="module")
+def schema() -> str:
+    """The unique target schema for this run (main.bundletest_cloud_<run>)."""
+    return SCHEMA
+
+
+@pytest.fixture(scope="module")
+def env():
+    backend = make_backend("cloud")
+    # Deploy from a per-run copy with the `bundletest_cloud` schema and `bundletest-cloud` bundle
+    # name suffixed by RUN_ID. The .sql/.lvdash.json artifacts hardcode the schema (DABs doesn't
+    # interpolate file contents), so substituting in a copy is how each run gets its own tables,
+    # volume, dashboard, and deploy path — the committed fixture keeps the readable placeholder.
+    tmp = Path(tempfile.mkdtemp(prefix="bundletest-cloud-"))
+    bundle_dir = tmp / "bundle"
+    shutil.copytree(BUNDLE, bundle_dir)
+    for path in bundle_dir.rglob("*"):
+        if path.suffix in (".yml", ".sql", ".json"):
+            text = path.read_text()
+            text = text.replace("bundletest_cloud", f"bundletest_cloud_{RUN_ID}")
+            text = text.replace("bundletest-cloud", f"bundletest-cloud-{RUN_ID}")
+            path.write_text(text)
+
+    backend.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}")
+    e = BundleEnv(str(bundle_dir), backend)
+    e.deploy()
+    try:
+        yield e
+    finally:
+        e.teardown()
+        try:
+            backend.execute_sql(f"DROP SCHEMA IF EXISTS {SCHEMA} CASCADE")
+        except Exception as exc:
+            warnings.warn(f"failed to drop {SCHEMA} (may be leaked): {exc}", stacklevel=1)
+        shutil.rmtree(tmp, ignore_errors=True)

experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py

@@ -0,0 +1,76 @@
+"""End-to-end validation of the cloud backend against a real workspace.
+
+Exercises the seam methods that can only be verified on cloud: deploy, run_job on the
+deployed jobs, execute_sql/table_schema round-trips, get_resource off `bundle summary`
+(which inlines a file_path dashboard's serialized form), and volume upload/read.
+"""
+
+import pytest
+
+
+def test_bronze_to_silver_to_gold(env, schema):
+    env.seed(
+        f"{schema}.raw_orders",
+        [
+            {"order_id": 1, "total_price": 10.0},
+            {"order_id": 1, "total_price": 10.0},  # duplicate
+            {"order_id": 2, "total_price": 5.0},
+            {"order_id": None, "total_price": 1.0},  # null id -> dropped
+        ],
+    )
+
+    assert env.run_job("transform_orders").succeeded  # bronze -> silver
+    silver = env.table(f"{schema}.orders")
+    assert silver.row_count() == 2
+    assert silver.has_no_nulls("order_id")
+    assert silver.column("order_id").is_unique()
+
+    assert env.run_job("aggregate_orders").succeeded  # silver -> gold
+    summary = env.table(f"{schema}.order_summary")
+    assert summary.row_count() == 1
+    assert summary.column("order_count").min() == 2
+    assert summary.column("total_revenue").min() == 15.0
+
+
+@pytest.mark.cloud_only
+def test_price_type_is_databricks_decimal(env, schema):
+    env.seed(f"{schema}.raw_orders", [{"order_id": 1, "total_price": 10.0}])
+    env.run_job("transform_orders")
+    assert env.table(f"{schema}.orders").schema["total_price"] == "decimal(10,2)"
+
+
+def test_job_is_wired_to_its_sql(env):
+    job = env.backend.get_resource("jobs", "transform_orders")
+    assert job["tasks"][0]["sql_task"]["file"]["path"].endswith("transform_orders.sql")
+
+
+def test_dashboard_source_tables_from_file_path(env, schema):
+    # The dashboard is defined by file_path, not inline, yet source_tables() still resolves:
+    # `bundle summary` inlines the file's serialized form at config-load, so get_resource has it.
+    dashboard = env.dashboard("orders_overview")
+    assert dashboard.exists()
+    assert dashboard.source_tables() == [f"{schema}.order_summary"]
+
+
+def test_uploaded_csv_is_readable(env, tmp_path):
+    csv = tmp_path / "orders.csv"
+    csv.write_text("order_id,total_price\n1,10.0\n2,5.0\n")
+
+    env.volume("raw_data").upload(str(csv))
+
+    orders = env.volume("raw_data").file("orders.csv")
+    assert orders.exists()
+    assert orders.row_count() == 2
+    assert "order_id" in orders.columns
+
+
+@pytest.mark.cloud_only
+def test_deployed_job_carries_server_filled_fields(env):
+    # get_deployed reads the workspace's stored object, so it carries values the server filled
+    # in or normalized that our databricks.yml never declared — what get_resource (the declared
+    # config) cannot show. This is the point of validating against real deployment.
+    deployed = env.backend.get_deployed("jobs", "transform_orders")
+    assert deployed["settings"]["name"] == "transform_orders"
+    assert deployed["settings"]["format"] == "MULTI_TASK"  # server-normalized
+    assert deployed["settings"]["max_concurrent_runs"] == 1  # server default
+    assert deployed["run_as_user_name"]  # server-assigned

experimental/bundletest/examples/orders_bundle/tests/conftest.py

@@ -4,9 +4,17 @@
 
 import pytest
 from bundletest import bundle_env
+from bundletest.env import current_backend_kind
 
 BUNDLE = str(Path(__file__).resolve().parent.parent)
 
+# This gallery declares one of every resource kind for the local backend to READ; several
+# aren't deployable (an external location, an alert query_id, models), so it is never actually
+# deployed. The cloud backend really runs `bundle deploy`, so skip the gallery there — the
+# cloud fixture is examples/cloud_orders.
+if current_backend_kind() == "cloud":
+    collect_ignore_glob = ["test_*.py"]
+
 
 @pytest.fixture
 def env():

experimental/bundletest/pyproject.toml

@@ -15,7 +15,10 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
-dev = ["pytest>=7.0"]
+# The cloud backend talks to a real workspace through the Databricks SDK. Kept optional and
+# lazily imported so a local-only install (and its DuckDB backend) never pulls it in.
+cloud = ["databricks-sdk>=0.40"]
+dev = ["pytest>=7.0", "databricks-sdk>=0.40"]
 
 [project.entry-points.pytest11]
 bundletest = "bundletest.pytest_plugin"