create and use canonical_name for TestConfig

Author: toriwei

sqlmesh/dbt/basemodel.py

@@ -305,7 +305,7 @@ def sqlmesh_model_kwargs(
         jinja_macros.add_globals(self._model_jinja_context(model_context, dependencies))
 
         model_kwargs = {
-            "audits": [(test.name, {}) for test in self.tests],
+            "audits": [(test.canonical_name, {}) for test in self.tests],
             "column_descriptions": column_descriptions_to_sqlmesh(self.columns) or None,
             "depends_on": {
                 model.canonical_name(context) for model in model_context.refs.values()

sqlmesh/dbt/loader.py

@@ -172,8 +172,7 @@ def _load_audits(
                 for test in package.tests.values():
                     logger.debug("Converting '%s' to sqlmesh format", test.name)
                     try:
-                        audit = test.to_sqlmesh(package_context)
-                        audits[audit.name] = audit
+                        audits[test.canonical_name] = test.to_sqlmesh(package_context)
 
                     except BaseMissingReferenceError as e:
                         ref_type = "model" if isinstance(e, MissingModelError) else "source"

sqlmesh/dbt/test.py

@@ -109,6 +109,10 @@ def _validate_severity(cls, v: t.Union[Severity, str]) -> Severity:
     def _lowercase_name(cls, v: str) -> str:
         return v.lower()
 
+    @property
+    def canonical_name(self) -> str:
+        return f"{self.package_name}.{self.name}" if self.package_name else self.name
+
     @property
     def is_standalone(self) -> bool:
         # A test is standalone if:
@@ -154,14 +158,11 @@ def to_sqlmesh(self, context: DbtContext) -> Audit:
         blocking = self.severity == Severity.ERROR
 
         audit: Audit
-        audit_name = self.name
-        if self.package_name and self.package_name != context.project_name:
-            audit_name = f"{self.package_name}.{self.name}"
 
         if self.is_standalone:
             jinja_macros.add_globals({"this": self.relation_info})
             audit = StandaloneAudit(
-                name=audit_name,
+                name=self.name,
                 dbt_node_info=self.node_info,
                 dialect=self.dialect(context),
                 skip=skip,
@@ -178,7 +179,7 @@ def to_sqlmesh(self, context: DbtContext) -> Audit:
             )
         else:
             audit = ModelAudit(
-                name=audit_name,
+                name=self.name,
                 dbt_node_info=self.node_info,
                 dialect=self.dialect(context),
                 skip=skip,

tests/dbt/test_model.py

@@ -190,23 +190,23 @@ def test_manifest_filters_standalone_tests_from_models(
     # Should only have "not_null" test, not the "relationships" test
     model1_audit_names = [audit[0] for audit in model1_snapshot.model.audits]
     assert len(model1_audit_names) == 1
-    assert model1_audit_names[0] == "not_null_model1_id"
+    assert model1_audit_names[0] == "local.not_null_model1_id"
 
     # Verify model2 has its non-standalone test
     model2_audit_names = [audit[0] for audit in model2_snapshot.model.audits]
     assert len(model2_audit_names) == 1
-    assert model2_audit_names[0] == "not_null_model2_id"
+    assert model2_audit_names[0] == "local.not_null_model2_id"
 
     # Verify the standalone test (relationships) exists as a StandaloneAudit
     all_non_standalone_audits = [name for name in context._audits]
     assert sorted(all_non_standalone_audits) == [
-        "not_null_model1_id",
-        "not_null_model2_id",
+        "local.not_null_model1_id",
+        "local.not_null_model2_id",
     ]
 
     standalone_audits = [name for name in context._standalone_audits]
     assert len(standalone_audits) == 1
-    assert standalone_audits[0] == "relationships_model1_id__id__ref_model2_"
+    assert standalone_audits[0] == "local.relationships_model1_id__id__ref_model2_"
 
     plan_builder = context.plan_builder()
     dag = plan_builder._build_dag()