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

* 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

Author: QMalcolm

.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

@@ -184,8 +184,7 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
             meta=meta,
             loader=source.loader,
             loaded_at_field=loaded_at_field,
-            # 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

@@ -792,7 +792,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,
@@ -1308,7 +1312,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

@@ -1886,6 +1886,10 @@ def basic_parsed_source_definition_dict():
         "tags": [],
         "config": {
             "enabled": True,
+            "freshness": {
+                "warn_after": {},
+                "error_after": {},
+            },
         },
         "unrendered_config": {},
     }
@@ -1916,6 +1920,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

@@ -404,10 +404,31 @@ def assertEqualNodes(node_one, node_two):
               - unique
 """
 
+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}
+"""
+
 
 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,
@@ -448,6 +469,19 @@ def test__read_basic_source(self):
         self.assertEqual(source_values[0].table.description, "")
         self.assertEqual(len(source_values[0].table.columns), 0)
 
+    @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)
+        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"
+
     def test__parse_basic_source(self):
         block = self.file_block_for(SINGLE_TABLE_SOURCE, "test_one.yml")
         dct = yaml_from_file(block.file)