[dagster-dbt] Support asset checks in dbt_cloud_assets

Author: maximearmstrong

python_modules/libraries/dagster-dbt/dagster_dbt/cloud_v2/asset_decorator.py

@@ -53,4 +53,9 @@ def dbt_cloud_assets(
             exclude=exclude,
         ),
         op_tags=op_tags,
+        check_specs=workspace.load_check_specs(
+            dagster_dbt_translator=dagster_dbt_translator,
+            select=select,
+            exclude=exclude,
+        )
     )

python_modules/libraries/dagster-dbt/dagster_dbt/cloud_v2/run_handler.py

@@ -8,6 +8,7 @@
     AssetMaterialization,
     MetadataValue,
     Output,
+    AssetCheckResult,
 )
 from dagster._annotations import preview
 from dagster._record import record
@@ -195,7 +196,25 @@ def to_default_asset_events(
                     test_unique_id=unique_id,
                 )
 
-                if asset_check_key is not None:
+                if (
+                    context
+                    and has_asset_def
+                    and asset_check_key is not None
+                    and asset_check_key in context.selected_asset_check_keys
+                ):
+                    # The test is an asset check in an asset, so yield an `AssetCheckResult`.
+                    yield AssetCheckResult(
+                        passed=result_status == TestStatus.Pass,
+                        asset_key=asset_check_key.asset_key,
+                        check_name=asset_check_key.name,
+                        metadata=metadata,
+                        severity=(
+                            AssetCheckSeverity.WARN
+                            if result_status == TestStatus.Warn
+                            else AssetCheckSeverity.ERROR
+                        ),
+                    )
+                elif asset_check_key is not None:
                     yield AssetCheckEvaluation(
                         passed=result_status == TestStatus.Pass,
                         asset_key=asset_check_key.asset_key,

python_modules/libraries/dagster-dbt/dagster_dbt_tests/cloud_v2/test_resources.py

@@ -209,9 +209,13 @@ def my_dbt_cloud_assets(context: AssetExecutionContext, dbt_cloud: DbtCloudWorks
     asset_materialization_events = result.get_asset_materialization_events()
     asset_check_evaluation = result.get_asset_check_evaluations()
 
-    # materializations are successful outputs, asset check evaluations are not outputs
+    # materializations and check results are successful outputs
     outputs = [event for event in result.all_events if event.is_successful_output]
-    assert len(outputs) == 8
+    assert len(outputs) == 28
+
+    # materialization outputs have metadata, asset check outputs don't
+    outputs_with_metadata = [output for output in outputs if output.step_output_data.metadata]
+    assert len(outputs_with_metadata) == 8
 
     # 8 asset materializations
     assert len(asset_materialization_events) == 8