From 6a4468fbbf90c3e14cc471d91f3429ca2b6e589f Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Thu, 10 Sep 2026 15:19:59 +0000 Subject: [PATCH 01/12] experimental/bundletest: add the cloud backend Implements the second Backend behind the seam so the same isolation tests run against a real workspace with BUNDLETEST_BACKEND=cloud. deploy/summary/destroy go through the databricks CLI; seed/run/query go through the Databricks SDK. - execute_sql submits to the Statement Execution API, polls to terminal, paginates chunks, and casts each string cell to its manifest column type (row_count() == 2, not "2"); NULLs stay None and it raises on a failed statement. - table_schema reads DESCRIBE TABLE, returning Databricks type spelling. - run_job runs the deployed job (jobs.run_now) and maps result_state/duration. - get_resource returns the rendered `bundle summary` config, hydrating serialized_dashboard/serialized_space from the workspace so source_tables() works for file_path-only dashboards and genie spaces. - volumes: put_file/read_volume_file resolve the /Volumes// shorthand to the real UC path and parse csv/json/parquet with duckdb. The SDK is an optional `cloud` extra, lazily imported, so a local-only install never pulls it in. Adds unit tests for the pure logic (casting, literal/type rendering, namespace discovery, volume-path resolution) that run without a workspace. Co-authored-by: Isaac --- experimental/bundletest/README.md | 25 +- experimental/bundletest/pyproject.toml | 5 +- .../src/bundletest/backends/cloud.py | 317 ++++++++++++++++++ experimental/bundletest/src/bundletest/env.py | 4 +- .../bundletest/tests/test_cloud_backend.py | 151 +++++++++ 5 files changed, 498 insertions(+), 4 deletions(-) create mode 100644 experimental/bundletest/src/bundletest/backends/cloud.py create mode 100644 experimental/bundletest/tests/test_cloud_backend.py diff --git a/experimental/bundletest/README.md b/experimental/bundletest/README.md index de23eb8136d..927e375bca9 100644 --- a/experimental/bundletest/README.md +++ b/experimental/bundletest/README.md @@ -44,8 +44,9 @@ The same test runs against either backend, chosen by the `BUNDLETEST_BACKEND` en deployed job (so "wrong table name" is genuinely caught), and DuckDB's typing is strict (no silent coercion). It is not Databricks SQL, so genuinely dialect-dependent checks still 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 @@ -72,3 +73,23 @@ uv venv --python 3.12 uv pip install -e ".[dev]" uv run pytest -v ``` + +### Run it on cloud + +The cloud backend needs the Databricks SDK (the `cloud` extra) and a real workspace: + +```sh +uv pip install -e ".[cloud]" +export BUNDLETEST_BACKEND=cloud +export BUNDLETEST_PROFILE= # from ~/.databrickscfg +export BUNDLETEST_WAREHOUSE_ID= # used for seeding + assertion queries +# optional: BUNDLETEST_TARGET= +# bundle variables are read the normal DABs way, e.g. BUNDLE_VAR_warehouse_id= +uv run pytest -v +``` + +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. diff --git a/experimental/bundletest/pyproject.toml b/experimental/bundletest/pyproject.toml index 263fa6dd8df..010822dc563 100644 --- a/experimental/bundletest/pyproject.toml +++ b/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" diff --git a/experimental/bundletest/src/bundletest/backends/cloud.py b/experimental/bundletest/src/bundletest/backends/cloud.py new file mode 100644 index 00000000000..6d66925136c --- /dev/null +++ b/experimental/bundletest/src/bundletest/backends/cloud.py @@ -0,0 +1,317 @@ +"""Cloud backend: runs the same tests against a real Databricks workspace. + +The fidelity tier behind the ``Backend`` seam. Where the local DuckDB backend simulates, +this deploys the bundle for real and drives the workspace, so the ``cloud_only`` assertions +(Databricks type names, SLA timing, notebook/Python jobs, permissions) actually run instead +of skipping. + +Design invariants — the cloud analogues of the DuckDB backend's: + +1. **Same source of truth.** ``run_job`` runs the *deployed* job (``jobs.run_now`` on the + real cluster/warehouse), not a reimplementation — the whole point of the fidelity tier. + +2. **Real names, no rewriting.** Seeded tables and job targets use their real + ``catalog.schema.table`` names. Missing schemas are created (``CREATE SCHEMA IF NOT + EXISTS``) — the UC equivalent of the local backend's ATTACH; there is no reserved-catalog + restriction here (``main`` is a fine UC catalog), so this backend never raises + ``LocalUnsupported``. + +3. **Fail loud.** A failed statement raises; a failed job run comes back as a FAILED + ``RunResult``. Nothing is silently coerced to green. + +Config comes from the environment: ``BUNDLETEST_PROFILE`` (CLI/SDK auth profile), +``BUNDLETEST_WAREHOUSE_ID`` (the SQL warehouse for seeding/queries; required for any SQL), +``BUNDLETEST_TARGET`` (bundle target). Bundle variables are supplied the normal DABs way +(``BUNDLE_VAR_`` env vars or the target), so ``deploy`` stays bundle-agnostic. +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import tempfile +import time +from pathlib import Path +from typing import Any + +import duckdb +import yaml + +from bundletest.backend import RunResult + +# DuckDB reader per file extension, reused to parse a volume file downloaded from the +# workspace (duckdb is already a base dependency, so no extra parser is pulled in). +_FILE_READERS = {".csv": "read_csv", ".json": "read_json", ".parquet": "read_parquet"} + +# A three-part UC name catalog.schema.table. Anchored to an identifier start so numeric +# literals never match. Only used to discover which schemas a job's SQL writes to, so +# over-matching (e.g. a column ref) is harmless — CREATE SCHEMA IF NOT EXISTS is idempotent. +_QUALIFIED = re.compile(r"\b([A-Za-z_]\w*)\.([A-Za-z_]\w*)\.[A-Za-z_]\w*") + +# Statement Execution API column type_name -> value caster. Every value arrives as a string +# in data_array; the assertion layer needs native types (row_count() == 2, not "2"). +_INT_TYPES = {"BYTE", "SHORT", "INT", "LONG"} +_FLOAT_TYPES = {"FLOAT", "DOUBLE", "DECIMAL"} + + +def _sql_type(values: list[Any]) -> str: + for v in values: + if v is None: + continue + if isinstance(v, bool): + return "BOOLEAN" + if isinstance(v, int): + return "BIGINT" + if isinstance(v, float): + return "DOUBLE" + return "STRING" + return "STRING" + + +def _sql_literal(v: Any) -> str: + if v is None: + return "NULL" + if isinstance(v, bool): + return "true" if v else "false" + if isinstance(v, (int, float)): + return repr(v) + # Spark SQL string literals honor backslash escapes, so escape the backslash before + # doubling the quote — otherwise a literal '\' or "'" in seed data breaks the INSERT. + return "'" + str(v).replace("\\", "\\\\").replace("'", "''") + "'" + + +def _cast_value(raw: str | None, type_name: str) -> Any: + """Cast one string cell from data_array to the native type its column declares.""" + if raw is None: + return None + if type_name == "BOOLEAN": + return raw.lower() == "true" + if type_name in _INT_TYPES: + return int(raw) + if type_name in _FLOAT_TYPES: + return float(raw) + return raw + + +def _schemas_in(sql: str) -> list[str]: + """The distinct ``catalog.schema`` prefixes of the three-part names in ``sql``.""" + out: list[str] = [] + for m in _QUALIFIED.finditer(sql): + ns = f"{m.group(1)}.{m.group(2)}" + if ns not in out: + out.append(ns) + return out + + +class CloudBackend: + """Cloud ``Backend`` implementation. Deploys the bundle and drives a real workspace.""" + + def __init__( + self, + profile: str | None = None, + warehouse_id: str | None = None, + target: str | None = None, + ) -> None: + self._profile = profile or os.environ.get("BUNDLETEST_PROFILE") + self._warehouse_id = warehouse_id or os.environ.get("BUNDLETEST_WAREHOUSE_ID") + self._target = target or os.environ.get("BUNDLETEST_TARGET") + self._client: Any = None # lazy WorkspaceClient; constructing it resolves auth + self._bundle_path = "" + self._config: dict[str, Any] = {} + self._summary: dict[str, Any] | None = None + self._seeded: set[str] = set() + + # --- lifecycle --- + def deploy(self, bundle_path: str) -> None: + self._bundle_path = bundle_path + self._summary = None + path = Path(bundle_path, "databricks.yml") + self._config = yaml.safe_load(path.read_text()) if path.exists() else {} + self._bundle("deploy") + + def teardown(self) -> None: + # Drop what we seeded, then destroy the bundle. Best-effort: teardown must not raise. + for fqn in self._seeded: + try: + self.execute_sql(f"DROP TABLE IF EXISTS {fqn}") + except Exception: + pass + try: + self._bundle("destroy", "--auto-approve") + except Exception: + pass + + # --- scaffolding --- + def seed_table(self, fqn: str, rows: list[dict[str, Any]]) -> None: + if not rows: + raise ValueError(f"cannot seed {fqn!r} with no rows") + self._ensure_schema(fqn) + columns = list(rows[0].keys()) + coldefs = ", ".join(f"{c} {_sql_type([r.get(c) for r in rows])}" for c in columns) + self.execute_sql(f"CREATE OR REPLACE TABLE {fqn} ({coldefs})") + values = ", ".join("(" + ", ".join(_sql_literal(r.get(c)) for c in columns) + ")" for r in rows) + self.execute_sql(f"INSERT INTO {fqn} ({', '.join(columns)}) VALUES {values}") + self._seeded.add(fqn) + + # --- execution --- + def run_job(self, name: str, params: dict[str, Any] | None = None) -> RunResult: + from databricks.sdk.service.jobs import RunResultState + + job_id = int(self.get_resource("jobs", name)["id"]) + self._ensure_job_schemas(name) + job_params = {k: str(v) for k, v in (params or {}).items()} + run = self._ws().jobs.run_now(job_id, job_parameters=job_params or None).result() + succeeded = run.state.result_state == RunResultState.SUCCESS + return RunResult( + "SUCCESS" if succeeded else "FAILED", + (run.run_duration or 0) / 1000, + str(run.run_id), + error="" if succeeded else (run.state.state_message or ""), + ) + + # --- data plane --- + def execute_sql(self, query: str) -> list[tuple]: + from databricks.sdk.service.sql import Disposition, Format, StatementState + + warehouse_id = self._require_warehouse() + se = self._ws().statement_execution + resp = se.execute_statement( + statement=query, + warehouse_id=warehouse_id, + wait_timeout="30s", + disposition=Disposition.INLINE, + format=Format.JSON_ARRAY, + ) + while resp.status.state in (StatementState.PENDING, StatementState.RUNNING): + time.sleep(1) + resp = se.get_statement(resp.statement_id) + if resp.status.state != StatementState.SUCCEEDED: + err = resp.status.error + detail = err.message if err else resp.status.state.value + raise RuntimeError(f"statement failed: {detail}") + + columns = resp.manifest.schema.columns or [] + types = [c.type_name.value for c in columns] + rows = list(resp.result.data_array or []) + # Inline results past the first chunk are fetched by index. + nxt = resp.result.next_chunk_index + while nxt is not None: + chunk = se.get_statement_result_chunk_n(resp.statement_id, nxt) + rows += chunk.data_array or [] + nxt = chunk.next_chunk_index + return [tuple(_cast_value(v, types[i]) for i, v in enumerate(row)) for row in rows] + + def table_schema(self, fqn: str) -> dict[str, str]: + # DESCRIBE emits (col_name, data_type, comment); trailing partition/detail rows have a + # blank or '#'-prefixed col_name. data_type is already the Databricks spelling + # (e.g. "decimal(10,2)"), which is exactly what the cloud_only type test wants. + schema: dict[str, str] = {} + for name, dtype, *_ in self.execute_sql(f"DESCRIBE TABLE {fqn}"): + if not name or name.startswith("#"): + break + schema[name] = dtype + return schema + + # --- control plane --- + def get_resource(self, kind: str, name: str) -> dict[str, Any]: + # `...[kind][name]` raises KeyError for an undeclared resource, which exists() expects. + cfg = self._resources()[kind][name] + # source_tables() needs the rendered serialized definition. A resource inlined in + # databricks.yml already carries it; a file_path-only one does not, so read it back + # from the deployed resource. + if kind == "dashboards" and "serialized_dashboard" not in cfg: + dashboard = self._ws().lakeview.get(self._deployed_id(cfg, name)) + return {**cfg, "serialized_dashboard": dashboard.serialized_dashboard} + if kind == "genie_spaces" and "serialized_space" not in cfg: + space = self._ws().genie.get_space(self._deployed_id(cfg, name), include_serialized_space=True) + return {**cfg, "serialized_space": space.serialized_space} + return cfg + + def put_file(self, dst: str, src: str) -> None: + if not os.path.exists(src): + raise FileNotFoundError(f"upload source not found: {src}") + with open(src, "rb") as f: + self._ws().files.upload(self._volume_path(dst), f, overwrite=True) + + def read_volume_file(self, volume: str, filename: str) -> list[dict[str, Any]]: + path = self._volume_path(f"/Volumes/{volume}/{filename}") + suffix = Path(filename).suffix + reader = _FILE_READERS.get(suffix) + if reader is None: + raise ValueError(f"cannot read {suffix!r} files") + try: + resp = self._ws().files.download(path) + except Exception as e: + raise FileNotFoundError(f"no file {filename!r} in volume {volume!r}: {e}") from e + with tempfile.NamedTemporaryFile(suffix=suffix) as tmp: + tmp.write(resp.contents.read()) + tmp.flush() + cur = duckdb.connect().execute(f"SELECT * FROM {reader}('{tmp.name}')") + cols = [d[0] for d in cur.description] + return [dict(zip(cols, row, strict=True)) for row in cur.fetchall()] + + # --- internals --- + def _ws(self) -> Any: + if self._client is None: + from databricks.sdk import WorkspaceClient + + self._client = WorkspaceClient(profile=self._profile) + return self._client + + def _require_warehouse(self) -> str: + if not self._warehouse_id: + raise RuntimeError("cloud SQL needs a warehouse — set BUNDLETEST_WAREHOUSE_ID") + return self._warehouse_id + + def _bundle(self, *args: str) -> str: + cmd = ["databricks", "bundle", *args] + if self._target: + cmd += ["-t", self._target] + env = dict(os.environ) + if self._profile: + env["DATABRICKS_CONFIG_PROFILE"] = self._profile + out = subprocess.run(cmd, cwd=self._bundle_path, env=env, capture_output=True, text=True) + if out.returncode != 0: + raise RuntimeError(f"`{' '.join(cmd)}` failed: {out.stderr.strip()}") + return out.stdout + + def _resources(self) -> dict[str, Any]: + if self._summary is None: + self._summary = json.loads(self._bundle("summary", "-o", "json")) + return self._summary.get("resources", {}) + + @staticmethod + def _deployed_id(cfg: dict[str, Any], name: str) -> str: + # A RuntimeError (not KeyError) so exists() doesn't mistake a hydration miss for absence. + rid = cfg.get("id") + if not rid: + raise RuntimeError(f"resource {name!r} has no deployed id yet — deploy first") + return str(rid) + + def _ensure_schema(self, fqn: str) -> None: + parts = fqn.split(".") + if len(parts) == 3: + self.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {parts[0]}.{parts[1]}") + + def _ensure_job_schemas(self, name: str) -> None: + """Create the schemas a job's SQL writes to, so its unmodified statements resolve — + the cloud analogue of the local backend's namespace preparation.""" + for task in self._config.get("resources", {}).get("jobs", {}).get(name, {}).get("tasks", []): + sql_task = task.get("sql_task") + if not sql_task: + continue + sql = Path(self._bundle_path, sql_task["file"]["path"]).read_text() + for ns in _schemas_in(sql): + self.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {ns}") + + def _volume_path(self, dst: str) -> str: + """Resolve ``/Volumes//`` to a real UC volume path. + + The first segment after ``/Volumes/`` is the volume *resource* name — the same + contract the local backend uses on both put_file and read_volume_file — which we + expand to ``/Volumes////`` via the deployed config.""" + _, resource, *rest = dst.strip("/").split("/") + vol = self.get_resource("volumes", resource) + return "/".join(["/Volumes", vol["catalog_name"], vol["schema_name"], vol["name"], *rest]) diff --git a/experimental/bundletest/src/bundletest/env.py b/experimental/bundletest/src/bundletest/env.py index e985e49071b..66d10a34f01 100644 --- a/experimental/bundletest/src/bundletest/env.py +++ b/experimental/bundletest/src/bundletest/env.py @@ -278,7 +278,9 @@ def make_backend(kind: str, **kwargs: Any) -> Backend: return DuckDBBackend(**kwargs) if kind == "cloud": - raise NotImplementedError("the cloud backend arrives in a follow-up PR on top of this base branch") + from bundletest.backends.cloud import CloudBackend + + return CloudBackend(**kwargs) raise ValueError(f"unknown backend {kind!r} (expected 'local' or 'cloud')") diff --git a/experimental/bundletest/tests/test_cloud_backend.py b/experimental/bundletest/tests/test_cloud_backend.py new file mode 100644 index 00000000000..627ca0b9db2 --- /dev/null +++ b/experimental/bundletest/tests/test_cloud_backend.py @@ -0,0 +1,151 @@ +"""Cloud backend unit tests that need no workspace. + +The workspace round-trips (deploy, run, statement submit) can only be exercised against a +real cloud, but the logic most likely to be wrong is pure and testable here: casting the +Statement Execution API's all-string results to native types (the base-compat trap), +literal/type rendering for seeding, namespace discovery, and volume-path resolution. +""" + +from types import SimpleNamespace + +import pytest +from bundletest.backends.cloud import ( + CloudBackend, + _cast_value, + _schemas_in, + _sql_literal, + _sql_type, +) +from databricks.sdk.service.sql import ColumnInfoTypeName as T +from databricks.sdk.service.sql import StatementState + + +def test_sql_type_inference(): + assert _sql_type([True, False]) == "BOOLEAN" + assert _sql_type([1, 2]) == "BIGINT" + assert _sql_type([1.5]) == "DOUBLE" + assert _sql_type(["x"]) == "STRING" + assert _sql_type([None, None]) == "STRING" # all-null -> default + assert _sql_type([None, 3]) == "BIGINT" # first non-null wins + + +def test_sql_literal_escaping(): + assert _sql_literal(None) == "NULL" + assert _sql_literal(True) == "true" + assert _sql_literal(5) == "5" + assert _sql_literal(2.5) == "2.5" + assert _sql_literal("a'b") == "'a''b'" # single quotes doubled + assert _sql_literal("a\\b") == "'a\\\\b'" # backslash escaped for Spark SQL + + +def test_cast_value_by_type(): + assert _cast_value(None, "INT") is None + assert _cast_value("2", "LONG") == 2 + assert _cast_value("2.5", "DOUBLE") == 2.5 + assert _cast_value("15.00", "DECIMAL") == 15.0 + assert _cast_value("true", "BOOLEAN") is True + assert _cast_value("false", "BOOLEAN") is False + assert _cast_value("hi", "STRING") == "hi" + + +def test_schemas_in_scans_three_part_names_only(): + sql = "CREATE TABLE shop.silver.orders AS SELECT * FROM shop.bronze.raw WHERE x = 10.00" + # 10.00 is a literal, not a namespace; each catalog.schema appears once. + assert _schemas_in(sql) == ["shop.silver", "shop.bronze"] + + +def _column(name, type_name): + return SimpleNamespace(name=name, type_name=type_name) + + +def _resp(state, columns=None, data=None, next_chunk_index=None, error=None): + return SimpleNamespace( + statement_id="s1", + status=SimpleNamespace(state=state, error=error), + manifest=SimpleNamespace(schema=SimpleNamespace(columns=columns or [])), + result=SimpleNamespace(data_array=data, next_chunk_index=next_chunk_index), + ) + + +class _FakeStatements: + """Stand-in for w.statement_execution: canned submit + chunk fetch, no network.""" + + def __init__(self, first, chunks=None, poll=None): + self._first = first + self._chunks = chunks or {} + self._poll = list(poll or []) + + def execute_statement(self, **kwargs): + return self._first + + def get_statement(self, statement_id): + return self._poll.pop(0) + + def get_statement_result_chunk_n(self, statement_id, chunk_index): + return self._chunks[chunk_index] + + +def _backend_with(statements): + be = CloudBackend(warehouse_id="w") + be._client = SimpleNamespace(statement_execution=statements) + return be + + +def test_execute_sql_casts_and_paginates_chunks(): + cols = [_column("i", T.INT), _column("d", T.DOUBLE), _column("s", T.STRING)] + first = _resp( + StatementState.SUCCEEDED, + columns=cols, + data=[["1", "1.5", "a"]], + next_chunk_index=1, + ) + chunk1 = SimpleNamespace(data_array=[["2", "2.5", "b"]], next_chunk_index=None) + be = _backend_with(_FakeStatements(first, chunks={1: chunk1})) + + assert be.execute_sql("SELECT ...") == [(1, 1.5, "a"), (2, 2.5, "b")] + + +def test_execute_sql_polls_until_terminal(): + cols = [_column("n", T.LONG)] + pending = _resp(StatementState.PENDING) + done = _resp(StatementState.SUCCEEDED, columns=cols, data=[["42"]]) + be = _backend_with(_FakeStatements(pending, poll=[done])) + + assert be.execute_sql("SELECT 42") == [(42,)] + + +def test_execute_sql_raises_on_failure(): + failed = _resp( + StatementState.FAILED, + error=SimpleNamespace(message="Table not found: nope"), + ) + be = _backend_with(_FakeStatements(failed)) + with pytest.raises(RuntimeError, match="Table not found"): + be.execute_sql("SELECT * FROM nope") + + +def test_execute_sql_requires_warehouse(): + be = CloudBackend() # no warehouse configured + be._warehouse_id = None + with pytest.raises(RuntimeError, match="BUNDLETEST_WAREHOUSE_ID"): + be.execute_sql("SELECT 1") + + +def test_volume_path_resolves_resource_name(): + be = CloudBackend() + # Pretend the bundle summary is already loaded, so no subprocess/client is touched. + be._summary = { + "resources": {"volumes": {"raw_data": {"catalog_name": "shop", "schema_name": "bronze", "name": "raw_data"}}} + } + assert be._volume_path("/Volumes/raw_data/orders.csv") == "/Volumes/shop/bronze/raw_data/orders.csv" + # A nested path keeps its tail. + assert be._volume_path("/Volumes/raw_data/sub/f.csv") == "/Volumes/shop/bronze/raw_data/sub/f.csv" + + +def test_get_resource_keeps_inline_serialized_dashboard(): + be = CloudBackend() + inline = {"serialized_dashboard": {"datasets": []}, "id": "abc"} + be._summary = {"resources": {"dashboards": {"d": inline}}} + # Already inline -> returned as-is, no workspace call (client stays None). + assert be.get_resource("dashboards", "d") is inline + assert be._client is None From d9abb117f477f4e741cd26dc79cced8870e56bb0 Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Thu, 10 Sep 2026 15:24:53 +0000 Subject: [PATCH 02/12] experimental/bundletest: guard execute_sql for no-result-set statements MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A successful DDL/DML statement (CREATE/INSERT/DROP, and the schema preparation that seeding does) returns SUCCEEDED with result=None and no manifest, so reading result.data_array raised AttributeError — on the very first real seed. Return an empty list in that case, and pin it with a result=None test case the fake missed. Co-authored-by: Isaac --- .../bundletest/src/bundletest/backends/cloud.py | 6 +++++- experimental/bundletest/tests/test_cloud_backend.py | 13 +++++++++++++ 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/experimental/bundletest/src/bundletest/backends/cloud.py b/experimental/bundletest/src/bundletest/backends/cloud.py index 6d66925136c..4a1f41fa4f7 100644 --- a/experimental/bundletest/src/bundletest/backends/cloud.py +++ b/experimental/bundletest/src/bundletest/backends/cloud.py @@ -192,7 +192,11 @@ def execute_sql(self, query: str) -> list[tuple]: detail = err.message if err else resp.status.state.value raise RuntimeError(f"statement failed: {detail}") - columns = resp.manifest.schema.columns or [] + # A successful DDL/DML statement (CREATE/INSERT/DROP, and the schema-prep here) + # returns no result set, so there is nothing to read or cast. + if resp.result is None: + return [] + columns = (resp.manifest.schema.columns if resp.manifest and resp.manifest.schema else None) or [] types = [c.type_name.value for c in columns] rows = list(resp.result.data_array or []) # Inline results past the first chunk are fetched by index. diff --git a/experimental/bundletest/tests/test_cloud_backend.py b/experimental/bundletest/tests/test_cloud_backend.py index 627ca0b9db2..3b7c8f0e960 100644 --- a/experimental/bundletest/tests/test_cloud_backend.py +++ b/experimental/bundletest/tests/test_cloud_backend.py @@ -105,6 +105,19 @@ def test_execute_sql_casts_and_paginates_chunks(): assert be.execute_sql("SELECT ...") == [(1, 1.5, "a"), (2, 2.5, "b")] +def test_execute_sql_returns_empty_for_ddl_with_no_result_set(): + # A successful CREATE/INSERT/DROP comes back SUCCEEDED with result=None (and no manifest); + # touching result.data_array would crash. seed/teardown/schema-prep all rely on this. + ddl = SimpleNamespace( + statement_id="s1", + status=SimpleNamespace(state=StatementState.SUCCEEDED, error=None), + manifest=None, + result=None, + ) + be = _backend_with(_FakeStatements(ddl)) + assert be.execute_sql("CREATE OR REPLACE TABLE t (a INT)") == [] + + def test_execute_sql_polls_until_terminal(): cols = [_column("n", T.LONG)] pending = _resp(StatementState.PENDING) From ccc43320e20bea48cccd54fa0be916241c2760fa Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Thu, 10 Sep 2026 15:44:54 +0000 Subject: [PATCH 03/12] experimental/bundletest: escape SQL string literals the Spark way MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live validation against a real warehouse showed a doubled single quote ('') does NOT escape a quote in Databricks/Spark SQL — it drops it ('a''b' -> ab). Spark escapes string literals with a backslash, so seed values with an embedded quote were silently corrupted. Escape both the backslash and the single quote with a backslash instead (verified end-to-end: 'a''b'->ab vs 'a\'b'->a'b on the warehouse). Co-authored-by: Isaac --- experimental/bundletest/src/bundletest/backends/cloud.py | 7 ++++--- experimental/bundletest/tests/test_cloud_backend.py | 5 +++-- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/experimental/bundletest/src/bundletest/backends/cloud.py b/experimental/bundletest/src/bundletest/backends/cloud.py index 4a1f41fa4f7..1fc876ded52 100644 --- a/experimental/bundletest/src/bundletest/backends/cloud.py +++ b/experimental/bundletest/src/bundletest/backends/cloud.py @@ -77,9 +77,10 @@ def _sql_literal(v: Any) -> str: return "true" if v else "false" if isinstance(v, (int, float)): return repr(v) - # Spark SQL string literals honor backslash escapes, so escape the backslash before - # doubling the quote — otherwise a literal '\' or "'" in seed data breaks the INSERT. - return "'" + str(v).replace("\\", "\\\\").replace("'", "''") + "'" + # Databricks/Spark SQL escapes string literals with a backslash, and (unlike ANSI SQL) + # a doubled quote '' is NOT an escaped quote — it drops the quote. So escape the + # backslash first, then the single quote, both with a backslash. + return "'" + str(v).replace("\\", "\\\\").replace("'", "\\'") + "'" def _cast_value(raw: str | None, type_name: str) -> Any: diff --git a/experimental/bundletest/tests/test_cloud_backend.py b/experimental/bundletest/tests/test_cloud_backend.py index 3b7c8f0e960..a2e219f6d9b 100644 --- a/experimental/bundletest/tests/test_cloud_backend.py +++ b/experimental/bundletest/tests/test_cloud_backend.py @@ -34,8 +34,9 @@ def test_sql_literal_escaping(): assert _sql_literal(True) == "true" assert _sql_literal(5) == "5" assert _sql_literal(2.5) == "2.5" - assert _sql_literal("a'b") == "'a''b'" # single quotes doubled - assert _sql_literal("a\\b") == "'a\\\\b'" # backslash escaped for Spark SQL + # Spark escapes with a backslash; a doubled quote would drop the quote (verified live). + assert _sql_literal("a'b") == "'a\\'b'" # single quote -> \' + assert _sql_literal("a\\b") == "'a\\\\b'" # backslash -> \\ def test_cast_value_by_type(): From 79deeb54ebe1fa40eab9831e6f83429cec140f41 Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Fri, 11 Sep 2026 09:17:52 +0000 Subject: [PATCH 04/12] experimental/bundletest: add a deployable cloud E2E validation bundle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The local gallery (examples/orders_bundle) declares one of every resource kind for static-config reading and is deliberately NOT deployable (fake external locations, metastore-admin catalog, nonexistent models). The cloud backend really deploys, so it needs a deployable fixture: two SQL jobs, a managed volume, and a file_path dashboard, all under main.bundletest_cloud. The module-scoped fixture creates that schema, deploys, and drops it CASCADE afterwards; its tests are collected only when BUNDLETEST_BACKEND=cloud. Validated end-to-end against a real workspace (azure-dogfood): deploy, run_job on both deployed jobs, execute_sql/table_schema round-trips, get_resource off `bundle summary` (job id, volume catalog/schema/name, and the file_path dashboard's serialized_dashboard, which the summary inlines), source_tables, and volume upload/read — 5 passed, clean teardown. Co-authored-by: Isaac --- .../dashboards/orders_overview.lvdash.json | 19 ++++++ .../examples/cloud_orders/databricks.yml | 51 ++++++++++++++ .../cloud_orders/src/aggregate_orders.sql | 4 ++ .../cloud_orders/src/transform_orders.sql | 5 ++ .../examples/cloud_orders/tests/conftest.py | 37 +++++++++++ .../cloud_orders/tests/test_cloud_e2e.py | 66 +++++++++++++++++++ 6 files changed, 182 insertions(+) create mode 100644 experimental/bundletest/examples/cloud_orders/dashboards/orders_overview.lvdash.json create mode 100644 experimental/bundletest/examples/cloud_orders/databricks.yml create mode 100644 experimental/bundletest/examples/cloud_orders/src/aggregate_orders.sql create mode 100644 experimental/bundletest/examples/cloud_orders/src/transform_orders.sql create mode 100644 experimental/bundletest/examples/cloud_orders/tests/conftest.py create mode 100644 experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py diff --git a/experimental/bundletest/examples/cloud_orders/dashboards/orders_overview.lvdash.json b/experimental/bundletest/examples/cloud_orders/dashboards/orders_overview.lvdash.json new file mode 100644 index 00000000000..7d85dedbfc1 --- /dev/null +++ b/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": [] + } + ] +} diff --git a/experimental/bundletest/examples/cloud_orders/databricks.yml b/experimental/bundletest/examples/cloud_orders/databricks.yml new file mode 100644 index 00000000000..d4c16a7752e --- /dev/null +++ b/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 diff --git a/experimental/bundletest/examples/cloud_orders/src/aggregate_orders.sql b/experimental/bundletest/examples/cloud_orders/src/aggregate_orders.sql new file mode 100644 index 00000000000..f7e115c0047 --- /dev/null +++ b/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; diff --git a/experimental/bundletest/examples/cloud_orders/src/transform_orders.sql b/experimental/bundletest/examples/cloud_orders/src/transform_orders.sql new file mode 100644 index 00000000000..436a1f7ba2b --- /dev/null +++ b/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; diff --git a/experimental/bundletest/examples/cloud_orders/tests/conftest.py b/experimental/bundletest/examples/cloud_orders/tests/conftest.py new file mode 100644 index 00000000000..90243d835fd --- /dev/null +++ b/experimental/bundletest/examples/cloud_orders/tests/conftest.py @@ -0,0 +1,37 @@ +"""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; on the local backend there is nothing to run. The env is +module-scoped (deploy once — deploys take minutes and cost compute, unlike the local +backend's fresh per-test DuckDB). The target schema is created up front and dropped +CASCADE afterwards, so the run leaves nothing behind even if `bundle destroy` half-fails. +""" + +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) +SCHEMA = "main.bundletest_cloud" + +# 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 env(): + backend = make_backend("cloud") + backend.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}") + e = BundleEnv(BUNDLE, backend) + e.deploy() + try: + yield e + finally: + e.teardown() + try: + backend.execute_sql(f"DROP SCHEMA IF EXISTS {SCHEMA} CASCADE") + except Exception: + pass diff --git a/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py b/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py new file mode 100644 index 00000000000..7452b5e1923 --- /dev/null +++ b/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py @@ -0,0 +1,66 @@ +"""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` +(including hydrating a file_path dashboard's serialized form), and volume upload/read. +""" + +import pytest + +SCHEMA = "main.bundletest_cloud" + + +def test_bronze_to_silver_to_gold(env): + 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): + 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_serialized_is_hydrated_from_file_path(env): + # The dashboard is defined by file_path, so its serialized form isn't in databricks.yml; + # the cloud backend must read it back from the deployed dashboard for source_tables(). + 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 From 7792599a8092543fce02c0168234749d100a029e Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Fri, 11 Sep 2026 09:20:42 +0000 Subject: [PATCH 05/12] experimental/bundletest: drop dead serialized_* hydration from get_resource MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live deploy showed `bundle summary -o json` already inlines serialized_dashboard for a file_path dashboard, and the config mutators confirm the same for serialized_space (configure_dashboards_serialized_dashboard / configure_genie_space_serialized_space read the file_path at load time). So the `if "serialized_dashboard" not in cfg` hydration branches could never fire — get_resource just returns the rendered summary config, which already carries the serialized form source_tables() needs. Removes the dead lakeview.get / genie.get_space branches and the _deployed_id helper. Verified: the cloud E2E dashboard source_tables test (file_path source) still passes. Co-authored-by: Isaac --- .../src/bundletest/backends/cloud.py | 23 ++++--------------- 1 file changed, 4 insertions(+), 19 deletions(-) diff --git a/experimental/bundletest/src/bundletest/backends/cloud.py b/experimental/bundletest/src/bundletest/backends/cloud.py index 1fc876ded52..b56f7a77452 100644 --- a/experimental/bundletest/src/bundletest/backends/cloud.py +++ b/experimental/bundletest/src/bundletest/backends/cloud.py @@ -222,17 +222,10 @@ def table_schema(self, fqn: str) -> dict[str, str]: # --- control plane --- def get_resource(self, kind: str, name: str) -> dict[str, Any]: # `...[kind][name]` raises KeyError for an undeclared resource, which exists() expects. - cfg = self._resources()[kind][name] - # source_tables() needs the rendered serialized definition. A resource inlined in - # databricks.yml already carries it; a file_path-only one does not, so read it back - # from the deployed resource. - if kind == "dashboards" and "serialized_dashboard" not in cfg: - dashboard = self._ws().lakeview.get(self._deployed_id(cfg, name)) - return {**cfg, "serialized_dashboard": dashboard.serialized_dashboard} - if kind == "genie_spaces" and "serialized_space" not in cfg: - space = self._ws().genie.get_space(self._deployed_id(cfg, name), include_serialized_space=True) - return {**cfg, "serialized_space": space.serialized_space} - return cfg + # `bundle summary` gives the rendered config, and its config mutators inline + # serialized_dashboard/serialized_space from a file_path at load time, so the + # serialized form source_tables() needs is already here — no workspace read needed. + return self._resources()[kind][name] def put_file(self, dst: str, src: str) -> None: if not os.path.exists(src): @@ -287,14 +280,6 @@ def _resources(self) -> dict[str, Any]: self._summary = json.loads(self._bundle("summary", "-o", "json")) return self._summary.get("resources", {}) - @staticmethod - def _deployed_id(cfg: dict[str, Any], name: str) -> str: - # A RuntimeError (not KeyError) so exists() doesn't mistake a hydration miss for absence. - rid = cfg.get("id") - if not rid: - raise RuntimeError(f"resource {name!r} has no deployed id yet — deploy first") - return str(rid) - def _ensure_schema(self, fqn: str) -> None: parts = fqn.split(".") if len(parts) == 3: From 56751ae416e5e4bc9dc608831b9037ed15dcc9a1 Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Fri, 11 Sep 2026 09:24:39 +0000 Subject: [PATCH 06/12] experimental/bundletest: document the cloud fixture in the README Point the "Run it on cloud" section at the examples/cloud_orders fixture and the actual BUNDLETEST_BACKEND=cloud invocation. Co-authored-by: Isaac --- experimental/bundletest/README.md | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/experimental/bundletest/README.md b/experimental/bundletest/README.md index 927e375bca9..78178511322 100644 --- a/experimental/bundletest/README.md +++ b/experimental/bundletest/README.md @@ -76,18 +76,18 @@ uv run pytest -v ### Run it on cloud -The cloud backend needs the Databricks SDK (the `cloud` extra) and a real workspace: +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 -uv pip install -e ".[cloud]" export BUNDLETEST_BACKEND=cloud -export BUNDLETEST_PROFILE= # from ~/.databrickscfg -export BUNDLETEST_WAREHOUSE_ID= # used for seeding + assertion queries -# optional: BUNDLETEST_TARGET= -# bundle variables are read the normal DABs way, e.g. BUNDLE_VAR_warehouse_id= -uv run pytest -v +export BUNDLETEST_PROFILE= # from ~/.databrickscfg +export BUNDLETEST_WAREHOUSE_ID= # used for seeding + assertion queries +export BUNDLE_VAR_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 From 0c3998376d3ee3220877213bafe7315ec274522b Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Fri, 11 Sep 2026 09:25:48 +0000 Subject: [PATCH 07/12] experimental/bundletest: expand cloud backend unit tests Cover run_job result mapping (SUCCESS/FAILED, state_message, run_duration ms->seconds), table_schema DESCRIBE row filtering (blank / '#'-prefixed), seed_table CREATE+INSERT with inferred types and Spark-escaped literals + teardown tracking, and best-effort teardown. All via the fake-client pattern, no workspace. 22 tests in the file, full suite 73 passed. Co-authored-by: Isaac --- .../bundletest/tests/test_cloud_backend.py | 263 ++++++++++++++++++ 1 file changed, 263 insertions(+) diff --git a/experimental/bundletest/tests/test_cloud_backend.py b/experimental/bundletest/tests/test_cloud_backend.py index a2e219f6d9b..f060de82606 100644 --- a/experimental/bundletest/tests/test_cloud_backend.py +++ b/experimental/bundletest/tests/test_cloud_backend.py @@ -163,3 +163,266 @@ def test_get_resource_keeps_inline_serialized_dashboard(): # Already inline -> returned as-is, no workspace call (client stays None). assert be.get_resource("dashboards", "d") is inline assert be._client is None + + +def test_run_job_success_result_mapping(): + from databricks.sdk.service.jobs import RunResultState + + # Successful job run: succeeded=True, run_duration ms -> seconds conversion + be = CloudBackend(warehouse_id="w") + be._summary = {"resources": {"jobs": {"myjob": {"id": "123"}}}} + be._config = {"resources": {"jobs": {"myjob": {"tasks": []}}}} + # Mock the workspace client + be._client = SimpleNamespace( + jobs=SimpleNamespace( + run_now=lambda job_id, job_parameters=None: SimpleNamespace( + result=lambda: SimpleNamespace( + state=SimpleNamespace(result_state=RunResultState.SUCCESS), + run_duration=2000, + run_id=456, + ) + ) + ), + statement_execution=_FakeStatements(_resp(StatementState.SUCCEEDED)), + ) + + result = be.run_job("myjob") + assert result.result_state == "SUCCESS" + assert result.succeeded is True + assert result.duration_seconds == 2.0 + assert result.run_id == "456" + assert result.error == "" + + +def test_run_job_failed_result_mapping(): + from databricks.sdk.service.jobs import RunResultState + + # Failed job run: succeeded=False, error populated, run_duration converted + be = CloudBackend(warehouse_id="w") + be._summary = {"resources": {"jobs": {"failing_job": {"id": "789"}}}} + be._config = {"resources": {"jobs": {"failing_job": {"tasks": []}}}} + be._client = SimpleNamespace( + jobs=SimpleNamespace( + run_now=lambda job_id, job_parameters=None: SimpleNamespace( + result=lambda: SimpleNamespace( + state=SimpleNamespace( + result_state=RunResultState.FAILED, + state_message="Task failed: invalid syntax", + ), + run_duration=5000, + run_id=999, + ) + ) + ), + statement_execution=_FakeStatements(_resp(StatementState.SUCCEEDED)), + ) + + result = be.run_job("failing_job") + assert result.result_state == "FAILED" + assert result.succeeded is False + assert result.duration_seconds == 5.0 + assert result.run_id == "999" + assert result.error == "Task failed: invalid syntax" + + +def test_run_job_failed_with_no_message(): + from databricks.sdk.service.jobs import RunResultState + + # Failed run with no state_message uses empty string + be = CloudBackend(warehouse_id="w") + be._summary = {"resources": {"jobs": {"job": {"id": "1"}}}} + be._config = {"resources": {"jobs": {"job": {"tasks": []}}}} + be._client = SimpleNamespace( + jobs=SimpleNamespace( + run_now=lambda job_id, job_parameters=None: SimpleNamespace( + result=lambda: SimpleNamespace( + state=SimpleNamespace( + result_state=RunResultState.FAILED, + state_message=None, + ), + run_duration=0, + run_id=111, + ) + ) + ), + statement_execution=_FakeStatements(_resp(StatementState.SUCCEEDED)), + ) + + result = be.run_job("job") + assert result.succeeded is False + assert result.error == "" + + +def test_table_schema_filters_special_rows(): + # DESCRIBE TABLE returns real columns, then blank name or '#'-prefixed rows; + # only return actual columns in the schema dict. + cols = [ + _column("col_name", T.STRING), + _column("data_type", T.STRING), + _column("comment", T.STRING), + ] + first = _resp( + StatementState.SUCCEEDED, + columns=cols, + data=[ + ["id", "LONG", ""], + ["name", "STRING", ""], + ["created_at", "STRING", ""], + ["", "", ""], # blank col_name marks end of real columns + ["# Partition Information", "STRING", ""], + ], + ) + be = _backend_with(_FakeStatements(first)) + + schema = be.table_schema("main.default.users") + assert schema == {"id": "LONG", "name": "STRING", "created_at": "STRING"} + + +def test_table_schema_stops_at_hash_prefix(): + # DESCRIBE can also have '#'-prefixed row as the break indicator. + cols = [ + _column("col_name", T.STRING), + _column("data_type", T.STRING), + _column("comment", T.STRING), + ] + first = _resp( + StatementState.SUCCEEDED, + columns=cols, + data=[ + ["x", "INT", ""], + ["#Partition", "STRING", ""], # starts with '#', marks end + ["y", "INT", ""], + ], + ) + be = _backend_with(_FakeStatements(first)) + + schema = be.table_schema("main.default.t") + assert schema == {"x": "INT"} + + +def test_seed_table_creates_with_inferred_types(): + # seed_table creates CREATE OR REPLACE with inferred types from rows. + submitted = [] + + def capture_execute(query): + submitted.append(query) + return [] + + be = CloudBackend(warehouse_id="w") + be.execute_sql = capture_execute + be._ensure_schema = lambda fqn: None + + rows = [ + {"id": 1, "name": "Alice", "score": 95.5}, + {"id": 2, "name": "Bob", "score": 87.3}, + ] + be.seed_table("main.default.scores", rows) + + assert len(submitted) == 2 + create_sql = submitted[0] + assert "CREATE OR REPLACE TABLE main.default.scores" in create_sql + assert "id BIGINT" in create_sql + assert "name STRING" in create_sql + assert "score DOUBLE" in create_sql + + +def test_seed_table_inserts_with_escaped_literals(): + # seed_table inserts with properly escaped string literals. + submitted = [] + + def capture_execute(query): + submitted.append(query) + return [] + + be = CloudBackend(warehouse_id="w") + be.execute_sql = capture_execute + be._ensure_schema = lambda fqn: None + + rows = [ + {"name": "Alice"}, + {"name": "O'Brien"}, # single quote needs escaping + {"name": "Path\\to\\file"}, # backslash needs escaping + ] + be.seed_table("main.default.names", rows) + + insert_sql = submitted[1] + assert "INSERT INTO main.default.names" in insert_sql + # Spark escapes with backslash: single quote -> \', backslash -> \\ + assert "'Alice'" in insert_sql + assert "'O\\'Brien'" in insert_sql + assert "'Path\\\\to\\\\file'" in insert_sql + + +def test_seed_table_tracks_fqn_for_teardown(): + # seed_table adds fqn to be._seeded so teardown will drop it. + submitted = [] + + def capture_execute(query): + submitted.append(query) + return [] + + be = CloudBackend(warehouse_id="w") + be.execute_sql = capture_execute + be._ensure_schema = lambda fqn: None + + rows = [{"x": 1}] + be.seed_table("catalog.schema.table_one", rows) + be.seed_table("catalog.schema.table_two", rows) + + assert "catalog.schema.table_one" in be._seeded + assert "catalog.schema.table_two" in be._seeded + + +def test_teardown_does_not_raise_on_execute_sql_failure(): + # teardown is best-effort: does not raise when execute_sql fails. + be = CloudBackend() + be._seeded = {"main.default.t1", "main.default.t2"} + + def fail_execute(query): + raise RuntimeError("warehouse down") + + be.execute_sql = fail_execute + + # Should not raise, even with both tables failing to drop + be.teardown() + + +def test_teardown_does_not_raise_on_bundle_destroy_failure(): + # teardown is best-effort: does not raise when bundle destroy fails. + be = CloudBackend() + be._seeded = set() # no seeded tables + + def fail_bundle(*args): + raise RuntimeError("bundle destroy failed") + + be._bundle = fail_bundle + + # Should not raise + be.teardown() + + +def test_teardown_cleans_up_both_seeded_and_bundle(): + # teardown calls DROP TABLE on each seeded table, then bundle destroy. + dropped = [] + destroyed = [] + + def capture_execute(query): + dropped.append(query) + return [] + + def capture_bundle(*args): + destroyed.append(args) + return "" + + be = CloudBackend() + be._seeded = {"main.default.t1", "main.default.t2"} + be.execute_sql = capture_execute + be._bundle = capture_bundle + + be.teardown() + + # All seeded tables should be dropped + assert len(dropped) == 2 + assert all("DROP TABLE IF EXISTS" in q for q in dropped) + # Bundle destroy should be called + assert ("destroy", "--auto-approve") in destroyed From 1593112bb9f4faf66c597314963083c92ba37d62 Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Fri, 11 Sep 2026 09:30:37 +0000 Subject: [PATCH 08/12] experimental/bundletest: address cloud backend review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From an independent review of the diff: - read_volume_file: write the downloaded file into a temp dir and read it via a forward-slash path with an explicitly-closed connection, instead of handing a still-open NamedTemporaryFile to DuckDB — the latter fails on Windows (can't reopen an open temp file; backslash path breaks the SQL literal). Matches the local backend's approach. - seed_table: backtick-quote column identifiers so a reserved word or special char works, matching the local backend's quoting. - _ensure_job_schemas: skip a non-file sql_task and make CREATE SCHEMA best-effort, so a query/alert task or a misparsed struct reference can't abort run_job. Adds a read_volume_file unit test; kept DECIMAL->float deliberately (float-literal equality in assertions). Local suite 74 passed; cloud E2E re-verified 5 passed on azure-dogfood. Co-authored-by: Isaac --- .../src/bundletest/backends/cloud.py | 45 +++++++++++++------ .../bundletest/tests/test_cloud_backend.py | 19 ++++++-- 2 files changed, 48 insertions(+), 16 deletions(-) diff --git a/experimental/bundletest/src/bundletest/backends/cloud.py b/experimental/bundletest/src/bundletest/backends/cloud.py index b56f7a77452..385458a0efa 100644 --- a/experimental/bundletest/src/bundletest/backends/cloud.py +++ b/experimental/bundletest/src/bundletest/backends/cloud.py @@ -30,6 +30,7 @@ import json import os import re +import shutil import subprocess import tempfile import time @@ -150,10 +151,13 @@ def seed_table(self, fqn: str, rows: list[dict[str, Any]]) -> None: raise ValueError(f"cannot seed {fqn!r} with no rows") self._ensure_schema(fqn) columns = list(rows[0].keys()) - coldefs = ", ".join(f"{c} {_sql_type([r.get(c) for r in rows])}" for c in columns) + # Backtick-quote column identifiers so a reserved word (order, end) or special char + # works, matching the local backend's quoting. + coldefs = ", ".join(f"`{c}` {_sql_type([r.get(c) for r in rows])}" for c in columns) self.execute_sql(f"CREATE OR REPLACE TABLE {fqn} ({coldefs})") values = ", ".join("(" + ", ".join(_sql_literal(r.get(c)) for c in columns) + ")" for r in rows) - self.execute_sql(f"INSERT INTO {fqn} ({', '.join(columns)}) VALUES {values}") + collist = ", ".join(f"`{c}`" for c in columns) + self.execute_sql(f"INSERT INTO {fqn} ({collist}) VALUES {values}") self._seeded.add(fqn) # --- execution --- @@ -243,12 +247,22 @@ def read_volume_file(self, volume: str, filename: str) -> list[dict[str, Any]]: resp = self._ws().files.download(path) except Exception as e: raise FileNotFoundError(f"no file {filename!r} in volume {volume!r}: {e}") from e - with tempfile.NamedTemporaryFile(suffix=suffix) as tmp: - tmp.write(resp.contents.read()) - tmp.flush() - cur = duckdb.connect().execute(f"SELECT * FROM {reader}('{tmp.name}')") - cols = [d[0] for d in cur.description] - return [dict(zip(cols, row, strict=True)) for row in cur.fetchall()] + # DuckDB re-opens the file by path, so write it into a temp dir and pass a forward-slash + # path: a NamedTemporaryFile can't be reopened while open on Windows, and its backslash + # path would break the SQL string literal. + tmp_dir = tempfile.mkdtemp(prefix="bundletest-vol-") + try: + local = Path(tmp_dir) / f"data{suffix}" + local.write_bytes(resp.contents.read()) + con = duckdb.connect() + try: + cur = con.execute(f"SELECT * FROM {reader}('{local.as_posix()}')") + cols = [d[0] for d in cur.description] + return [dict(zip(cols, row, strict=True)) for row in cur.fetchall()] + finally: + con.close() + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) # --- internals --- def _ws(self) -> Any: @@ -287,14 +301,19 @@ def _ensure_schema(self, fqn: str) -> None: def _ensure_job_schemas(self, name: str) -> None: """Create the schemas a job's SQL writes to, so its unmodified statements resolve — - the cloud analogue of the local backend's namespace preparation.""" + the cloud analogue of the local backend's namespace preparation. Best-effort: a + non-file sql_task (query/alert) is skipped, and a misparsed reference (a struct field + read as catalog.schema) just fails to create — the real error, if any, surfaces on run.""" for task in self._config.get("resources", {}).get("jobs", {}).get(name, {}).get("tasks", []): - sql_task = task.get("sql_task") - if not sql_task: + file = (task.get("sql_task") or {}).get("file") + if not file: continue - sql = Path(self._bundle_path, sql_task["file"]["path"]).read_text() + sql = Path(self._bundle_path, file["path"]).read_text() for ns in _schemas_in(sql): - self.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {ns}") + try: + self.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {ns}") + except Exception: + pass def _volume_path(self, dst: str) -> str: """Resolve ``/Volumes//`` to a real UC volume path. diff --git a/experimental/bundletest/tests/test_cloud_backend.py b/experimental/bundletest/tests/test_cloud_backend.py index f060de82606..e5276259e38 100644 --- a/experimental/bundletest/tests/test_cloud_backend.py +++ b/experimental/bundletest/tests/test_cloud_backend.py @@ -6,6 +6,7 @@ literal/type rendering for seeding, namespace discovery, and volume-path resolution. """ +import io from types import SimpleNamespace import pytest @@ -156,6 +157,18 @@ def test_volume_path_resolves_resource_name(): assert be._volume_path("/Volumes/raw_data/sub/f.csv") == "/Volumes/shop/bronze/raw_data/sub/f.csv" +def test_read_volume_file_parses_downloaded_bytes(): + # Downloaded bytes are written to a temp dir and read by duckdb (not a reopened + # NamedTemporaryFile, which fails on Windows). No real workspace is touched. + be = CloudBackend() + be._summary = {"resources": {"volumes": {"raw": {"catalog_name": "c", "schema_name": "s", "name": "raw"}}}} + be._client = SimpleNamespace( + files=SimpleNamespace(download=lambda path: SimpleNamespace(contents=io.BytesIO(b"a,b\n1,x\n2,y\n"))) + ) + rows = be.read_volume_file("raw", "orders.csv") + assert rows == [{"a": 1, "b": "x"}, {"a": 2, "b": "y"}] + + def test_get_resource_keeps_inline_serialized_dashboard(): be = CloudBackend() inline = {"serialized_dashboard": {"datasets": []}, "id": "abc"} @@ -321,9 +334,9 @@ def capture_execute(query): assert len(submitted) == 2 create_sql = submitted[0] assert "CREATE OR REPLACE TABLE main.default.scores" in create_sql - assert "id BIGINT" in create_sql - assert "name STRING" in create_sql - assert "score DOUBLE" in create_sql + assert "`id` BIGINT" in create_sql + assert "`name` STRING" in create_sql + assert "`score` DOUBLE" in create_sql def test_seed_table_inserts_with_escaped_literals(): From 8e2c6c5698a11bbef09fe90969133211e4e6e64a Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Fri, 11 Sep 2026 09:54:15 +0000 Subject: [PATCH 09/12] experimental/bundletest: make the suite backend-aware for cloud runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Running the whole suite with BUNDLETEST_BACKEND=cloud errored on everything because two collections aren't cloud things: - examples/orders_bundle is the local static-config gallery (one of every resource kind, several not deployable) — it's read, never deployed. Skip its collection on cloud, the mirror of examples/cloud_orders skipping on local. The cloud fixture is examples/cloud_orders. - tests/test_assertions.py exercises the local backend's SQL/routing behavior with throwaway bundles that have no `bundle:` name; pin its bundle_env calls to backend="local" so the env var can't send them to a cloud deploy. Now `BUNDLETEST_BACKEND=cloud ... pytest` runs only the cloud-appropriate tests (40 passed); local is unchanged (74 passed, 1 skipped). Co-authored-by: Isaac --- .../bundletest/examples/orders_bundle/tests/conftest.py | 8 ++++++++ experimental/bundletest/tests/test_assertions.py | 6 +++--- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/experimental/bundletest/examples/orders_bundle/tests/conftest.py b/experimental/bundletest/examples/orders_bundle/tests/conftest.py index fdfb9da8bac..04b3e890ad1 100644 --- a/experimental/bundletest/examples/orders_bundle/tests/conftest.py +++ b/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(): diff --git a/experimental/bundletest/tests/test_assertions.py b/experimental/bundletest/tests/test_assertions.py index 169f7a7ab7b..c03f007ea38 100644 --- a/experimental/bundletest/tests/test_assertions.py +++ b/experimental/bundletest/tests/test_assertions.py @@ -70,7 +70,7 @@ def test_wrong_table_name_fails_red(tmp_path): " sql_task:\n file:\n path: job.sql\n", "CREATE OR REPLACE TABLE app.gold.out AS SELECT * FROM app.bronze.does_not_exist;", ) - with bundle_env(str(tmp_path)) as env: + with bundle_env(str(tmp_path), backend="local") as env: result = env.run_job("j") assert not result.succeeded assert "does_not_exist" in result.error @@ -82,7 +82,7 @@ def test_notebook_task_skips(tmp_path): "resources:\n jobs:\n j:\n tasks:\n - task_key: t\n" " notebook_task:\n notebook_path: /nb\n", ) - with bundle_env(str(tmp_path)) as env: + with bundle_env(str(tmp_path), backend="local") as env: with pytest.raises(LocalUnsupported): env.run_job("j") @@ -94,7 +94,7 @@ def test_databricks_only_function_skips(tmp_path): " sql_task:\n file:\n path: job.sql\n", "SELECT from_utc_timestamp(now(), 'UTC');", ) - with bundle_env(str(tmp_path)) as env: + with bundle_env(str(tmp_path), backend="local") as env: with pytest.raises(LocalUnsupported): env.run_job("j") From 31c9a8607137dff92afe8883a9f4a643014511fa Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Fri, 11 Sep 2026 11:47:01 +0000 Subject: [PATCH 10/12] experimental/bundletest: fix stale hydration wording in the cloud E2E test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_resource no longer reads the serialized form back from the workspace — `bundle summary` inlines a file_path dashboard's serialized_dashboard at config-load. Update the module docstring and rename test_dashboard_serialized_is_hydrated_from_file_path -> test_dashboard_source_tables_from_file_path so the comments describe the actual mechanism. Co-authored-by: Isaac --- .../examples/cloud_orders/tests/test_cloud_e2e.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py b/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py index 7452b5e1923..d9a485d531e 100644 --- a/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py +++ b/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py @@ -2,7 +2,7 @@ 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` -(including hydrating a file_path dashboard's serialized form), and volume upload/read. +(which inlines a file_path dashboard's serialized form), and volume upload/read. """ import pytest @@ -46,9 +46,9 @@ def test_job_is_wired_to_its_sql(env): assert job["tasks"][0]["sql_task"]["file"]["path"].endswith("transform_orders.sql") -def test_dashboard_serialized_is_hydrated_from_file_path(env): - # The dashboard is defined by file_path, so its serialized form isn't in databricks.yml; - # the cloud backend must read it back from the deployed dashboard for source_tables(). +def test_dashboard_source_tables_from_file_path(env): + # 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"] From b49e8417f69f8a556abe6624f91d6b3ece902d37 Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Fri, 11 Sep 2026 12:06:40 +0000 Subject: [PATCH 11/12] experimental/bundletest: add cloud-only get_deployed for server-state validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_resource returns the DECLARED config (bundle summary) — the shared-seam shape every resource handle depends on. It can't show what the server filled in or normalized, which is part of validating a real deployment. Add a separate, cloud-only CloudBackend.get_deployed( kind, name) that reads the resource back from the workspace via the SDK and returns the raw server object as a dict. It's additive and cloud-only (not in the Backend protocol, no lossy shape-normalization, get_resource untouched), so tests that use it must be @cloud_only. Scoped (YAGNI) to the kinds validated live against examples/cloud_orders: jobs, dashboards, volumes; other kinds raise a clear error. Live-verified on azure-dogfood: a deployed job carries server-filled fields absent from databricks.yml (settings.format=MULTI_TASK, max_concurrent_runs=1, run_as_user_name). Cloud E2E 6 passed; local 76 passed, 1 skipped. Design agreed with the base/seam owner (separate accessor over routing through get_resource). Co-authored-by: Isaac --- .../cloud_orders/tests/test_cloud_e2e.py | 12 ++++++++++++ .../src/bundletest/backends/cloud.py | 17 +++++++++++++++++ .../bundletest/tests/test_cloud_backend.py | 18 ++++++++++++++++++ 3 files changed, 47 insertions(+) diff --git a/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py b/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py index d9a485d531e..fb6795209bd 100644 --- a/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py +++ b/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py @@ -64,3 +64,15 @@ def test_uploaded_csv_is_readable(env, tmp_path): 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 diff --git a/experimental/bundletest/src/bundletest/backends/cloud.py b/experimental/bundletest/src/bundletest/backends/cloud.py index 385458a0efa..030247a9545 100644 --- a/experimental/bundletest/src/bundletest/backends/cloud.py +++ b/experimental/bundletest/src/bundletest/backends/cloud.py @@ -231,6 +231,23 @@ def get_resource(self, kind: str, name: str) -> dict[str, Any]: # serialized form source_tables() needs is already here — no workspace read needed. return self._resources()[kind][name] + def get_deployed(self, kind: str, name: str) -> dict[str, Any]: + """Read the resource back from the workspace as the server stored it — server shape, + with the values the server filled in or normalized. This differs from get_resource, + which returns the *declared* config; use this to validate what deployment actually did. + + Cloud-only by nature (there is no server locally), so it lives only on this backend and + tests that call it must be ``@cloud_only``. Returns the SDK object as a dict.""" + cfg = self.get_resource(kind, name) + if kind == "jobs": + return self._ws().jobs.get(int(cfg["id"])).as_dict() + if kind == "dashboards": + return self._ws().lakeview.get(cfg["id"]).as_dict() + if kind == "volumes": + fqn = f"{cfg['catalog_name']}.{cfg['schema_name']}.{cfg['name']}" + return self._ws().volumes.read(fqn).as_dict() + raise ValueError(f"get_deployed is not implemented for kind {kind!r} (have: jobs, dashboards, volumes)") + def put_file(self, dst: str, src: str) -> None: if not os.path.exists(src): raise FileNotFoundError(f"upload source not found: {src}") diff --git a/experimental/bundletest/tests/test_cloud_backend.py b/experimental/bundletest/tests/test_cloud_backend.py index e5276259e38..6e4cdd74b77 100644 --- a/experimental/bundletest/tests/test_cloud_backend.py +++ b/experimental/bundletest/tests/test_cloud_backend.py @@ -169,6 +169,24 @@ def test_read_volume_file_parses_downloaded_bytes(): assert rows == [{"a": 1, "b": "x"}, {"a": 2, "b": "y"}] +def test_get_deployed_dispatches_to_sdk_get(): + # get_deployed reads the server object (not declared config); the job id from summary is + # coerced to int for jobs.get. + be = CloudBackend(warehouse_id="w") + be._summary = {"resources": {"jobs": {"j": {"id": "42"}}}} + be._client = SimpleNamespace( + jobs=SimpleNamespace(get=lambda job_id: SimpleNamespace(as_dict=lambda: {"job_id": job_id})) + ) + assert be.get_deployed("jobs", "j") == {"job_id": 42} + + +def test_get_deployed_unsupported_kind_raises(): + be = CloudBackend() + be._summary = {"resources": {"clusters": {"c": {"id": "1"}}}} + with pytest.raises(ValueError, match="not implemented"): + be.get_deployed("clusters", "c") + + def test_get_resource_keeps_inline_serialized_dashboard(): be = CloudBackend() inline = {"serialized_dashboard": {"datasets": []}, "id": "abc"} From 64007864146f67fb19f0f4163d44d1d28428ba35 Mon Sep 17 00:00:00 2001 From: Sankalp-Mittal Date: Fri, 11 Sep 2026 14:06:11 +0000 Subject: [PATCH 12/12] experimental/bundletest: harden cloud backend (Codex review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. Isolation: the cloud fixture now uses a UNIQUE per-run schema (main.bundletest_cloud_) and bundle name, deployed from a per-run copy with the placeholder substituted (the .sql / .lvdash.json artifacts hardcode the schema and DABs doesn't interpolate file contents). Two concurrent runs no longer collide, and the schema is still created up front + dropped CASCADE. Tests take a `schema` fixture instead of a fixed constant. 2. Cleanup no longer masks failures: teardown logs a warning on a failed drop/`bundle destroy` (and the fixture warns on a failed schema drop) so leaked, billable resources aren't hidden — still non-raising, still doesn't mask the primary test failure. 3. Narrowed broad excepts: read_volume_file catches only databricks.sdk.errors.NotFound (so auth/permission errors surface instead of masquerading as a missing file); _ensure_job_schemas catches only the RuntimeError execute_sql raises for a statement failure (auth errors propagate). Local 76 passed, 1 skipped; ruff clean. Co-authored-by: Isaac --- .../examples/cloud_orders/tests/conftest.py | 42 +++++++++++++++---- .../cloud_orders/tests/test_cloud_e2e.py | 20 ++++----- .../src/bundletest/backends/cloud.py | 27 ++++++++---- 3 files changed, 62 insertions(+), 27 deletions(-) diff --git a/experimental/bundletest/examples/cloud_orders/tests/conftest.py b/experimental/bundletest/examples/cloud_orders/tests/conftest.py index 90243d835fd..5a1a9fbdb8e 100644 --- a/experimental/bundletest/examples/cloud_orders/tests/conftest.py +++ b/experimental/bundletest/examples/cloud_orders/tests/conftest.py @@ -1,12 +1,16 @@ """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; on the local backend there is nothing to run. The env is -module-scoped (deploy once — deploys take minutes and cost compute, unlike the local -backend's fresh per-test DuckDB). The target schema is created up front and dropped -CASCADE afterwards, so the run leaves nothing behind even if `bundle destroy` half-fails. +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 @@ -14,18 +18,39 @@ from bundletest.env import current_backend_kind, make_backend BUNDLE = str(Path(__file__).resolve().parent.parent) -SCHEMA = "main.bundletest_cloud" +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_).""" + 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(BUNDLE, backend) + e = BundleEnv(str(bundle_dir), backend) e.deploy() try: yield e @@ -33,5 +58,6 @@ def env(): e.teardown() try: backend.execute_sql(f"DROP SCHEMA IF EXISTS {SCHEMA} CASCADE") - except Exception: - pass + except Exception as exc: + warnings.warn(f"failed to drop {SCHEMA} (may be leaked): {exc}", stacklevel=1) + shutil.rmtree(tmp, ignore_errors=True) diff --git a/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py b/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py index fb6795209bd..440a6253718 100644 --- a/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py +++ b/experimental/bundletest/examples/cloud_orders/tests/test_cloud_e2e.py @@ -7,12 +7,10 @@ import pytest -SCHEMA = "main.bundletest_cloud" - -def test_bronze_to_silver_to_gold(env): +def test_bronze_to_silver_to_gold(env, schema): env.seed( - f"{SCHEMA}.raw_orders", + f"{schema}.raw_orders", [ {"order_id": 1, "total_price": 10.0}, {"order_id": 1, "total_price": 10.0}, # duplicate @@ -22,23 +20,23 @@ def test_bronze_to_silver_to_gold(env): ) assert env.run_job("transform_orders").succeeded # bronze -> silver - silver = env.table(f"{SCHEMA}.orders") + 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") + 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): - env.seed(f"{SCHEMA}.raw_orders", [{"order_id": 1, "total_price": 10.0}]) +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)" + assert env.table(f"{schema}.orders").schema["total_price"] == "decimal(10,2)" def test_job_is_wired_to_its_sql(env): @@ -46,12 +44,12 @@ def test_job_is_wired_to_its_sql(env): assert job["tasks"][0]["sql_task"]["file"]["path"].endswith("transform_orders.sql") -def test_dashboard_source_tables_from_file_path(env): +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"] + assert dashboard.source_tables() == [f"{schema}.order_summary"] def test_uploaded_csv_is_readable(env, tmp_path): diff --git a/experimental/bundletest/src/bundletest/backends/cloud.py b/experimental/bundletest/src/bundletest/backends/cloud.py index 030247a9545..eb36aed061f 100644 --- a/experimental/bundletest/src/bundletest/backends/cloud.py +++ b/experimental/bundletest/src/bundletest/backends/cloud.py @@ -28,6 +28,7 @@ from __future__ import annotations import json +import logging import os import re import shutil @@ -42,6 +43,8 @@ from bundletest.backend import RunResult +log = logging.getLogger(__name__) + # DuckDB reader per file extension, reused to parse a volume file downloaded from the # workspace (duckdb is already a base dependency, so no extra parser is pulled in). _FILE_READERS = {".csv": "read_csv", ".json": "read_json", ".parquet": "read_parquet"} @@ -134,16 +137,17 @@ def deploy(self, bundle_path: str) -> None: self._bundle("deploy") def teardown(self) -> None: - # Drop what we seeded, then destroy the bundle. Best-effort: teardown must not raise. + # Drop what we seeded, then destroy the bundle. Best-effort (must not raise), but a + # failed cleanup leaks real resources — log it rather than swallow it silently. for fqn in self._seeded: try: self.execute_sql(f"DROP TABLE IF EXISTS {fqn}") - except Exception: - pass + except Exception as e: + log.warning("teardown: failed to drop seeded table %s (may be leaked): %s", fqn, e) try: self._bundle("destroy", "--auto-approve") - except Exception: - pass + except Exception as e: + log.warning("teardown: `bundle destroy` failed (resources may be leaked): %s", e) # --- scaffolding --- def seed_table(self, fqn: str, rows: list[dict[str, Any]]) -> None: @@ -260,9 +264,13 @@ def read_volume_file(self, volume: str, filename: str) -> list[dict[str, Any]]: reader = _FILE_READERS.get(suffix) if reader is None: raise ValueError(f"cannot read {suffix!r} files") + from databricks.sdk.errors import NotFound + try: resp = self._ws().files.download(path) - except Exception as e: + except NotFound as e: + # Only a genuine missing file becomes FileNotFoundError (FileHandle.exists() keys on + # it); auth/permission errors must surface, not masquerade as "not found". raise FileNotFoundError(f"no file {filename!r} in volume {volume!r}: {e}") from e # DuckDB re-opens the file by path, so write it into a temp dir and pass a forward-slash # path: a NamedTemporaryFile can't be reopened while open on Windows, and its backslash @@ -329,8 +337,11 @@ def _ensure_job_schemas(self, name: str) -> None: for ns in _schemas_in(sql): try: self.execute_sql(f"CREATE SCHEMA IF NOT EXISTS {ns}") - except Exception: - pass + except RuntimeError as e: + # A misparsed reference (e.g. a struct field read as catalog.schema) fails + # here; ignore it and let a real error surface on run. Narrow to the + # statement failure execute_sql raises so auth/permission errors propagate. + log.debug("skipping schema prep for %s: %s", ns, e) def _volume_path(self, dst: str) -> str: """Resolve ``/Volumes//`` to a real UC volume path.