detect changes to fixture files with state:modified

core/dbt/contracts/graph/nodes.py

@@ -1091,14 +1091,21 @@ def tags(self) -> List[str]:
         tags = self.config.tags
         return [tags] if isinstance(tags, str) else tags
 
-    def build_unit_test_checksum(self):
+    def build_unit_test_checksum(self, project_root: str, fixture_paths: List[str]):
         # everything except 'description'
-        data = f"{self.model}-{self.given}-{self.expect}-{self.overrides}".encode("utf-8")
-        self.checksum = hashlib.new("sha256", data).hexdigest()
+        data = f"{self.model}-{self.given}-{self.expect}-{self.overrides}"
+
+        # include underlying fixture data - doesn't work with partial parsing
+        for input in self.given:
+            if input.fixture:
+                data += f"-{input.get_rows(project_root, fixture_paths)}"
+
+        self.checksum = hashlib.new("sha256", data.encode("utf-8")).hexdigest()
 
     def same_contents(self, other: Optional["UnitTestDefinition"]) -> bool:
         if other is None:
             return False
+
         return self.checksum == other.checksum
 
 

core/dbt/parser/unit_tests.py

@@ -236,7 +236,9 @@ def parse(self) -> ParseResult:
                 config=unit_test_config,
             )
             # for calculating state:modified
-            unit_test_definition.build_unit_test_checksum()
+            unit_test_definition.build_unit_test_checksum(
+                self.schema_parser.project.project_root, self.schema_parser.project.fixture_paths
+            )
             self.manifest.add_unit_test(self.yaml.file, unit_test_definition)
 
         return ParseResult()

tests/functional/unit_testing/fixtures.py

@@ -122,7 +122,7 @@
       rows:
         - {c: 2}
 
-  - name: test_my_model_empty
+  - name: test_depends_on_fixture
     model: my_model
     given:
       - input: ref('my_model_a')
@@ -155,7 +155,7 @@
       rows:
         - {macro_call: override, var_call: var_override, env_var_call: env_var_override, invocation_id: 123}
 
-  - name: test_my_model_string_concat
+  - name: test_has_string_c_ab
     model: my_model
     given:
       - input: ref('my_model_a')

tests/functional/unit_testing/test_state.py

@@ -14,6 +14,7 @@
     my_model_b_sql,
     test_my_model_simple_fixture_yml,
     test_my_model_fixture_csv,
+    test_my_model_b_fixture_csv as test_my_model_fixture_csv_modified,
 )
 
 
@@ -61,13 +62,32 @@ def test_state_modified(self, project):
         results = run_dbt(["unit-test", "--select", "state:modified", "--state", "state"])
         assert len(results) == 0
 
+        # change underlying fixture file
+        write_file(
+            test_my_model_fixture_csv_modified,
+            project.project_root,
+            "tests",
+            "fixtures",
+            "test_my_model_fixture.csv",
+        )
+        # TODO: remove --no-partial-parse as part of https://github.com/dbt-labs/dbt-core/issues/9067
+        results = run_dbt(
+            ["--no-partial-parse", "unit-test", "--select", "state:modified", "--state", "state"],
+            expect_pass=True,
+        )
+        assert len(results) == 1
+        assert results[0].node.name.endswith("test_depends_on_fixture")
+        # reset changes
+        self.copy_state(project.project_root)
+
         # change unit test definition of a single unit test
         with_changes = test_my_model_simple_fixture_yml.replace("{string_c: ab}", "{string_c: bc}")
         write_config_file(with_changes, project.project_root, "models", "test_my_model.yml")
         results = run_dbt(
             ["unit-test", "--select", "state:modified", "--state", "state"], expect_pass=False
         )
         assert len(results) == 1
+        assert results[0].node.name.endswith("test_has_string_c_ab")
 
         # change underlying model logic
         write_config_file(

tests/unit/test_unit_test_parser.py

@@ -132,7 +132,7 @@ def test_basic(self):
             fqn=["snowplow", "my_model", "test_my_model"],
             config=UnitTestConfig(),
         )
-        expected.build_unit_test_checksum()
+        expected.build_unit_test_checksum("anything", "anything")
         assertEqualNodes(unit_test, expected)
 
     def test_unit_test_config(self):