testing-scaffold

Testing Scaffold

Every generated app must include tests/test_data.py and tests/test_logic.py. Tests must be present even if minimal — this enforces testability from day one.

Rules

• Data layer tests mock data.WorkspaceClient — never hit Unity Catalog or the warehouse
• Logic layer tests use small hardcoded pandas DataFrames
• Always use pytest — no unittest directly
• Test function names: test_<function>_<condition>
• No Spark, no live Databricks SDK calls in any test

tests/test_data.py

"""Tests for data.py — all WorkspaceClient calls are mocked."""
import pytest
import pandas as pd
from unittest.mock import patch, MagicMock
from data import DataAccessError, load_orders


def _make_result(rows, cols, state="SUCCEEDED", error_message=None):
    """Helper: build a mock Statement Execution result."""
    result = MagicMock()
    result.status.state.value = state
    result.status.error.message = error_message
    result.manifest.schema.columns = [MagicMock(name=c) for c in cols]
    for column, name in zip(result.manifest.schema.columns, cols):
        column.name = name
    result.result.data_array = rows
    return result


@patch("data.WorkspaceClient")
def test_load_orders_returns_dataframe(mock_client_cls):
    mock_client = mock_client_cls.return_value
    mock_client.statement_execution.execute_statement.return_value = _make_result(
        rows=[["2024-01-01", "North", "100.0", "1", "1"]],
        cols=["order_date", "region", "amount", "order_id", "customer_id"],
    )

    config = {"table_name": "prod.gold.orders", "warehouse_id": "wh-1", "row_limit": 1000}
    df = load_orders(config, "2024-01-01", "2024-01-31")
    assert isinstance(df, pd.DataFrame)
    assert list(df.columns) == ["order_date", "region", "amount", "order_id", "customer_id"]
    assert len(df) == 1


@patch("data.WorkspaceClient")
def test_load_orders_raises_data_access_error_on_failure(mock_client_cls):
    mock_client = mock_client_cls.return_value
    mock_client.statement_execution.execute_statement.side_effect = Exception("Warehouse timeout")

    config = {"table_name": "prod.gold.orders", "warehouse_id": "wh-1", "row_limit": 1000}
    with pytest.raises(DataAccessError, match="unavailable"):
        load_orders(config, "2024-01-01", "2024-01-31")


@patch("data.WorkspaceClient")
def test_load_orders_raises_on_non_gold_schema(mock_client_cls):
    config = {"table_name": "prod.silver.orders", "warehouse_id": "wh-1", "row_limit": 1000}
    with pytest.raises(DataAccessError, match="Gold"):
        load_orders(config, "2024-01-01", "2024-01-31")
    mock_client_cls.return_value.statement_execution.execute_statement.assert_not_called()