csv file fixtures

.changes/unreleased/Features-20231106-194752.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add support of csv file fixtures to unit testing
+time: 2023-11-06T19:47:52.501495-06:00
+custom:
+  Author: emmyoop
+  Issue: "8290"

core/dbt/config/project.py

@@ -652,6 +652,13 @@
             generic_test_paths.append(os.path.join(test_path, "generic"))
         return generic_test_paths
 
+    @property
+    def fixture_paths(self):
+        fixture_paths = []
+        for test_path in self.test_paths:
+            fixture_paths.append(os.path.join(test_path, "fixtures"))
+        return fixture_paths
+
     def __str__(self):
         cfg = self.to_project_config(with_packages=True)
         return str(cfg)

core/dbt/contracts/graph/unparsed.py

@@ -4,6 +4,7 @@
 from io import StringIO
 
 from dbt import deprecations
+from dbt.clients.system import find_matching
 from dbt.node_types import NodeType
 from dbt.contracts.graph.semantic_models import (
     Defaults,
@@ -784,42 +785,73 @@
         return UnitTestFormat.Dict
 
     @property
-    def rows(self) -> Union[str, List[Dict[str, Any]]]:
-        return []
+    def rows(self) -> Optional[Union[str, List[Dict[str, Any]]]]:
+        return None
+
+    @property
+    def fixture(self) -> Optional[str]:
+        return None
 
-    def get_rows(self) -> List[Dict[str, Any]]:
+    def get_rows(self, project_root: str, paths: List[str]) -> List[Dict[str, Any]]:
         if self.format == UnitTestFormat.Dict:
             assert isinstance(self.rows, List)
             return self.rows
         elif self.format == UnitTestFormat.CSV:
-            assert isinstance(self.rows, str)
-            dummy_file = StringIO(self.rows)
-            reader = csv.DictReader(dummy_file)
             rows = []
-            for row in reader:
-                rows.append(row)
+            if self.fixture is not None:
+                assert isinstance(self.fixture, str)
+                file_path = self.get_fixture_path(self.fixture, project_root, paths)
+                with open(file_path, newline="") as csvfile:
+                    reader = csv.DictReader(csvfile)
+                    for row in reader:
+                        rows.append(row)
+            else:  # using inline csv
+                assert isinstance(self.rows, str)
+                dummy_file = StringIO(self.rows)
+                reader = csv.DictReader(dummy_file)
+                rows = []
+                for row in reader:
+                    rows.append(row)
             return rows
 
+    def get_fixture_path(self, fixture: str, project_root: str, paths: List[str]) -> str:
+        fixture_path = f"{fixture}.csv"
+        matches = find_matching(project_root, paths, fixture_path)
+        if len(matches) == 0:
+            raise ParsingError(f"Could not find fixture file {fixture} for unit test")
+        elif len(matches) > 1:
+            raise ParsingError(
+                f"Found multiple fixture files named {fixture} at {[d['relative_path'] for d in matches]}. Please use a unique name for each fixture file."
+            )
+
+        return matches[0]["relative_path"]
+
     def validate_fixture(self, fixture_type, test_name) -> None:
-        if (self.format == UnitTestFormat.Dict and not isinstance(self.rows, list)) or (
-            self.format == UnitTestFormat.CSV and not isinstance(self.rows, str)
-        ):
+        if self.format == UnitTestFormat.Dict and not isinstance(self.rows, list):
             raise ParsingError(
                 f"Unit test {test_name} has {fixture_type} rows which do not match format {self.format}"
             )
+        if self.format == UnitTestFormat.CSV and not (
+            isinstance(self.rows, str) or isinstance(self.fixture, str)
+        ):
+            raise ParsingError(
+                f"Unit test {test_name} has {fixture_type} rows or fixtures which do not match format {self.format}.  Expected string."
+            )
 
 
 @dataclass
 class UnitTestInputFixture(dbtClassMixin, UnitTestFixture):
     input: str
-    rows: Union[str, List[Dict[str, Any]]] = ""
+    rows: Optional[Union[str, List[Dict[str, Any]]]] = None
     format: UnitTestFormat = UnitTestFormat.Dict
+    fixture: Optional[str] = None
 
 
 @dataclass
 class UnitTestOutputFixture(dbtClassMixin, UnitTestFixture):
-    rows: Union[str, List[Dict[str, Any]]] = ""
+    rows: Optional[Union[str, List[Dict[str, Any]]]] = None
     format: UnitTestFormat = UnitTestFormat.Dict
+    fixture: Optional[str] = None
 
 
 @dataclass

core/dbt/parser/unit_tests.py

@@ -67,7 +67,10 @@ def parse_unit_test_case(self, test_case: UnitTestDefinition):
             original_file_path=test_case.original_file_path,
             unique_id=test_case.unique_id,
             config=UnitTestNodeConfig(
-                materialized="unit", expected_rows=test_case.expect.get_rows()
+                materialized="unit",
+                expected_rows=test_case.expect.get_rows(
+                    self.root_project.project_root, self.root_project.fixture_paths
+                ),
             ),
             raw_code=tested_node.raw_code,
             database=tested_node.database,
@@ -122,7 +125,10 @@ def parse_unit_test_case(self, test_case: UnitTestDefinition):
             input_unique_id = f"model.{package_name}.{input_name}"
             input_node = ModelNode(
                 raw_code=self._build_fixture_raw_code(
-                    given.get_rows(), original_input_node_columns
+                    given.get_rows(
+                        self.root_project.project_root, self.root_project.fixture_paths
+                    ),
+                    original_input_node_columns,
                 ),
                 resource_type=NodeType.Model,
                 package_name=package_name,