Handle versioned models

Author: treysp

sqlmesh/dbt/manifest.py

@@ -744,7 +744,12 @@ def _test_model(node: ManifestNode) -> t.Optional[str]:
     attached_node = getattr(node, "attached_node", None)
     if attached_node:
         pieces = attached_node.split(".")
-        return pieces[-1] if pieces[0] in ["model", "seed"] else None
+        if pieces[0] in ["model", "seed"]:
+            # versioned models have format "model.package.model_name.v1" (4 parts)
+            if len(pieces) == 4:
+                return f"{pieces[2]}_{pieces[3]}"
+            return pieces[-1]
+        return None
 
     key_name = getattr(node, "file_key_name", None)
     if key_name:
@@ -824,31 +829,30 @@ def _build_test_name(node: ManifestNode, dependencies: Dependencies) -> str:
         source_parts = list(dependencies.sources)[0].split(".")
         source_name = "_".join(source_parts) if len(source_parts) == 2 else source_parts[-1]
     entity_name = model_name or source_name or ""
+    entity_name = f"_{entity_name}" if entity_name else ""
 
     name_prefix = ""
     if namespace := getattr(node.test_metadata, "namespace", None):
         name_prefix += f"{namespace}_"
     if source_name and not model_name:
         name_prefix += "source_"
 
-    name_suffix = f"_{entity_name}"
-    if column_name := getattr(node, "column_name", None):
-        name_suffix += f"_{column_name}"
-
     metadata_kwargs = node.test_metadata.kwargs
     arg_val_parts = []
     for arg, val in sorted(metadata_kwargs.items()):
-        if arg in ("model", "column_name"):
+        if arg == "model":
             continue
         if isinstance(val, dict):
             val = list(val.values())
         val = [re.sub("[^0-9a-zA-Z_]+", "_", str(v)) for v in ensure_list(val)]
         arg_val_parts.extend(val)
-    arg_vals = ("__" + "__".join(arg_val_parts)) if arg_val_parts else ""
+    unique_args = "__".join(arg_val_parts) if arg_val_parts else ""
+    unique_args = f"_{unique_args}" if unique_args else ""
 
-    auto_name = f"{name_prefix}{node.test_metadata.name}{name_suffix}{arg_vals}"
+    auto_name = f"{name_prefix}{node.test_metadata.name}{entity_name}{unique_args}"
 
     if node.name == auto_name:
         return node.name
 
-    return f"{name_prefix}{node.name}{name_suffix}{arg_vals}"
+    custom_prefix = name_prefix if source_name and not model_name else ""
+    return f"{custom_prefix}{node.name}{entity_name}{unique_args}"

tests/dbt/test_test.py

@@ -29,6 +29,7 @@ def test_tests_get_unique_names(tmp_path: Path, create_empty_project) -> None:
     # Create schema.yml with:
     # 1. Same test on model and source, both with/without custom test name
     # 2. Same test on same model with different args, both with/without custom test name
+    # 3. Versioned model with tests (both built-in and custom named)
     schema_yaml = {
         "version": 2,
         "sources": [
@@ -81,24 +82,61 @@ def test_tests_get_unique_names(tmp_path: Path, create_empty_project) -> None:
                         ],
                     },
                 ],
-            }
+            },
+            {
+                "name": "versioned_model",
+                "columns": [
+                    {
+                        "name": "id",
+                        "data_tests": [
+                            {"not_null": {}},
+                            {"not_null": {"name": "custom_versioned_notnull"}},
+                        ],
+                    },
+                    {
+                        "name": "amount",
+                        "data_tests": [
+                            {"accepted_values": {"values": ["low", "high"]}},
+                        ],
+                    },
+                ],
+                "versions": [
+                    {"v": 1},
+                    {"v": 2},
+                ],
+            },
         ],
     }
 
     schema_file = model_dir / "schema.yml"
     with open(schema_file, "w", encoding="utf-8") as f:
         yaml.dump(schema_yaml, f)
 
+    # Create versioned model files
+    versioned_model_v1_file = model_dir / "versioned_model_v1.sql"
+    with open(versioned_model_v1_file, "w", encoding="utf-8") as f:
+        f.write("SELECT 1 as id, 'low' as amount")
+
+    versioned_model_v2_file = model_dir / "versioned_model_v2.sql"
+    with open(versioned_model_v2_file, "w", encoding="utf-8") as f:
+        f.write("SELECT 1 as id, 'low' as amount")
+
     context = Context(paths=project_dir)
 
     all_audit_names = list(context._audits.keys()) + list(context._standalone_audits.keys())
     assert sorted(all_audit_names) == [
         "local.accepted_values_my_model_status__value1__value2",
         "local.accepted_values_my_model_status__value1__value2__value3",
+        "local.accepted_values_versioned_model_v1_amount__low__high",
+        "local.accepted_values_versioned_model_v2_amount__low__high",
         "local.custom_accepted_values_name_my_model_status__value1__value2",
         "local.custom_accepted_values_name_my_model_status__value1__value2__value3",
         "local.custom_notnull_name_my_model_id",
+        "local.custom_versioned_notnull_versioned_model_v1_id",
+        "local.custom_versioned_notnull_versioned_model_v2_id",
         "local.not_null_my_model_id",
+        "local.not_null_versioned_model_v1_id",
+        "local.not_null_versioned_model_v2_id",
         "local.source_custom_notnull_name_raw_my_source_id",
         "local.source_not_null_raw_my_source_id",
     ]