Fix #9511: Handle exceptions for failing on-run-* hooks in `source …

…freshness` (#9763)

.changes/unreleased/Fixes-20240316-231152.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Handle exceptions for failing on-run-* hooks in source freshness
+time: 2024-03-16T23:11:52.819014-07:00
+custom:
+  Author: aranke
+  Issue: "9511"

core/dbt/artifacts/schemas/freshness/v3/freshness.py

@@ -107,7 +107,11 @@ class FreshnessExecutionResultArtifact(
 
     @classmethod
     def from_result(cls, base: FreshnessResult):
-        processed = [process_freshness_result(r) for r in base.results]
+        processed = [
+            process_freshness_result(r)
+            for r in base.results
+            if isinstance(r, SourceFreshnessResult)
+        ]
         return cls(
             metadata=base.metadata,
             results=processed,

tests/functional/sources/test_source_freshness.py

@@ -430,6 +430,44 @@ def test_hooks_do_run_for_source_freshness(
         assert "on-run-end" in log_output
 
 
+class TestHooksInSourceFreshnessError:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "schema.yml": error_models_schema_yml,
+            "model.sql": error_models_model_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "config-version": 2,
+            "on-run-start": ["select fake_column from table_does_not_exist"],
+            "flags": {
+                "source_freshness_run_project_hooks": True,
+            },
+        }
+
+    def test_hooks_do_not_run_for_source_freshness(
+        self,
+        project,
+    ):
+        run_result_error = None
+
+        def run_result_error_probe(e):
+            nonlocal run_result_error
+            if (
+                e.info.name == "RunResultError"
+                and e.info.level == "error"
+                and "on-run-start" in e.info.msg
+            ):
+                run_result_error = e.info.msg
+
+        runner = dbtRunner(callbacks=[run_result_error_probe])
+        runner.invoke(["source", "freshness"])
+        assert 'relation "table_does_not_exist" does not exist' in run_result_error
+
+
 class TestHooksInSourceFreshnessDisabled(SuccessfulSourceFreshnessTest):
     @pytest.fixture(scope="class")
     def project_config_update(self):