Repository: TobikoData/sqlmesh
Pull request #1822 (2023-12-07 22:24:39)

Feat: Detect circular references from dbt tests
In the dbt workflow, tests are run separately from transformation, so it is okay to have circular test references. In SQLMesh, tests (aka audits) are run during transformation to prevent bad data from propagating and thus the test dependencies are included in dag construction. This PR provides a clearer error message on how to fix these circular test references, so that the project can use the SQLMesh workflow.


Patch:
diff --git a/sqlmesh/dbt/basemodel.py b/sqlmesh/dbt/basemodel.py
index 7151e791..345ab343 100644
--- a/sqlmesh/dbt/basemodel.py
+++ b/sqlmesh/dbt/basemodel.py
@@ -217,12 +217,45 @@ class BaseModelConfig(GeneralConfig):
         dependencies.macros = []
         return dependencies
 
+    def check_for_circular_test_refs(self, context: DbtContext) -> None:
+        """
+        Checks for direct circular references between two models and raises an exception if found.
+        This addresses the most common circular reference seen when importing a dbt project -
+        relationship tests in both directions. In the future, we may want to increase coverage by
+        checking for indirect circular references.
+
+        Args:
+            context: The dbt context this model resides within.
+
+        Returns:
+            None
+        """
+        for test in self.tests:
+            for ref in test.dependencies.refs:
+                model = context.refs[ref]
+                if ref == self.name or ref in self.dependencies.refs:
+                    continue
+                elif self.name in model.dependencies.refs:
+                    raise ConfigError(
+                        f"Test '{test.name}' for model '{self.name}' depends on downstream model '{model.name}'."
+                        " Move the test to the downstream model to avoid circular references."
+                    )
+                elif self.name in model.tests_ref_source_dependencies.refs:
+                    circular_test = next(
+                        test.name for test in model.tests if ref in test.dependencies.refs
+                    )
+                    raise ConfigError(
+                        f"Circular reference detected between tests for models '{self.name}' and '{model.name}':"
+                        f" '{test.name}' ({self.name}), '{circular_test}' ({model.name})."
+                    )
+
     @property
     def sqlmesh_config_fields(self) -> t.Set[str]:
         return {"description", "owner", "stamp", "storage_format"}
 
     def sqlmesh_model_kwargs(self, context: DbtContext) -> t.Dict[str, t.Any]:
         """Get common sqlmesh model parameters"""
+        self.check_for_circular_test_refs(context)
         model_context = context.context_for_dependencies(
             self.dependencies.union(self.tests_ref_source_dependencies)
         )


Test patch:
diff --git a/tests/dbt/test_model.py b/tests/dbt/test_model.py
new file mode 100644
index 00000000..241cf9ad
--- /dev/null
+++ b/tests/dbt/test_model.py
@@ -0,0 +1,44 @@
+import pytest
+
+from sqlmesh.dbt.common import Dependencies
+from sqlmesh.dbt.context import DbtContext
+from sqlmesh.dbt.model import ModelConfig
+from sqlmesh.dbt.test import TestConfig
+from sqlmesh.utils.errors import ConfigError
+
+
+def test_model_test_circular_references() -> None:
+    upstream_model = ModelConfig(name="upstream")
+    downstream_model = ModelConfig(name="downstream", dependencies=Dependencies(refs={"upstream"}))
+    context = DbtContext(_refs={"upstream": upstream_model, "downstream": downstream_model})
+
+    # Test and downstream model references
+    downstream_test = TestConfig(
+        name="downstream_with_upstream",
+        sql="",
+        dependencies=Dependencies(refs={"upstream", "downstream"}),
+    )
+    upstream_test = TestConfig(
+        name="upstream_with_downstream",
+        sql="",
+        dependencies=Dependencies(refs={"upstream", "downstream"}),
+    )
+    downstream_model.tests = [downstream_test]
+    downstream_model.check_for_circular_test_refs(context)
+
+    downstream_model.tests = []
+    upstream_model.tests = [upstream_test]
+    with pytest.raises(ConfigError, match="downstream model"):
+        upstream_model.check_for_circular_test_refs(context)
+
+    downstream_model.tests = [downstream_test]
+    with pytest.raises(ConfigError, match="downstream model"):
+        upstream_model.check_for_circular_test_refs(context)
+    downstream_model.check_for_circular_test_refs(context)
+
+    # Test only references
+    downstream_model.dependencies = Dependencies()
+    with pytest.raises(ConfigError, match="between tests"):
+        upstream_model.check_for_circular_test_refs(context)
+    with pytest.raises(ConfigError, match="between tests"):
+        downstream_model.check_for_circular_test_refs(context)