Ensure source node .freshness is equal to node's .config.freshness (#11719) (#11723)

* Ensure source node `.freshness` is equal to node's `.config.freshness`

* Default source config freshness to empty spec if no freshenss spec is given

* Update contract tests for source nodes

(cherry picked from commit 2e6d4f4)

Co-authored-by: Quigley Malcolm <[email protected]>

Author: github-actions[bot]

.changes/unreleased/Fixes-20250609-175239.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Ensure source node `.freshness` is equal to node's `.config.freshness`
+time: 2025-06-09T17:52:39.978403-05:00
+custom:
+  Author: QMalcolm
+  Issue: "11717"

core/dbt/artifacts/resources/v1/source_definition.py

@@ -20,7 +20,7 @@
 class SourceConfig(BaseConfig):
     enabled: bool = True
     event_time: Any = None
-    freshness: Optional[FreshnessThreshold] = None
+    freshness: Optional[FreshnessThreshold] = field(default_factory=FreshnessThreshold)
 
 
 @dataclass

core/dbt/parser/sources.py

@@ -206,8 +206,7 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
             loader=source.loader,
             loaded_at_field=loaded_at_field,
             loaded_at_query=loaded_at_query,
-            # The setting to an empty freshness object is to maintain what we were previously doing if no freshenss was specified
-            freshness=config.freshness or FreshnessThreshold(),
+            freshness=config.freshness,
             quoting=quoting,
             resource_type=NodeType.Source,
             fqn=target.fqn,

tests/functional/artifacts/expected_manifest.py

@@ -818,7 +818,11 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "config": {
                     "enabled": True,
                     "event_time": None,
-                    "freshness": None,
+                    "freshness": {
+                        "error_after": {"count": None, "period": None},
+                        "warn_after": {"count": None, "period": None},
+                        "filter": None,
+                    },
                 },
                 "quoting": {
                     "database": None,
@@ -1358,7 +1362,11 @@ def expected_references_manifest(project):
                 "config": {
                     "enabled": True,
                     "event_time": None,
-                    "freshness": None,
+                    "freshness": {
+                        "error_after": {"count": None, "period": None},
+                        "warn_after": {"count": None, "period": None},
+                        "filter": None,
+                    },
                 },
                 "quoting": {
                     "database": False,

tests/unit/contracts/graph/test_nodes_parsed.py

@@ -1900,6 +1900,10 @@ def basic_parsed_source_definition_dict():
         "tags": [],
         "config": {
             "enabled": True,
+            "freshness": {
+                "warn_after": {},
+                "error_after": {},
+            },
         },
         "unrendered_config": {},
         "doc_blocks": [],
@@ -1931,6 +1935,10 @@ def complex_parsed_source_definition_dict():
         "tags": ["my_tag"],
         "config": {
             "enabled": True,
+            "freshness": {
+                "warn_after": {},
+                "error_after": {},
+            },
         },
         "freshness": {"warn_after": {"period": "hour", "count": 1}, "error_after": {}},
         "loaded_at_field": "loaded_at",

tests/unit/parser/test_parser.py

@@ -453,6 +453,22 @@ def assertEqualNodes(node_one, node_two):
       - name: my_table
         loaded_at_query: "select 1 as id"
 """
+
+SOURCE_FRESHNESS_AT_TABLE_AND_CONFIG = """
+sources:
+  - name: my_source
+    loaded_at_field: test
+    tables:
+      - name: my_table
+        freshness:
+            warn_after: {count: 1, period: hour}
+            error_after: {count: 1, period: day}
+        config:
+            freshness:
+                warn_after: {count: 2, period: hour}
+                error_after: {count: 2, period: day}
+"""
+
 SOURCE_FIELD_AT_CUSTOM_FRESHNESS_BOTH_AT_TABLE = """
 sources:
   - name: my_source
@@ -476,6 +492,12 @@ def assertEqualNodes(node_one, node_two):
 class SchemaParserTest(BaseParserTest):
     def setUp(self):
         super().setUp()
+        # Reset `warn_error` to False so we don't raise warnigns about top level freshness as errors
+        set_from_args(
+            Namespace(warn_error=False, state_modified_compare_more_unrendered_values=False),
+            None,
+        )
+
         self.parser = SchemaParser(
             project=self.snowplow_project_config,
             manifest=self.manifest,
@@ -557,6 +579,19 @@ def test_parse_source_field_at_custom_freshness_both_at_table_fails(self, _):
         with self.assertRaises(ParsingError):
             self.source_patcher.parse_source(unpatched_src_default)
 
+    @mock.patch("dbt.parser.sources.get_adapter")
+    def test_parse_source_resulting_node_freshness_matches_config_freshness(self, _):
+        block = self.file_block_for(SOURCE_FRESHNESS_AT_TABLE_AND_CONFIG, "test_one.yml")
+        dct = yaml_from_file(block.file, validate=True)
+        self.parser.parse_file(block, dct)
+        unpatched_src_default = self.parser.manifest.sources["source.snowplow.my_source.my_table"]
+        src_default = self.source_patcher.parse_source(unpatched_src_default)
+        assert src_default.freshness == src_default.config.freshness
+        assert src_default.freshness.warn_after.count == 2
+        assert src_default.freshness.warn_after.period == "hour"
+        assert src_default.freshness.error_after.count == 2
+        assert src_default.freshness.error_after.period == "day"
+
     @mock.patch("dbt.parser.sources.get_adapter")
     def test_parse_source_field_at_custom_freshness_both_at_source_fails(self, _):
         block = self.file_block_for(