Skip to content

Commit

Permalink
add test for python check on dbt asset (#21419)
Browse files Browse the repository at this point in the history
Add coverage for this case
  • Loading branch information
johannkm committed May 3, 2024
1 parent 40295ea commit 4c52b63
Showing 1 changed file with 24 additions and 1 deletion.
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
AssetsDefinition,
AssetSelection,
ExecuteInProcessResult,
asset_check,
materialize,
)
from dagster_dbt.asset_decorator import dbt_assets
Expand Down Expand Up @@ -283,6 +284,7 @@ def _materialize_dbt_assets(
selection: Optional[AssetSelection],
expected_dbt_selection: Optional[Set[str]] = None,
dagster_dbt_translator=dagster_dbt_translator_with_checks,
additional_assets: Optional[List[AssetsDefinition]] = None,
) -> ExecuteInProcessResult:
@dbt_assets(manifest=manifest, dagster_dbt_translator=dagster_dbt_translator)
def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
Expand All @@ -295,7 +297,7 @@ def my_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
yield from cli_invocation.stream()

result = materialize(
[my_dbt_assets],
[my_dbt_assets] + (additional_assets or []),
resources={"dbt": DbtCliResource(project_dir=os.fspath(test_asset_checks_path))},
selection=selection,
raise_on_error=False,
Expand Down Expand Up @@ -364,6 +366,27 @@ def test_materialize_asset_and_checks(
}


def test_materialize_asset_and_checks_with_python_check(
test_asset_checks_manifest: Dict[str, Any], dbt_commands: List[List[str]]
) -> None:
@asset_check(asset=AssetKey(["customers"]))
def my_python_check():
return AssetCheckResult(passed=True)

result = _materialize_dbt_assets(
test_asset_checks_manifest,
dbt_commands,
selection=AssetSelection.assets(AssetKey(["customers"])),
expected_dbt_selection={"fqn:test_dagster_asset_checks.customers"},
additional_assets=[my_python_check],
)
assert result.success
assert len(result.get_asset_materialization_events()) == 1
assert len(result.get_asset_check_evaluations()) == 5
assert "my_python_check" in {e.check_name for e in result.get_asset_check_evaluations()}
assert len(result.get_asset_observation_events()) == 6


def test_materialize_asset_checks_disabled(
test_asset_checks_manifest: Dict[str, Any], dbt_commands: List[List[str]]
) -> None:
Expand Down

0 comments on commit 4c52b63

Please sign in to comment.