I spent way too long fighting with Dagster tests before I figured this out. The problem wasn’t Dagster itself - it was that I kept trying to test my data pipeline assets the wrong way.
The breakthrough came when I stopped trying to mock individual functions and started mocking entire resources. Turns out, Dagster’s resource system is perfect for testing if you use it right.
The Problem: Testing Assets That Need Real Databases
Here’s the situation. I’m building a medallion architecture data pipeline with Dagster - bronze, silver, and gold layers. The bronze layer ingests raw data from a Firebird database and writes it to Delta Lake. Pretty standard stuff!!
But how do you test an asset like this without spinning up a real Firebird database? And even if you did, how do you make the tests fast and reproducible?
@asset(
group_name="bronze_billing",
description="Raw billing data ingested from Firebird database to bronze layer",
compute_kind="delta_lake",
)
def bronze_billing_raw(
context: AssetExecutionContext,
firebird_resource: MockFirebirdParquetResource,
delta_lake: DeltaLakeResource,
) -> dict:
"""Ingest raw billing data from Firebird to bronze Delta Lake."""
context.log.info("Starting bronze billing data ingestion")
# Read raw billing data from source
df = firebird_resource.read_billing_data()
context.log.info(f"Read {len(df)} rows from source")
# Add ingestion metadata columns
df = df.with_columns([
pl.lit("firebird").alias("_source_system"),
pl.lit(context.run_id).alias("_dagster_run_id"),
])
# Write to Delta Lake bronze layer
table_path = delta_lake.write_delta(
df=df,
table_name="billing_raw",
layer="bronze",
mode="overwrite",
)
return {
"row_count": len(df),
"columns": df.columns,
"table_path": table_path,
}
This asset depends on two resources: firebird_resource and delta_lake. The first attempt at testing this was a mess of mocks and patches. It didn’t work.
The Key Insight: Resources Are Your Test Seam
The trick is to recognize that Dagster resources are already designed to be swappable. You don’t need to mock the asset function at all - you just need to provide different resource implementations for testing.
I created a MockFirebirdParquetResource that reads from parquet files instead of connecting to a real database:
class MockFirebirdParquetResource(ConfigurableResource):
"""
Mock Firebird database resource that reads from parquet files.
Used for development and testing when a live Firebird connection
is not available.
"""
parquet_path: str
def _read_parquet(self, filename: str) -> pl.DataFrame:
file_path = Path(self.parquet_path) / f"{filename}.parquet"
return pl.read_parquet(file_path)
def read_billing_data(self) -> pl.DataFrame:
"""Read mock billing data from parquet file."""
return self._read_parquet("billing_raw")
def read_bill_items_data(self) -> pl.DataFrame:
"""Read mock bill items data from parquet file."""
return self._read_parquet("bill_items_raw")
The beautiful part? This mock resource has the exact same interface as the real FirebirdResource (which I’ll build later). My asset code doesn’t need to know or care which one it’s using.
The Pytest Fixture That Saved Everything
Once I had the mock resource pattern figured out, the tests became straightforward. I use pytest fixtures to set up temporary directories with test data:
@pytest.fixture
def temp_parquet_path():
"""Create temporary parquet files for testing."""
temp_dir = tempfile.mkdtemp()
# Create mock billing data
billing_df = pl.DataFrame({
"BILL_ID": [1, 2, 3],
"PATIENT_NR": ["P001", "P002", "P003"],
"AMOUNT": [100.0, 200.0, 150.0],
"BILL_DATE": ["2024-01-01", "2024-01-02", "2024-01-03"],
})
billing_df.write_parquet(Path(temp_dir) / "billing_raw.parquet")
# Create mock bill items data
bill_items_df = pl.DataFrame({
"ITEM_ID": [1, 2, 3, 4],
"BILL_ID": [1, 1, 2, 3],
"DESCRIPTION": ["Item A", "Item B", "Item C", "Item D"],
"ITEM_AMOUNT": [50.0, 50.0, 200.0, 150.0],
})
bill_items_df.write_parquet(Path(temp_dir) / "bill_items_raw.parquet")
yield temp_dir
shutil.rmtree(temp_dir)
@pytest.fixture
def temp_delta_path():
"""Create a temporary directory for Delta Lake tables."""
temp_dir = tempfile.mkdtemp()
yield temp_dir
shutil.rmtree(temp_dir)
These fixtures create real parquet files and real Delta Lake directories - but they’re temporary and isolated. Each test gets its own clean slate.
Mocking the Dagster Context
The last piece of the puzzle was mocking Dagster’s AssetExecutionContext. I don’t need the full complexity of Dagster’s context - just enough to satisfy the asset’s interface:
class MockContext:
"""Mock Dagster context for testing."""
def __init__(self):
self.run_id = "test-run-123"
self._metadata = {}
self._logs = []
@property
def log(self):
return self
def info(self, msg):
self._logs.append(("info", msg))
def add_output_metadata(self, metadata):
self._metadata.update(metadata)
This is way simpler than trying to instantiate a real Dagster context, and it gives me full control over the test environment.
The Complete Test Pattern
Here’s what a complete test looks like with this pattern:
def test_bronze_billing_raw_writes_to_delta(temp_parquet_path, temp_delta_path):
"""Test that bronze_billing_raw writes data to Delta Lake."""
context = MockContext()
firebird_resource = MockFirebirdParquetResource(parquet_path=temp_parquet_path)
delta_lake = DeltaLakeResource(base_path=temp_delta_path)
result = bronze_billing_raw(
context=context,
firebird_resource=firebird_resource,
delta_lake=delta_lake,
)
assert result["row_count"] == 3
assert "BILL_ID" in result["columns"]
assert "_source_system" in result["columns"]
assert "_dagster_run_id" in result["columns"]
# Verify Delta table was created
assert delta_lake.table_exists("billing_raw", "bronze")
# Read back and verify
read_df = delta_lake.read_delta("billing_raw", "bronze")
assert len(read_df) == 3
assert read_df["_source_system"][0] == "firebird"
No mocking libraries. No patching. No brittle tests that break when you refactor. Just direct calls to your asset functions with test resources.
The test verifies the full end-to-end behavior - reading from a source, transforming the data, writing to Delta Lake, and being able to read it back. That’s way more valuable than testing individual functions in isolation.
Why This Pattern Works
The mock resource pattern works because it aligns with how Dagster already thinks about dependencies. Resources are meant to be configurable and swappable. By creating mock resources that implement the same interface as production resources, you get:
- Fast tests - No network calls, no real databases
- Isolated tests - Each test gets fresh temp directories
- Real behavior - Testing actual Delta Lake writes, not mocks
- Maintainable tests - Changes to asset logic don’t break tests
- Development mode - The same mock resources work in
make dev
The key insight was recognizing that the resource abstraction is your test seam. Don’t fight it by mocking at the function level. Embrace it by providing different resource implementations.
Now when I add a new asset, the testing pattern is clear: create fixtures for test data, instantiate the mock resources, call the asset function, and verify the output. It’s predictable, it’s reliable, and it actually tests what matters.