fix(dagster): emit no check result for an audit that did not run

An end-of-build audit skipped because it reads a resource the build did
not build has passed: false in the JSON checks, so Dagster reported it
as a failed asset check, even in a green build. The Dagster integration
now emits no check result for a skipped audit, which Dagster shows as
not evaluated, and logs the audit and its skip reason instead. The JSON
shape is unchanged.

Author: kvlonge

crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/results.rs

@@ -28,6 +28,8 @@ const MATERIALIZE: &str = "materialize";
 const MATERIALIZATION: &str = "materialization";
 const CHECK: &str = "check";
 const AUDIT_CHECK_KIND: &str = "audit";
+/// The status of an audit that did not run; Dagster gets no check result for it.
+const SKIPPED_CHECK_STATUS: &str = "skipped";
 
 /// DAG nodes upstream first, or in artifact order when edges form a cycle.
 pub(crate) fn sorted_nodes<'py>(dag: &Bound<'py, PyAny>) -> PyResult<Vec<Bound<'py, PyAny>>> {
@@ -283,6 +285,9 @@ fn check_plans<'py>(
 ) -> PyResult<(Vec<Bound<'py, PyAny>>, HashSet<CheckKey>)> {
     let mut seen: HashSet<CheckKey> = seen;
     let py: Python<'py> = dag.py();
+    if get(check, "status")?.eq(SKIPPED_CHECK_STATUS)? {
+        return Ok((Vec::new(), seen));
+    }
     let dag_check: Option<Bound<'py, PyAny>> = dag_check_for(dag, check)?;
     let (asset_ids, check_name): (Vec<String>, String) = match &dag_check {
         Some(dag_check) if !check_enabled(dag_check)? => return Ok((Vec::new(), seen)),

src/sqlbuild/integrations/dagster/_helpers/invocation.py

@@ -23,6 +23,7 @@
 
 _CHECK_RESULT: str = "check"
 _ASSET_MATERIALIZATION: str = "materialization"
+_SKIPPED_CHECK_STATUS: str = "skipped"
 
 
 def _log_invocation(*, context: Any, invocation: SqlBuildCliInvocation) -> None:
@@ -42,6 +43,21 @@ def _log_invocation(*, context: Any, invocation: SqlBuildCliInvocation) -> None:
             logger.info("  %s", line)
 
 
+def _log_skipped_checks(*, context: Any, payload: Mapping[str, Any]) -> None:
+    """Log each audit that did not run and why; Dagster gets no check result for it."""
+
+    logger: Any = getattr(context, "log", None)
+    if logger is None:
+        return
+    for check in payload.get("checks", ()):
+        if check.get("status") == _SKIPPED_CHECK_STATUS:
+            logger.info(
+                "SQLBuild audit %s did not run: %s",
+                check.get("name"),
+                check.get("skip_reason") or "skipped",
+            )
+
+
 def _start_stream_future(
     *,
     executor: ThreadPoolExecutor,
@@ -178,6 +194,7 @@ def _build_results_from_execution_payload(
     command: tuple[str, ...],
     context: Any,
 ) -> tuple[Any, ...]:
+    _log_skipped_checks(context=context, payload=payload)
     return _dagster_events(
         dg=dg,
         plans=dagster_payload_results(

tests/e2e/src/sqlbuild/integrations/dagster/_test_types.py

@@ -152,3 +152,12 @@ class DagsterAuditAttachmentChecksE2ETestCase:
     description: str
     command: tuple[str, ...]
     expected_checked_assets: tuple[tuple[str, ...], ...]
+
+
+@dataclass(frozen=True)
+class DagsterSkippedAuditCheckE2ETestCase:
+    """A green build whose end audit is skipped, and the check Dagster must not report as failed."""
+
+    description: str
+    skipped_check_name: str
+    expected_log_line: str

tests/e2e/src/sqlbuild/integrations/dagster/test_audit_attachment_checks.py

@@ -19,13 +19,38 @@
 )
 from tests.e2e.src.sqlbuild.integrations.dagster._test_types import (
     DagsterAuditAttachmentChecksE2ETestCase,
+    DagsterSkippedAuditCheckE2ETestCase,
 )
 from tests.e2e.src.sqlbuild.integrations.dagster.helpers import (
     ATTACHMENT_CHECK_PROJECT_FILES,
     AuditCheckOutcome,
     audit_check_outcome,
 )
 
+_SKIPPED_AUDIT_PROJECT_FILES: dict[str, str] = {
+    "sqlbuild_project.toml": (
+        'name = "skipped_audit_checks"\nadapter = "duckdb"\n\n'
+        '[connection]\ndatabase = "warehouse.duckdb"\n\n[defaults]\nmaterialized = "table"\n'
+    ),
+    "hooks/python/skips.py": (
+        "from sqlbuild.hooks import hook\n\n\n"
+        "@hook\ndef skip_model(ctx):\n"
+        "    '''Test hook skip_model.'''\n"
+        "    return ctx.skip(reason='paused')\n"
+    ),
+    "models/customers.sql": (
+        "MODEL (description 'Test model customers.', pre_hooks [python(\"skip_model\")]);\n"
+        "SELECT 1 AS customer_id\n"
+    ),
+    "models/orders.sql": "MODEL (description 'Test model orders.');\nSELECT 1 AS order_id\n",
+    "audits/singular/orders_have_customers.sql": (
+        "AUDIT ();\n"
+        'SELECT o.order_id FROM __ref("orders") AS o\n'
+        'LEFT JOIN __ref("customers") AS c ON c.customer_id = o.order_id\n'
+        "WHERE c.customer_id IS NULL\n"
+    ),
+}
+
 
 @pytest.mark.parametrize(
     "test_case",
@@ -137,3 +162,49 @@ def sqlbuild_attachment_checks(context: AssetExecutionContext) -> Iterator[objec
     assert checks.declared <= checks.evaluated
     assert checks.all_passed is True
     assert "integration output degraded" not in capsys.readouterr().err
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        DagsterSkippedAuditCheckE2ETestCase(
+            description="an end audit over a hook-skipped first-run model is not a failed check",
+            skipped_check_name="audit__orders_have_customers",
+            expected_log_line=(
+                "SQLBuild audit orders_have_customers did not run: reads customers, which was not "
+                "built"
+            ),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_skipped_end_audit_when_building_through_dagster_then_no_failed_check_is_reported(
+    test_case: DagsterSkippedAuditCheckE2ETestCase,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    project_dir: Path = prepare_inline_project(
+        tmp_path=tmp_path,
+        project_name="skipped_audit_checks",
+        repo_files=_SKIPPED_AUDIT_PROJECT_FILES,
+    )
+    sqlbuild_project: SqlBuildProject = SqlBuildProject(
+        project_dir=project_dir, sqb_command=(str(REPO_ROOT / ".venv" / "bin" / "sqb"),)
+    )
+    sqlbuild_project.prepare()
+
+    @sqlbuild_assets(project=sqlbuild_project, required_resource_keys={"sqb"})
+    def sqlbuild_skipped_audit(context: AssetExecutionContext) -> Iterator[object]:
+        yield from context.resources.sqb.cli(["build"], context=context).stream()
+
+    result: ExecuteInProcessResult = materialize(
+        [sqlbuild_skipped_audit],
+        resources={"sqb": SqlBuildCliResource(project_dir=sqlbuild_project)},
+    )
+
+    checks: AuditCheckOutcome = audit_check_outcome(result=result, assets=sqlbuild_skipped_audit)
+    assert checks.success is True
+    assert test_case.skipped_check_name in {name for _, name in checks.declared}
+    assert test_case.skipped_check_name not in {name for _, name in checks.evaluated}
+    assert checks.all_passed is True
+    assert test_case.expected_log_line in capsys.readouterr().err