Implementing Fabric Notebook-Based Unit Tests with pytest and the Fabric REST API: Validating Delta Table Outputs, Row Counts, and Schema Contracts Before Promoting Data Across Medallion Layers

Introduction

Imagine your bronze-to-silver transformation notebook finishes successfully — no errors, clean exit code, pipeline shows green. Then three days later, someone in finance notices that a unit_price column is being silently coerced to double instead of decimal(18,4), causing rounding errors in revenue aggregations that flow all the way into your Direct Lake-powered Power BI reports. The pipeline never failed. The data was wrong the whole time.

This is the failure mode that automated unit testing is specifically designed to catch. In traditional software engineering, unit tests are table stakes — you don't ship code without them. In data engineering, the equivalent discipline is data contract testing: asserting that your transformation logic produces outputs with the exact row counts, schema shapes, null rates, and referential integrity you specified before promoting data to the next layer. In Microsoft Fabric, you have a compelling opportunity to wire this up natively: notebooks run PySpark, pytest can run inside a Spark session, and the Fabric REST API lets you trigger and monitor notebook execution from an orchestration pipeline. The result is a proper test-before-promote gate.

By the end of this lesson, you'll be able to build a complete test harness that runs inside a Fabric notebook, validates Delta table outputs against schema contracts and row count thresholds, and integrates with a Data Pipeline that gates promotion from silver to gold on test passage. This is not a toy example — the patterns here are what production data platforms at scale actually need.

What you'll learn:

• How to install and configure pytest inside a Fabric Spark notebook and write assertion-based test functions against Delta tables
• How to define schema contracts as versioned artifacts and validate actual Delta table schemas against them programmatically
• How to assert row count invariants, null rate thresholds, and referential integrity constraints between medallion layers
• How to invoke a test notebook via the Fabric REST API and parse its output to make a pass/fail promotion decision inside a Data Pipeline
• How to structure your test suite so it's maintainable, debuggable, and extensible as your lakehouse schema evolves

---

Prerequisites

You should be comfortable with:

• Writing PySpark transformations in Fabric notebooks — if you need a refresher, the article on transforming data with Spark notebooks in Microsoft Fabric covers the essentials
• The medallion architecture pattern in Fabric — the lesson on implementing the medallion architecture in Microsoft Fabric explains the layer semantics this lesson builds on
• Basic Python, including decorators and exception handling
• Familiarity with the Fabric REST API at a conceptual level — you don't need to be an expert, but you should know what a workspace ID and item ID are
• At least one lakehouse set up with Delta tables in multiple layers

---

Why pytest Inside a Notebook? The Architecture Decision

Before we write a single line of test code, let's reason through the design. You have several options for testing Delta table outputs in Fabric:

1. Inline assertions in the transformation notebook itself — Simple assert df.count() > 0 calls at the bottom of a notebook. These work, but they're tangled with transformation logic, hard to run independently, and produce no structured test report.

2. A separate Great Expectations or Soda Core integration — Powerful, but these tools require external dependencies, configuration servers, and significant setup. Great for mature platforms, overkill when you're starting.

3. pytest inside a dedicated test notebook — This is our target. It gives you the assertion expressiveness of pytest, structured output you can parse, independence from transformation logic, and a clean invocation surface via the Fabric REST API.

4. Unit tests in a local Python environment against sampled data — Useful for pure transformation function logic, but misses the actual Delta table on OneLake, which is where real schema drift and data quality issues live.

The notebook-based pytest approach wins for Fabric because it runs in the same Spark environment as your production notebooks, reads the actual Delta tables from OneLake, and produces output that a Data Pipeline can interrogate. You're testing the real artifact, not a sample or a mock.

The architecture we're building looks like this: a transformation pipeline runs your bronze-to-silver or silver-to-gold notebooks, then calls a separate "gate" notebook via the Fabric REST API. The gate notebook runs pytest, captures the JSON report, and exits with a non-zero code if tests fail. The pipeline reads the exit code and either continues to the promotion step or stops and raises an alert.

---

Setting Up Your Test Notebook

Create a new notebook in your Fabric workspace. Name it something unambiguous, like test_silver_sales_contracts. The naming convention matters — when you're operating at scale with dozens of tables, you need to know at a glance which test notebook covers which layer and entity.

Installing Dependencies

Your first cell should handle installation and imports:

# Cell 1: Dependencies
%pip install pytest pytest-json-report --quiet

import pytest
import json
import sys
import os
from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType, StructField, StringType, LongType, 
    DecimalType, TimestampType, IntegerType, BooleanType
)
from delta.tables import DeltaTable
from datetime import datetime, timedelta
import tempfile
import textwrap

The pytest-json-report plugin is critical — it lets you write the test results to a JSON file that you can read back and parse after the test run, rather than relying on stdout parsing.

Configuring the Spark Session and Lakehouse Paths

# Cell 2: Configuration
spark = SparkSession.builder.getOrCreate()

# Lakehouse paths — use abfss:// for programmatic access
# Replace with your actual workspace and lakehouse GUIDs
SILVER_LAKEHOUSE = "abfss://your-workspace-guid@onelake.dfs.fabric.microsoft.com/your-silver-lakehouse-guid/Tables"
GOLD_LAKEHOUSE = "abfss://your-workspace-guid@onelake.dfs.fabric.microsoft.com/your-gold-lakehouse-guid/Tables"

# Table paths
SILVER_SALES_PATH = f"{SILVER_LAKEHOUSE}/sales_orders"
SILVER_CUSTOMERS_PATH = f"{SILVER_LAKEHOUSE}/customers"
GOLD_FACT_SALES_PATH = f"{GOLD_LAKEHOUSE}/fact_sales"

# Test run metadata
TEST_RUN_ID = datetime.now().strftime("%Y%m%d_%H%M%S")

---

Defining Schema Contracts

The most important concept in this entire lesson is the schema contract — a versioned, explicit declaration of what a Delta table's schema is supposed to look like. This is your source of truth. The actual Delta table is what you're testing against.

A schema contract is not just a list of column names. It specifies:

• Column names
• Data types (with precision and scale for decimals)
• Nullability
• Optionally: expected value ranges, allowed categories, and uniqueness constraints

# Cell 3: Schema Contract Definitions

# Contract version — increment this when schema intentionally changes
CONTRACT_VERSION = "v2.1"

# Silver sales_orders contract
SILVER_SALES_CONTRACT = StructType([
    StructField("order_id",         StringType(),       nullable=False),
    StructField("customer_id",      StringType(),       nullable=False),
    StructField("order_date",       TimestampType(),    nullable=False),
    StructField("product_sku",      StringType(),       nullable=False),
    StructField("quantity",         IntegerType(),      nullable=False),
    StructField("unit_price",       DecimalType(18, 4), nullable=False),
    StructField("total_amount",     DecimalType(18, 4), nullable=False),
    StructField("currency_code",    StringType(),       nullable=True),
    StructField("is_cancelled",     BooleanType(),      nullable=False),
    StructField("ingested_at",      TimestampType(),    nullable=False),
    StructField("source_system",    StringType(),       nullable=True),
])

# Silver customers contract
SILVER_CUSTOMERS_CONTRACT = StructType([
    StructField("customer_id",      StringType(),       nullable=False),
    StructField("customer_name",    StringType(),       nullable=False),
    StructField("email",            StringType(),       nullable=True),
    StructField("country_code",     StringType(),       nullable=True),
    StructField("segment",          StringType(),       nullable=True),
    StructField("created_at",       TimestampType(),    nullable=False),
    StructField("updated_at",       TimestampType(),    nullable=False),
    StructField("is_active",        BooleanType(),      nullable=False),
])

def normalize_schema(schema: StructType) -> dict:
    """Convert a StructType to a comparable dict representation."""
    return {
        field.name: {
            "dataType": str(field.dataType),
            "nullable": field.nullable
        }
        for field in schema.fields
    }

The normalize_schema helper is important because direct StructType equality comparison in PySpark can be finicky with type representations. Converting to a dict gives you a stable, debuggable comparison surface.

---

Writing the pytest Test Functions

Now we get to the actual test writing. The structure here follows standard pytest conventions: test functions prefixed with test_, using assert statements, organized into classes that group related assertions.

Schema Validation Tests

# Cell 4: Schema contract tests

class TestSilverSalesSchema:
    """Validate that silver.sales_orders matches the declared contract."""
    
    @classmethod
    def setup_class(cls):
        """Load the table once for all tests in this class."""
        cls.df = spark.read.format("delta").load(SILVER_SALES_PATH)
        cls.actual_schema = normalize_schema(cls.df.schema)
        cls.contract_schema = normalize_schema(SILVER_SALES_CONTRACT)
    
    def test_all_contract_columns_present(self):
        """Every column declared in the contract must exist in the table."""
        missing_columns = set(self.contract_schema.keys()) - set(self.actual_schema.keys())
        assert not missing_columns, (
            f"Columns in contract but missing from table: {missing_columns}. "
            f"Contract version: {CONTRACT_VERSION}"
        )
    
    def test_no_unexpected_columns(self):
        """Table should not contain columns not declared in the contract.
        
        Unexpected columns aren't always an error — new columns might be 
        additive — but they should be a conscious decision. Adjust this 
        assertion based on your team's schema evolution policy.
        """
        unexpected_columns = set(self.actual_schema.keys()) - set(self.contract_schema.keys())
        assert not unexpected_columns, (
            f"Columns in table but not in contract: {unexpected_columns}. "
            f"Update contract {CONTRACT_VERSION} if this is intentional."
        )
    
    def test_column_data_types_match_contract(self):
        """Each column's data type must exactly match the contract."""
        type_mismatches = {}
        for col_name, contract_def in self.contract_schema.items():
            if col_name in self.actual_schema:
                actual_type = self.actual_schema[col_name]["dataType"]
                contract_type = contract_def["dataType"]
                if actual_type != contract_type:
                    type_mismatches[col_name] = {
                        "expected": contract_type,
                        "actual": actual_type
                    }
        
        assert not type_mismatches, (
            f"Data type mismatches found:\n"
            + "\n".join(
                f"  {col}: expected {v['expected']}, got {v['actual']}"
                for col, v in type_mismatches.items()
            )
        )
    
    def test_non_nullable_columns_have_no_contract_nullability_change(self):
        """Non-nullable columns in the contract must not be marked nullable in the table."""
        nullability_violations = {}
        for col_name, contract_def in self.contract_schema.items():
            if col_name in self.actual_schema:
                contract_nullable = contract_def["nullable"]
                actual_nullable = self.actual_schema[col_name]["nullable"]
                # Contract says NOT NULL but table allows NULL — this is a violation
                if not contract_nullable and actual_nullable:
                    nullability_violations[col_name] = {
                        "contract": "NOT NULL",
                        "actual": "NULLABLE"
                    }
        
        assert not nullability_violations, (
            f"Nullability contract violations: {nullability_violations}"
        )

Row Count and Freshness Tests

Schema correctness is necessary but not sufficient. A table with the right schema and zero rows is not a passing table.

# Cell 5: Row count and freshness tests

class TestSilverSalesDataQuality:
    """Validate data quality characteristics of silver.sales_orders."""
    
    @classmethod
    def setup_class(cls):
        cls.df = spark.read.format("delta").load(SILVER_SALES_PATH)
        cls.row_count = cls.df.count()
        # Cache for multiple column-level assertions
        cls.df.cache()
    
    @classmethod  
    def teardown_class(cls):
        cls.df.unpersist()
    
    def test_table_is_not_empty(self):
        """Silver table must have at least one row."""
        assert self.row_count > 0, (
            f"silver.sales_orders is empty. Row count: {self.row_count}. "
            f"Check the bronze-to-silver transformation notebook for failures."
        )
    
    def test_row_count_above_minimum_threshold(self):
        """Row count must exceed the historical minimum baseline.
        
        Set this to the minimum acceptable number of rows based on your 
        business context — e.g., if you ingest daily and have 2 years of 
        history, you'd expect at least 500,000 rows.
        """
        MINIMUM_ROW_COUNT = 10_000
        assert self.row_count >= MINIMUM_ROW_COUNT, (
            f"silver.sales_orders has {self.row_count:,} rows, "
            f"below minimum threshold of {MINIMUM_ROW_COUNT:,}. "
            f"Possible truncation or load failure."
        )
    
    def test_no_duplicate_order_ids(self):
        """order_id must be unique across all rows."""
        from pyspark.sql.functions import count, col
        
        duplicate_count = (
            self.df
            .groupBy("order_id")
            .agg(count("*").alias("cnt"))
            .filter(col("cnt") > 1)
            .count()
        )
        
        assert duplicate_count == 0, (
            f"Found {duplicate_count:,} duplicate order_id values in "
            f"silver.sales_orders. Deduplication step may have failed."
        )
    
    def test_non_nullable_columns_contain_no_nulls(self):
        """Columns declared NOT NULL in the contract must have zero null values."""
        from pyspark.sql.functions import col, sum as spark_sum, when
        
        non_nullable_cols = [
            field.name for field in SILVER_SALES_CONTRACT.fields 
            if not field.nullable
        ]
        
        null_counts = (
            self.df
            .select([
                spark_sum(when(col(c).isNull(), 1).otherwise(0)).alias(c)
                for c in non_nullable_cols
                if c in self.df.columns
            ])
            .collect()[0]
            .asDict()
        )
        
        columns_with_nulls = {col: count for col, count in null_counts.items() if count > 0}
        
        assert not columns_with_nulls, (
            f"Non-nullable columns contain null values:\n"
            + "\n".join(f"  {col}: {cnt:,} nulls" for col, cnt in columns_with_nulls.items())
        )
    
    def test_unit_price_is_positive(self):
        """unit_price must be greater than zero for all non-cancelled orders."""
        from pyspark.sql.functions import col
        
        invalid_prices = (
            self.df
            .filter(col("is_cancelled") == False)
            .filter(col("unit_price") <= 0)
            .count()
        )
        
        assert invalid_prices == 0, (
            f"Found {invalid_prices:,} non-cancelled orders with unit_price <= 0. "
            f"Possible sign error in price transformation."
        )
    
    def test_data_freshness(self):
        """Most recent ingested_at timestamp must be within the last 25 hours.
        
        This catches the case where your pipeline ran but ingested stale data,
        or where a partial load only refreshed historical partitions.
        """
        from pyspark.sql.functions import max as spark_max
        
        max_ingested_at = (
            self.df
            .agg(spark_max("ingested_at").alias("max_ts"))
            .collect()[0]["max_ts"]
        )
        
        freshness_threshold = datetime.utcnow() - timedelta(hours=25)
        
        assert max_ingested_at is not None, "No ingested_at timestamp found — table may be empty."
        assert max_ingested_at.replace(tzinfo=None) >= freshness_threshold, (
            f"Most recent ingested_at is {max_ingested_at}, "
            f"which is older than the 25-hour freshness threshold. "
            f"Pipeline may not have run or loaded the wrong partition."
        )

Referential Integrity Tests

Cross-table referential integrity is one of the most commonly violated data contracts in a lakehouse, and one of the hardest to catch without automated testing.

# Cell 6: Referential integrity tests

class TestSilverReferentialIntegrity:
    """Validate cross-table relationships in the silver layer."""
    
    @classmethod
    def setup_class(cls):
        cls.sales_df = spark.read.format("delta").load(SILVER_SALES_PATH)
        cls.customers_df = spark.read.format("delta").load(SILVER_CUSTOMERS_PATH)
    
    def test_all_sales_customer_ids_exist_in_customers(self):
        """Every customer_id in sales_orders must have a matching row in customers."""
        from pyspark.sql.functions import col
        
        orphaned_orders = (
            self.sales_df
            .join(
                self.customers_df.select("customer_id"),
                on="customer_id",
                how="left_anti"
            )
            .count()
        )
        
        assert orphaned_orders == 0, (
            f"Found {orphaned_orders:,} sales orders referencing customer_ids "
            f"that don't exist in silver.customers. "
            f"Check the customer dimension load order in your pipeline."
        )
    
    def test_currency_code_is_valid(self):
        """currency_code must be one of the approved ISO codes."""
        from pyspark.sql.functions import col
        
        VALID_CURRENCIES = {"USD", "EUR", "GBP", "CAD", "AUD", "JPY", "CHF"}
        
        invalid_currency_count = (
            self.sales_df
            .filter(col("currency_code").isNotNull())
            .filter(~col("currency_code").isin(list(VALID_CURRENCIES)))
            .count()
        )
        
        assert invalid_currency_count == 0, (
            f"Found {invalid_currency_count:,} orders with invalid currency_code values. "
            f"Allowed values: {VALID_CURRENCIES}"
        )

---

Running pytest Programmatically and Capturing Results

This is where the implementation gets interesting. You can't just call pytest.main() from a notebook cell and expect clean, parseable output. You need to configure it to write a JSON report to a temp file, then read that file back.

# Cell 7: Test runner

def run_test_suite(test_module_path: str) -> dict:
    """
    Run pytest against the given module path and return the parsed JSON report.
    
    Returns a dict with keys: passed, failed, errors, total, failures_detail
    """
    report_path = f"/tmp/test_report_{TEST_RUN_ID}.json"
    
    exit_code = pytest.main([
        test_module_path,
        "--json-report",
        f"--json-report-file={report_path}",
        "--tb=short",        # Short traceback format
        "-v",                # Verbose output to notebook stdout
        "--no-header",
        "-p", "no:warnings"  # Suppress PySpark warnings in test output
    ])
    
    # Read the JSON report
    with open(report_path, "r") as f:
        report = json.load(f)
    
    summary = report.get("summary", {})
    
    result = {
        "exit_code": exit_code,
        "passed": summary.get("passed", 0),
        "failed": summary.get("failed", 0),
        "errors": summary.get("errors", 0),
        "total": summary.get("total", 0),
        "duration_seconds": report.get("duration", 0),
        "failures_detail": []
    }
    
    # Extract failure details for structured logging
    for test in report.get("tests", []):
        if test.get("outcome") in ("failed", "error"):
            result["failures_detail"].append({
                "test_id": test.get("nodeid"),
                "outcome": test.get("outcome"),
                "message": test.get("call", {}).get("longrepr", "No detail available")
            })
    
    return result

But there's a subtlety here: the test classes we defined earlier are in the same notebook. pytest needs to discover them as a module. The cleanest approach in a Fabric notebook is to write the test classes to a temporary Python file and then point pytest at that file.

# Cell 8: Write test module to temp file and run

TEST_MODULE_CONTENT = """
import sys
sys.path.insert(0, '/tmp')

# Re-import all test classes defined in notebook cells above
# In practice, you'd write the full class definitions here,
# or import from a shared module in your lakehouse Files area.
# For this pattern, we write inline:

import pytest
import json
from datetime import datetime, timedelta
from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import col, count, sum as spark_sum, when, max as spark_max

spark = SparkSession.builder.getOrCreate()

SILVER_LAKEHOUSE = "{silver_path}"
SILVER_SALES_PATH = SILVER_LAKEHOUSE + "/sales_orders"
SILVER_CUSTOMERS_PATH = SILVER_LAKEHOUSE + "/customers"
CONTRACT_VERSION = "{contract_version}"

# ... (paste full class definitions here)
""".format(
    silver_path=SILVER_LAKEHOUSE,
    contract_version=CONTRACT_VERSION
)

TEST_FILE_PATH = f"/tmp/test_silver_sales_{TEST_RUN_ID}.py"

with open(TEST_FILE_PATH, "w") as f:
    f.write(TEST_MODULE_CONTENT)

# Run tests
test_results = run_test_suite(TEST_FILE_PATH)
print(json.dumps(test_results, indent=2))

Making the Notebook Exit with a Meaningful Code

The last cell of your test notebook should exit with a code that tells the orchestration pipeline whether tests passed or failed:

# Cell 9: Emit result and exit

# Log summary to a Delta audit table for historical tracking
audit_record = {
    "run_id": TEST_RUN_ID,
    "test_suite": "silver_sales_contracts",
    "contract_version": CONTRACT_VERSION,
    "total_tests": test_results["total"],
    "passed": test_results["passed"],
    "failed": test_results["failed"],
    "errors": test_results["errors"],
    "duration_seconds": test_results["duration_seconds"],
    "failures_json": json.dumps(test_results["failures_detail"]),
    "run_timestamp": datetime.utcnow().isoformat()
}

audit_df = spark.createDataFrame([audit_record])
(
    audit_df
    .write
    .format("delta")
    .mode("append")
    .save(f"{SILVER_LAKEHOUSE}/_test_audit/silver_sales_test_runs")
)

# Exit with structured output for pipeline consumption
exit_payload = {
    "status": "PASS" if test_results["exit_code"] == 0 else "FAIL",
    "passed": test_results["passed"],
    "failed": test_results["failed"],
    "run_id": TEST_RUN_ID
}

mssparkutils.notebook.exit(json.dumps(exit_payload))

The mssparkutils.notebook.exit() call is what makes this notebook useful as a pipeline step — the string you pass becomes the output of the Notebook activity in your Data Pipeline, which you can then inspect with a Set Variable activity or an If Condition.

---

Triggering Tests via the Fabric REST API

The notebook-based test pattern becomes truly powerful when you can trigger it remotely and await its result. The Fabric REST API exposes a runOnDemand endpoint for notebook items that lets you do exactly this.

Why Use the REST API Directly?

You might be wondering: if we can just add a Notebook activity to a Data Pipeline, why do we need the REST API at all? Two reasons. First, the REST API gives you maximum flexibility in how you invoke, poll, and process the test notebook's result — you can do it from any Python environment, including a different notebook, an Azure Function, or a CI/CD pipeline. Second, the REST API pattern lets you implement more sophisticated orchestration, like fanning out multiple test notebooks in parallel across medallion layers.

Authentication Setup

# Fabric REST API — run from an orchestrator notebook or external service

import requests
import time

# In Fabric notebooks, use mssparkutils to get a token
# This token has the identity of the notebook's runtime principal
token = mssparkutils.credentials.getToken("https://api.fabric.microsoft.com")

FABRIC_API_BASE = "https://api.fabric.microsoft.com/v1"
WORKSPACE_ID = "your-workspace-guid"
TEST_NOTEBOOK_ITEM_ID = "your-test-notebook-item-guid"  # Get from workspace URL

headers = {
    "Authorization": f"Bearer {token}",
    "Content-Type": "application/json"
}

Submitting a Notebook Run

def submit_notebook_run(workspace_id: str, notebook_id: str, parameters: dict = None) -> str:
    """
    Submit an on-demand notebook run via the Fabric REST API.
    Returns the operation ID for polling.
    """
    url = f"{FABRIC_API_BASE}/workspaces/{workspace_id}/items/{notebook_id}/jobs/instances?jobType=RunNotebook"
    
    payload = {}
    if parameters:
        payload["executionData"] = {
            "parameters": parameters
        }
    
    response = requests.post(url, headers=headers, json=payload)
    
    if response.status_code == 202:
        # Async accepted — get operation location from header
        operation_url = response.headers.get("Location")
        print(f"Notebook run submitted. Polling: {operation_url}")
        return operation_url
    else:
        raise RuntimeError(
            f"Failed to submit notebook run. "
            f"Status: {response.status_code}, Body: {response.text}"
        )

def poll_notebook_run(operation_url: str, poll_interval_seconds: int = 15, max_wait_minutes: int = 30) -> dict:
    """
    Poll the operation URL until the notebook run completes.
    Returns the final status dict.
    """
    max_polls = (max_wait_minutes * 60) // poll_interval_seconds
    
    for attempt in range(max_polls):
        response = requests.get(operation_url, headers=headers)
        status_data = response.json()
        
        status = status_data.get("status", "Unknown")
        print(f"[Poll {attempt + 1}] Status: {status}")
        
        if status in ("Succeeded", "Failed", "Cancelled"):
            return status_data
        
        time.sleep(poll_interval_seconds)
    
    raise TimeoutError(
        f"Notebook run did not complete within {max_wait_minutes} minutes."
    )

# Execute
operation_url = submit_notebook_run(
    workspace_id=WORKSPACE_ID,
    notebook_id=TEST_NOTEBOOK_ITEM_ID,
    parameters={
        "silver_lakehouse_path": SILVER_LAKEHOUSE,
        "contract_version": CONTRACT_VERSION
    }
)

final_status = poll_notebook_run(operation_url)
print(json.dumps(final_status, indent=2))

Parsing the Exit Payload in a Pipeline

When you use a Notebook activity inside a Fabric Data Pipeline, you access the notebook's exit value via @activity('RunTestNotebook').output.result.exitValue. Set up an If Condition activity immediately after with this expression:

@equals(json(activity('RunTestNotebook').output.result.exitValue).status, 'PASS')

In the True branch, wire your promotion step (the copy or write that moves data from silver to gold). In the False branch, wire an alert notification — see the lesson on implementing end-to-end pipeline error handling in Microsoft Fabric for a complete walkthrough of this alerting pattern.

---

Delta Table Version Assertions with Time Travel

One underused testing technique in Fabric is asserting against Delta table version history. After a transformation runs, you can verify that it produced a new version and that the version's operation metadata matches what you expect.

# Cell 10: Delta version and operation metadata tests

def test_delta_table_was_recently_written(table_path: str, max_age_minutes: int = 60) -> None:
    """
    Verify that the Delta table has a commit newer than max_age_minutes.
    This catches the case where the transformation notebook succeeded
    but wrote to the wrong table path.
    """
    delta_table = DeltaTable.forPath(spark, table_path)
    history_df = delta_table.history(5)  # Last 5 operations
    
    latest_commit = history_df.orderBy("version", ascending=False).first()
    latest_timestamp = latest_commit["timestamp"]
    
    age_minutes = (datetime.utcnow() - latest_timestamp.replace(tzinfo=None)).total_seconds() / 60
    
    assert age_minutes <= max_age_minutes, (
        f"Delta table at {table_path} has not been written to in {age_minutes:.1f} minutes. "
        f"Expected a write within the last {max_age_minutes} minutes. "
        f"Latest operation: {latest_commit['operation']} at {latest_timestamp}"
    )

def test_delta_operation_was_expected_type(table_path: str, expected_operation: str) -> None:
    """
    Verify the most recent Delta operation was of the expected type.
    Common values: 'WRITE', 'MERGE', 'DELETE', 'CREATE TABLE AS SELECT'
    """
    delta_table = DeltaTable.forPath(spark, table_path)
    history_df = delta_table.history(1)
    
    latest_operation = history_df.first()["operation"]
    
    assert latest_operation == expected_operation, (
        f"Expected most recent Delta operation to be '{expected_operation}', "
        f"but got '{latest_operation}'. "
        f"Check whether the transformation notebook used the correct write mode."
    )

# Usage
test_delta_table_was_recently_written(SILVER_SALES_PATH, max_age_minutes=90)
test_delta_operation_was_expected_type(SILVER_SALES_PATH, "MERGE")

This pattern is particularly valuable for slowly changing dimension implementations — you can assert that the most recent operation was a MERGE (not an accidental OVERWRITE that would destroy your SCD history). Similarly, if you're using Delta Lake time travel in your audit processes, these version assertions give you confidence that the history chain is intact.

---

Structuring Tests Across Medallion Layers

As your platform matures, you'll want a test suite per layer boundary. Here's the recommended structure:

lakehouse/Files/tests/
├── contracts/
│   ├── v2.1/
│   │   ├── silver_sales_contract.py
│   │   ├── silver_customers_contract.py
│   │   └── gold_fact_sales_contract.py
├── test_bronze_to_silver_gate.py
├── test_silver_to_gold_gate.py
└── conftest.py   ← shared fixtures: spark session, lakehouse paths

The pipeline structure mirrors this:

Bronze Ingest Pipeline
  → [Notebook: ingest_bronze_sales]
  → [Notebook: test_bronze_to_silver_gate]  ← runs pytest
  → [If Condition: tests_passed]
      → True: [Notebook: transform_to_silver]
              → [Notebook: test_silver_to_gold_gate]
              → [If Condition: tests_passed]
                  → True: [Notebook: promote_to_gold]
                  → False: [Web Activity: alert_on_call]
      → False: [Web Activity: alert_on_call]

Each gate notebook follows the exact same structure we've built. The contracts and test logic differ, but the invocation, result parsing, and audit logging patterns are identical.

---

Hands-On Exercise

Work through this exercise to solidify the concepts. It's designed to take approximately 90 minutes for someone with Fabric experience.

Setup

Ensure you have:

• A Fabric workspace with at least one lakehouse
• A silver-layer Delta table with at least 10,000 rows (use the sales data generator below if needed)
• Fabric REST API access (you'll need your workspace ID and notebook item IDs)

Data Generator (if needed)

# Run this in a setup notebook to create test data
from pyspark.sql import Row
from pyspark.sql.types import *
import random
from datetime import datetime, timedelta

def generate_sales_data(num_rows: int = 50_000):
    rng = random.Random(42)
    rows = []
    base_date = datetime(2024, 1, 1)
    
    for i in range(num_rows):
        rows.append(Row(
            order_id=f"ORD-{i:08d}",
            customer_id=f"CUST-{rng.randint(1, 5000):05d}",
            order_date=base_date + timedelta(days=rng.randint(0, 365)),
            product_sku=f"SKU-{rng.randint(100, 999)}",
            quantity=rng.randint(1, 20),
            unit_price=round(rng.uniform(9.99, 999.99), 4),
            total_amount=0.0,  # intentionally wrong — your test should catch this
            currency_code=rng.choice(["USD", "EUR", "GBP", "INVALID"]),  # intentional bad data
            is_cancelled=rng.random() < 0.05,
            ingested_at=datetime.utcnow(),
            source_system="erp_v2"
        ))
    return rows

schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("customer_id", StringType(), False),
    StructField("order_date", TimestampType(), False),
    StructField("product_sku", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),  # intentionally wrong type
    StructField("total_amount", DoubleType(), False),
    StructField("currency_code", StringType(), True),
    StructField("is_cancelled", BooleanType(), False),
    StructField("ingested_at", TimestampType(), False),
    StructField("source_system", StringType(), True),
])

df = spark.createDataFrame(generate_sales_data(), schema=schema)
df.write.format("delta").mode("overwrite").save(f"{SILVER_LAKEHOUSE}/sales_orders")

Exercise Tasks

Task 1: Run the schema contract test suite against the generated data. Identify which specific tests fail and why. The intentional errors are: unit_price is DoubleType in the data but DecimalType(18,4) in the contract, total_amount is zero for all rows, and INVALID appears as a currency code.

Task 2: Fix one of the schema violations by modifying the generated data and re-running only the type mismatch test using pytest -k test_column_data_types_match_contract. Verify the test passes while others still fail.

Task 3: Add a new test method test_total_amount_equals_quantity_times_unit_price to TestSilverSalesDataQuality. The test should sample 1000 random rows and verify that total_amount is within 0.01 of quantity * unit_price for each.

Task 4: Wire the test notebook into a simple two-step pipeline: Step 1 runs a "transformation" notebook (can just be a placeholder that echoes success), Step 2 runs your test notebook and gates on its mssparkutils.notebook.exit() value using an If Condition. Trigger the pipeline and observe the If Condition routing.

Task 5 (stretch): Modify the audit logging cell to also write to a lakehouse table named _test_audit/test_run_history. Then write a second test class TestAuditTableIntegrity that validates that the audit table itself has been written to in the current day — confirming that your test harness is running, not just your transformations.

---

Common Mistakes & Troubleshooting

pytest Can't Find Test Classes or Functions

Symptom: pytest.main() reports 0 tests collected.

Cause: pytest discovery requires test functions to be in files named test_*.py or *_test.py, and test classes must start with Test. If you're defining test classes in a notebook cell and passing the notebook itself to pytest, discovery will fail.

Fix: Always write test classes to a .py file first (as shown in Cell 8 above), then pass that file path to pytest.main(). Never pass a notebook path.

---

Spark Session Not Available Inside pytest Test Classes

Symptom: NameError: name 'spark' is not defined inside a test method.

Cause: The spark variable is defined in the notebook's global scope but not automatically imported into the test module file you wrote to /tmp.

Fix: In your test module file, always include spark = SparkSession.builder.getOrCreate() at the module level, before any test class definitions. The getOrCreate() will return the existing active session — you're not creating a new one.

---

mssparkutils.notebook.exit() Causes pytest to Miss Cleanup

Symptom: Test audit records aren't written, cleanup code doesn't run.

Cause: mssparkutils.notebook.exit() immediately terminates the notebook. If you call it before your audit write completes, or inside a try/finally block where the finally writes audit data, the order matters.

Fix: Always do your audit logging before calling mssparkutils.notebook.exit(). Use try/except around the audit write if you want test failure to still be reported even when the audit write fails.

---

REST API Returns 403 on Notebook Run Submission

Symptom: {"errorCode": "ItemPermissionDenied"} when calling the run endpoint.

Cause: The identity running the API call (your notebook's runtime principal, a service principal, or your user identity) doesn't have Contributor or higher access to the workspace containing the test notebook.

Fix: Ensure the calling identity has at minimum Contributor role on the target workspace. Review the lesson on securing and governing Microsoft Fabric for a full breakdown of workspace roles and the permissions they grant. For service principal-based calling in CI/CD scenarios, the principal must be explicitly added to the workspace.

---

Test Run Times Out Due to Spark Session Startup

Symptom: First test in the suite takes 3-4 minutes, causing the overall pipeline to exceed its timeout.

Cause: The first spark.read.format("delta").load(...) call in a fresh Spark session triggers executor allocation, which takes time on Fabric's serverless Spark.

Fix: Add a "warm-up" cell before your pytest call that reads a tiny Delta table or runs spark.sql("SELECT 1"). This pre-allocates executors so the first real test operation is fast. Also set a realistic timeout in your pipeline's Notebook activity — 15 minutes is reasonable for a test suite that reads multiple Delta tables.

---

DecimalType Comparison Fails Even When Types Look the Same

Symptom: test_column_data_types_match_contract fails for a decimal column, but both sides show DecimalType(18,4).

Cause: PySpark's string representation of DecimalType can vary. str(DecimalType(18,4)) produces "DecimalType(18,4)" but str(field.dataType) in some contexts produces "decimal(18,4)" (lowercase, different format).

Fix: In your normalize_schema function, normalize the type string explicitly:

def normalize_schema(schema: StructType) -> dict:
    return {
        field.name: {
            "dataType": field.dataType.simpleString(),  # Use simpleString() not str()
            "nullable": field.nullable
        }
        for field in schema.fields
    }

simpleString() produces consistent output like "decimal(18,4)" regardless of context.

---

Schema Contract Tests Pass But Data Is Still Wrong After Promotion

Symptom: All tests pass, but gold layer consumers report incorrect aggregations.

Cause: Your tests validated schema and row counts, but didn't test business logic invariants. For example: total_amount = quantity * unit_price might be structurally valid (correct types, no nulls) but semantically wrong (calculated with the wrong discount logic).

Fix: Add business logic invariant tests. For every calculated field, write a test that verifies the calculation against a sample. For aggregations, write a reconciliation test that compares silver-level totals to gold-level totals. The schema evolution article covers constraint enforcement patterns that complement this testing approach.

---

Summary & Next Steps

You've built a complete, production-viable test harness that runs pytest inside a Fabric Spark notebook, validates Delta table schemas against versioned contracts, asserts row count thresholds and business logic invariants, checks referential integrity across tables, and surfaces structured pass/fail output that a Data Pipeline can use to gate promotion across medallion layers.

The key design decisions we made — and why they work — are worth restating:

• pytest in a dedicated test notebook, not inline assertions, because test logic and transformation logic should be independently runnable, independently versioned, and independently debuggable
• Schema contracts as versioned Python artifacts, because a schema change should require an intentional, reviewed code change — not just a tolerated drift that only surfaces when a report breaks
• mssparkutils.notebook.exit() with a JSON payload, because structured output is parseable by pipeline conditions in a way that log strings are not
• A persistent audit Delta table, because test history is data, and data should be queryable — you want to trend test pass rates over time and correlate failures with deployment events

Immediate next steps:

1. Implement the audit table visualization: build a Power BI report over your _test_audit/test_run_history table showing test pass rates by suite, by date, and by contract version
2. Extend your schema contracts to include value distribution assertions — for example, asserting that segment column has a distribution within 5% of its historical baseline (a sudden shift from 60% "Enterprise" to 90% "SMB" might indicate a join condition regression)
3. Integrate your test notebooks with Fabric Git integration so that contract version bumps are PR-reviewable events
4. Look at parallelizing your test suite: for large tables, consider running schema tests against Delta metadata (which doesn't read data) and data quality tests against partition-filtered samples, keeping your test gate under 5 minutes even at scale

Data quality gates are not a nice-to-have. They're the difference between a lakehouse your business trusts and one they've learned to double-check manually. The technical lift is real, but the alternative — silent data corruption propagating through your medallion layers — is far more expensive.