Dagster 1.3.6中Asset Job测试遇阻：UnresolvedAssetJobDefinition问题

核心解决方案：通过Definitions关联资产与Job #

你遇到的错误本质是手动构建的AssetGraph没有正确将Job的资产选择与实际资产定义关联起来。最简洁的解决方式是用Dagster的Definitions统一管理资产和Job，这样测试时可以直接获取已解析的JobDefinition，无需手动处理资产列表。

步骤1：在项目中定义Definitions #

在你的包入口文件（比如my_package/__init__.py）中，将所有资产和Job注册到Definitions：

from dagster import Definitions
from .assets import source_asset_1, source_asset_2, asset_1, asset_2
from .jobs import my_job

defs = Definitions(
    assets=[source_asset_1, source_asset_2, asset_1, asset_2],
    jobs=[my_job]
)

步骤2：修改测试代码直接从Definitions获取Job #

测试时无需手动构建AssetGraph，直接从Definitions中提取已解析完成的Job：

import pytest
from my_package import defs
from typing import Mapping

@pytest.fixture
def test_resources() -> Mapping[str, object]:
    return {
        "parquet_io_manager": parquet_io_manager.configured({'data_path': DATA_FOLDER }),
    }

def test_my_job(test_resources: Mapping[str, object]):
    # 从Definitions中获取已解析的JobDefinition
    job = defs.get_job_def("name_for_my_job")
    result = job.execute_in_process(resources=test_resources)
    assert result.success

为什么之前的方法报错？ #

你手动调用AssetGraph.from_assets()时，可能存在以下问题：

• 资产的实际AssetKey与Job选择中定义的AssetKey不匹配（比如资产名称拼写错误、带命名空间但Job选择未指定）
• 手动构建的AssetGraph没有包含Job依赖的所有上游资产（即使你列了四个资产，可能还有隐式依赖未加入）
• UnresolvedAssetJobDefinition的resolve方法需要在完整的资产上下文（即Definitions管理的上下文）中才能正确匹配资产

替代方案：从Job的Selection自动筛选资产 #

如果不想依赖Definitions，可以通过Job的selection属性提取AssetKey，再从资产集合中筛选对应资产：

import pytest
from my_package import my_job, source_asset_1, source_asset_2, asset_1, asset_2
from dagster._core.definitions.asset_graph import AssetGraph
from typing import Mapping

@pytest.fixture
def test_resources() -> Mapping[str, object]:
    return {
        "parquet_io_manager": parquet_io_manager.configured({'data_path': DATA_FOLDER }),
    }

def test_my_job(test_resources: Mapping[str, object]):
    all_assets = [source_asset_1, source_asset_2, asset_1, asset_2]
    temp_graph = AssetGraph.from_assets(all_assets)
    # 解析Job的selection，获取选中的AssetKey集合
    selected_keys = set(my_job.selection.resolve(temp_graph).keys_by_output_name.keys())
    # 根据AssetKey筛选需要的资产
    selected_assets = [asset for asset in all_assets if asset.key in selected_keys]
    
    # 构建正确的AssetGraph并解析Job
    graph = AssetGraph.from_assets(selected_assets)
    job = my_job.resolve(asset_graph=graph)
    result = job.execute_in_process(resources=test_resources)
    assert result.success

这种方法可以避免手动重复列出资产，但仍需要维护所有资产的集合，不如Definitions方案简洁。

内容的提问来源于stack exchange，提问作者guillaume latour