The harness

import os

import pytest

MODE = os.environ.get("SPARK_TEST_MODE", "local")


@pytest.fixture(scope="session")
def spark():
    """Session-scoped: the JVM starts once per worker, not once per test."""
    if MODE == "connect":
        from databricks.connect import DatabricksSession

        session = DatabricksSession.builder.serverless(True).getOrCreate()
    elif MODE == "local":
        from delta import configure_spark_with_delta_pip
        from pyspark.sql import SparkSession

        builder = (
            SparkSession.builder.master("local[2]")
            .appName("dwh-tests")
            # pinned because production pins them
            .config("spark.sql.ansi.enabled", "true")
            .config("spark.sql.session.timeZone", "UTC")
            # pinned because 200 shuffle tasks over six rows is absurd
            .config("spark.sql.shuffle.partitions", "1")
            # STATIC configs — applied only when the session is created
            .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
            .config(
                "spark.sql.catalog.spark_catalog",
                "org.apache.spark.sql.delta.catalog.DeltaCatalog",
            )
            .config("spark.ui.enabled", "false")
        )
        # puts the io.delta JARs matching the installed delta-spark wheel on the
        # classpath; the two configs above name classes that are absent without it
        session = configure_spark_with_delta_pip(builder).getOrCreate()
    else:
        raise ValueError(f"SPARK_TEST_MODE must be local or connect, got {MODE!r}")

    yield session

    if MODE == "local":
        session.stop()


@pytest.fixture(scope="session")
def catalog() -> str:
    """The catalog the remote suites qualify against: dwh_test in CI, dwh_prod
    on the scheduled read-only run. Resolved lazily, so the local suite, which
    never requests it, needs no environment at all."""
    return os.environ["DWH_CATALOG"]


@pytest.fixture(scope="session")
def rates(spark):
    """silver.fx_rate, seeded once: EUR at 1, CHF at 0.94 on the factory's date."""
    from tests.factories import FX_RATE_SCHEMA, fx_rate_rows

    return spark.createDataFrame(fx_rate_rows(), FX_RATE_SCHEMA)


@pytest.fixture(autouse=True)
def _isolate(spark):
    """One session for the whole run means one test can poison the next."""
    yield
    for table in spark.catalog.listTables():
        if table.isTemporary:
            spark.catalog.dropTempView(table.name)
    spark.catalog.clearCache()

from pyspark.sql.functions import col

bad = spark.createDataFrame([("2,5",)], "quantity STRING")
bad.select(col("quantity").cast("decimal(12,3)")).collect()

# ansi.enabled = false  ->  [Row(quantity=None)]
# ansi.enabled = true   ->  [CAST_INVALID_INPUT] The value '2,5' of the type
#                           "STRING" cannot be cast to "DECIMAL(12,3)"

[tool.pytest.ini_options]
testpaths  = ["tests"]
# src layout: dwh.* and tests.* both import without an editable install
pythonpath = ["src", "."]
addopts    = "-n 4 --dist loadfile -q"
markers    = ["classic_only: known Spark Classic behaviour; expires 2026-10-01"]