# Testing an asset


Setup: the demo's `Orders` schema, which every example on this page uses

``` python
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`:


``` python
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:


``` python
@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](partitioning.md), declared in the setup cell above:


``` python
@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:

``` text
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.