Testing an asset

import dagster as dg
import dataframely as dy
import polars as pl

import dagster_dataframely as dd
from dagster_dataframely_demo.schema import Orders
from dagster_dataframely_demo._data import marketplace_orders, storefront_orders


from pathlib import Path

daily = dg.DailyPartitionsDefinition(start_date="2026-01-01")

Path("raw/orders").mkdir(parents=True, exist_ok=True)
storefront_orders().write_parquet("raw/orders/2026-01-02.parquet")

To test an asset, call it. Direct invocation is Dagster’s documented way to unit-test an asset. It needs no run, no IO manager and no instance. A call returns the same materializations and check results a run yields, as ordinary Python objects. The validated frame is the materialization’s value:

import pytest


@pytest.fixture
def quarantine_dir(tmp_path, monkeypatch):
    monkeypatch.setenv("DAGSTER_DATAFRAMELY_QUARANTINE_DIR", str(tmp_path))
    return tmp_path


@dd.asset(Orders, quarantine=True)
def orders(raw_orders: pl.DataFrame) -> pl.DataFrame:
    return raw_orders


def test_orders_quarantines_the_bad_lines(quarantine_dir):
    events = list(orders(dg.build_asset_context(), marketplace_orders()))
    tables = {
        event.asset_key: event.value
        for event in events
        if isinstance(event, dg.MaterializeResult)
    }
    checks = {
        event.check_name: event.passed
        for event in events
        if isinstance(event, dg.AssetCheckResult)
    }

    assert tables[dg.AssetKey(["orders"])].height == 12
    assert not checks["dy_rule__amount__min"]
    assert pl.read_parquet(quarantine_dir / "orders_quarantine.parquet").height == 8

A writer writes the quarantine, and the call does not yield it. The call yields one materialization, for the table, so the test reads the invalid rows from the parquet file, not from an event. In a run too, the IO manager writes the invalid rows, and Dagster records no materialization for the quarantine.

An asset with quarantine=True takes a context parameter, even when your function does not declare one. Pass the context first and the upstream frames after it, as Dagster does. A call has no IO manager, so it writes the invalid rows to the directory in DAGSTER_DATAFRAMELY_QUARANTINE_DIR instead. It reads that variable only when some rows are invalid.

An asset without quarantine=True needs neither the context nor the directory:

@dd.asset(Orders)
def plain_orders(raw_orders: pl.DataFrame) -> pl.DataFrame:
    return raw_orders


clean = storefront_orders()
events = list(plain_orders(clean))

The call yields a separate dg.AssetCheckResult for every check the asset declares, with its asset key set. So a call reports the same check names against the same asset keys as a run.

If the decorated function declares a context of its own, build one with dg.build_asset_context(). A partitioned root asset has no upstream inputs, so its call passes only a context with the partition key. daily is the dg.DailyPartitionsDefinition from Partitioning, declared in the setup cell above:

@dd.asset(Orders, partitions_def=daily)
def daily_orders(context: dg.AssetExecutionContext) -> pl.DataFrame:
    return pl.read_parquet(f"raw/orders/{context.partition_key}.parquet")


events = list(daily_orders(dg.build_asset_context(partition_key="2026-01-02")))

An asset that aborts raises the error from the call. Test the failure policy with pytest.raises(dd.errors.ValidationAbortError), and a frame whose columns or dtypes differ from the schema with pytest.raises(dd.errors.ColumnSchemaError).

Direct invocation does not support metadata added through the context. context.add_asset_metadata raises when you call the asset directly:

AttributeError: 'DirectAssetExecutionContext' object has no attribute '_step_execution_context'

A plain @dg.asset raises the same error, so the limit is Dagster’s, not this package’s. Return a dg.MaterializeResult instead: a call yields it with its metadata, tags and data version, so a test can read them.