experimental/bundletest: base for pytest-style DABs isolation testing (DuckDB local backend)

.github/workflows/python_push.yml

@@ -66,6 +66,26 @@ jobs:
       - name: Run lint
         run: go tool -modfile=tools/task/go.mod task pydabs-lint
 
+  bundletest:
+    name: bundletest
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository and submodules
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        with:
+          python-version: "3.12"
+          version: "0.6.5"
+
+      # Nothing runs the experimental/bundletest suite otherwise. uv.lock is gitignored
+      # (internal-proxy URLs), so this resolves fresh against pypi.org on the runner.
+      - name: Run bundletest
+        working-directory: experimental/bundletest
+        run: uv run --extra dev pytest -v
+
   python_docs:
     name: docs
     runs-on: ubuntu-latest

experimental/bundletest/.gitignore

@@ -0,0 +1,10 @@
+.venv/
+__pycache__/
+*.pyc
+*.egg-info/
+.pytest_cache/
+build/
+dist/
+
+# Generated behind the internal PyPI proxy; committing it trips check-lockfiles
+uv.lock

experimental/bundletest/README.md

@@ -0,0 +1,77 @@
+# bundletest
+
+Experimental, pytest-style **isolation testing** for Databricks Asset Bundles (DABs).
+
+Unit tests answer "does my function return the right value?" `bundletest` answers the
+next question up: **"does my deployed bundle resource actually produce the right table?"**
+That class of bug — a job wired to the wrong upstream, a renamed table, a transform
+regression — passes every unit test and only shows up after `databricks bundle deploy` +
+a real run.
+
+## The idea: test one component in isolation
+
+You write a small test for a *single* resource. You stand in for its upstream neighbors
+with `env.seed(...)`, run that one resource's **real deployed SQL**, and assert on its
+real output:
+
+```python
+def test_transform_dedupes(env):
+    env.seed(
+        "shop.bronze.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 -> filtered out
+        ],
+    )
+    env.run_job("transform_orders")  # runs src/transform_orders.sql for real
+    assert env.table("shop.silver.orders").row_count() == 2
+    assert env.table("shop.silver.orders").has_no_nulls("order_id")
+    assert env.table("shop.silver.orders").column("order_id").is_unique()
+```
+
+We isolate by substituting the component's **data-boundary neighbors**, never by faking
+the component's own output — faking the thing under test is a tautology that catches
+nothing.
+
+## Two backends, one seam
+
+The same test runs against either backend, chosen by the `BUNDLETEST_BACKEND` env var:
+
+- **`local`** (default) — **portable SQL smoke testing**: runs the job's *actual* deployed
+  `.sql` artifact against [DuckDB](https://duckdb.org). Zero infra, seconds to run. Because
+  it runs the same source of truth the bundle deploys (with strict typing, no silent
+  coercion), it catches the structural bugs portable SQL can express — wrong table name,
+  broken wiring, dedup/null-filter regressions. It is **not** a Databricks SQL emulator:
+  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.)*
+
+### How the local backend stays honest
+
+- **Real artifact, not a reimplementation.** `env.run_job` reads the job's `sql_task`
+  file from `databricks.yml` and runs that exact text.
+- **Names bound at the environment level, query body never rewritten.** Seeded tables and
+  target namespaces are created under their real `catalog.schema.table` names (DuckDB
+  `ATTACH` / `CREATE SCHEMA`) so the unmodified SQL resolves against them.
+- **Three-way routing, never a false green *or* a false red:**
+  - a missing table/column is a real bug → the run **fails** (red);
+  - a Databricks-only SQL function, a notebook/Python task, or a reserved catalog name
+    (`main`/`temp`/`system`) can't be judged locally → `LocalUnsupported` → the test
+    **skips with a reason**;
+  - anything that runs and disagrees with an assertion → **red**.
+
+Assertions you *know* are cloud-only (Databricks type naming, SLA timing, permissions)
+can also be fenced explicitly with `@pytest.mark.cloud_only`, which skips them on any
+non-cloud backend.
+
+## Run it
+
+```sh
+uv venv --python 3.12
+uv pip install -e ".[dev]"
+uv run pytest -v
+```

experimental/bundletest/examples/orders_bundle/databricks.yml

@@ -0,0 +1,38 @@
+# Minimal illustrative bundle. The local backend runs a job's real sql_task artifact
+# against DuckDB; the cloud backend (follow-up PR) deploys and runs it on a real workspace.
+bundle:
+  name: orders
+
+resources:
+  jobs:
+    # SQL job — runnable on the local backend.
+    transform_orders:
+      name: transform_orders
+      tasks:
+        - task_key: transform
+          sql_task:
+            warehouse_id: ${var.warehouse_id}
+            file:
+              path: src/transform_orders.sql
+
+    # SQL job — the second hop: silver -> gold. Chains after transform_orders.
+    aggregate_orders:
+      name: aggregate_orders
+      tasks:
+        - task_key: aggregate
+          sql_task:
+            warehouse_id: ${var.warehouse_id}
+            file:
+              path: src/aggregate_orders.sql
+
+    # Notebook job — NOT runnable locally; used to demonstrate the loud skip.
+    score_model:
+      name: score_model
+      tasks:
+        - task_key: score
+          notebook_task:
+            notebook_path: src/score_model
+
+variables:
+  warehouse_id:
+    description: SQL warehouse the transform runs on

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

@@ -0,0 +1,5 @@
+-- The real deployed artifact for the aggregate_orders job.
+-- Silver -> gold: one summary row over the cleaned orders.
+CREATE OR REPLACE TABLE shop.gold.order_summary AS
+SELECT COUNT(*) AS order_count, SUM(total_price) AS total_revenue
+FROM shop.silver.orders;

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

@@ -0,0 +1,6 @@
+-- The real deployed artifact for the transform_orders job.
+-- Bronze -> silver: drop duplicate/null-id rows and pin the price to a fixed decimal.
+CREATE OR REPLACE TABLE shop.silver.orders AS
+SELECT DISTINCT order_id, CAST(total_price AS DECIMAL(10, 2)) AS total_price
+FROM shop.bronze.raw_orders
+WHERE order_id IS NOT NULL;

experimental/bundletest/examples/orders_bundle/tests/README.md

@@ -0,0 +1,18 @@
+# Example gallery
+
+One test file per resource / capability the framework supports today. Each file's header
+lists precisely what it CAN and CANNOT test locally. New files land here as the framework
+grows, so this folder doubles as a record of capability over time.
+
+Supported now (local DuckDB backend):
+- `test_job_sql.py` — SQL job: run the real `.sql`, assert on output tables
+- `test_job_config.py` — read a resource's declared wiring (no execution)
+- `test_job_nonsql.py` — non-SQL job: skips loudly (boundary demo)
+- `test_volume.py` — volume upload + read the file back (row count, columns)
+- `test_pipeline_end_to_end.py` — chain two SQL jobs (bronze → silver → gold), assert final table
+
+Not yet (need the cloud backend or new handles):
+- pipelines (Lakeflow/DLT) — run + assert on output tables
+- dashboards — assert source tables / wiring
+- alerts, permissions, clusters — config + live state
+- running Python / Scala / R / notebook jobs

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

@@ -0,0 +1,14 @@
+"""Shared fixture for the example gallery: deploy the bundle once per test."""
+
+from pathlib import Path
+
+import pytest
+from bundletest import bundle_env
+
+BUNDLE = str(Path(__file__).resolve().parent.parent)
+
+
+@pytest.fixture
+def env():
+    with bundle_env(BUNDLE) as e:
+        yield e

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

@@ -0,0 +1,26 @@
+"""Resource config / wiring — read a resource's definition without running it.
+
+The local backend reads it straight from databricks.yml; the cloud backend would read the
+deployed config from the workspace API.
+
+CAN test locally:
+- a resource exists and is named correctly
+- a job's task wiring (task_key, task type, sql file path)
+- any declared field (cluster size, schedule, parameters, ...)
+
+CANNOT test locally — needs the cloud backend:
+- that the workspace actually accepted / deployed the config
+- server-defaulted or normalized values
+
+NOTE: only 'jobs' exist in this bundle today; the same get_resource call works for any
+resource kind (pipelines, volumes, ...) once they're added.
+"""
+
+
+def test_job_is_wired_to_its_sql(env):
+    job = env.backend.get_resource("jobs", "transform_orders")
+    assert job["name"] == "transform_orders"
+
+    task = job["tasks"][0]
+    assert task["task_key"] == "transform"
+    assert task["sql_task"]["file"]["path"] == "src/transform_orders.sql"

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

@@ -0,0 +1,26 @@
+"""score_model — a notebook job (notebook_task).
+
+Non-SQL tasks (notebook / Python / Scala / R) need a real cluster, so the local backend
+cannot run them.
+
+CAN test locally:
+- that the framework refuses to run it and signals LocalUnsupported
+  (in a normal run this surfaces as a SKIP with a reason — never a silent pass or false fail)
+
+CANNOT test locally — needs the cloud backend:
+- running the job at all
+- any assertion on its output tables
+
+On the cloud backend the same env.run_job(...) + env.table(...) assertions work unchanged,
+because assertions inspect the *output*, which is language-agnostic.
+"""
+
+import pytest
+from bundletest import LocalUnsupported
+
+
+def test_notebook_job_is_refused_locally(env):
+    # Normal user code is just `env.run_job("score_model")`, which auto-skips. Here we
+    # assert the guard fires so the boundary itself is covered.
+    with pytest.raises(LocalUnsupported):
+        env.run_job("score_model")

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

@@ -0,0 +1,49 @@
+"""transform_orders — a SQL job (sql_task).
+
+The local backend runs its real artifact (src/transform_orders.sql) against DuckDB, with
+shop.bronze.raw_orders seeded to stand in for the upstream ingest job.
+
+CAN test locally (DuckDB backend):
+- job runs its real .sql without error (result.succeeded)
+- output table row count / emptiness
+- no-nulls, uniqueness, min / max on columns
+- output column names (schema keys)
+
+CANNOT test locally — needs the cloud backend:
+- exact Databricks type names (decimal(10,2) vs DuckDB's DECIMAL(10,2))
+- Databricks-only SQL functions (e.g. from_utc_timestamp) -> skips loudly
+- run duration / SLA (local wall-clock is not the cluster's)
+- real warehouse / cluster behavior
+"""
+
+import pytest
+
+
+def test_transform_dedupes(env):
+    env.seed(
+        "shop.bronze.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
+        ],
+    )
+
+    result = env.run_job("transform_orders")
+    assert result.succeeded
+
+    silver = env.table("shop.silver.orders")
+    assert silver.row_count() == 2
+    assert silver.has_no_nulls("order_id")
+    assert silver.column("order_id").is_unique()
+    assert silver.column("total_price").min() == 5.0
+    assert silver.column("total_price").max() == 10.0
+    assert set(silver.schema) == {"order_id", "total_price"}
+
+
+@pytest.mark.cloud_only
+def test_price_type_is_databricks_decimal(env):
+    env.seed("shop.bronze.raw_orders", [{"order_id": 1, "total_price": 10.0}])
+    env.run_job("transform_orders")
+    assert env.table("shop.silver.orders").schema["total_price"] == "decimal(10,2)"

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

@@ -0,0 +1,36 @@
+"""End-to-end: chain two SQL jobs through the shared local environment.
+
+bronze -> (transform_orders) -> silver -> (aggregate_orders) -> gold. Both jobs run their
+real .sql; the first's output persists in the env's DuckDB connection for the second to read.
+
+CAN test locally (DuckDB backend):
+- an all-SQL chain, hop by hop, asserting the final table
+- intermediate tables between hops (silver here)
+
+CANNOT test locally — needs the cloud backend:
+- a chain containing any notebook / Python / pipeline step -> skips at that hop
+- automatic dependency ordering (here the test sequences run_job calls itself)
+"""
+
+
+def test_bronze_to_silver_to_gold(env):
+    env.seed(
+        "shop.bronze.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
+    assert env.table("shop.silver.orders").row_count() == 2  # intermediate hop
+
+    assert env.run_job("aggregate_orders").succeeded  # silver -> gold
+
+    summary = env.table("shop.gold.order_summary")
+    assert summary.row_count() == 1
+    # single summary row: min() over the column returns its only value
+    assert summary.column("order_count").min() == 2
+    assert summary.column("total_revenue").min() == 15.0

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

@@ -0,0 +1,26 @@
+"""Volumes — upload a file and read it back.
+
+The local backend copies the uploaded file into a temp volume filesystem and reads it with
+DuckDB, so you can assert on the file you're seeding.
+
+CAN test locally:
+- an upload lands and the file exists in the volume
+- the uploaded file's row count and columns (CSV / JSON / Parquet)
+
+CANNOT test locally — needs the cloud backend:
+- that a deployed job reading the volume produced the right table
+- listing many files, overwrite semantics, permissions
+- formats DuckDB can't read locally -> skips
+"""
+
+
+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

experimental/bundletest/examples/portable_bundle/databricks.yml

@@ -0,0 +1,21 @@
+# Minimal bundle whose one SQL job runs UNCHANGED on both backends: the local (DuckDB)
+# backend runs its real sql_task artifact; the cloud backend deploys and runs it on a real
+# workspace. The single test file under tests/ is exercised on whichever tier
+# BUNDLETEST_BACKEND selects — the same assertions, no per-backend branching.
+bundle:
+  name: portable
+
+resources:
+  jobs:
+    count_active:
+      name: count_active
+      tasks:
+        - task_key: count
+          sql_task:
+            warehouse_id: ${var.warehouse_id}
+            file:
+              path: src/count_active.sql
+
+variables:
+  warehouse_id:
+    description: SQL warehouse the job runs on (cloud backend only; ignored locally)

experimental/bundletest/examples/portable_bundle/src/count_active.sql

@@ -0,0 +1,6 @@
+-- Distinct users with a login event. Portable SQL: runs identically on DuckDB and
+-- Databricks, so the test that exercises it means the same thing on both tiers.
+CREATE OR REPLACE TABLE demo.gold.active_users AS
+SELECT DISTINCT user_id
+FROM demo.bronze.events
+WHERE user_id IS NOT NULL AND event = 'login';