chore: ignore partition_by config if dialect is snowflake

sqlmesh/dbt/model.py

@@ -567,6 +567,12 @@ def to_sqlmesh(
                     self.name,
                     "views" if isinstance(kind, ViewKind) else "ephemeral models",
                 )
+            elif context.target.dialect == "snowflake":
+                logger.warning(
+                    "Ignoring partition_by config for model '%s' targeting %s. The partition_by config is not supported for Snowflake.",
+                    self.name,
+                    context.target.dialect,
+                )
             else:
                 partitioned_by = []
                 if isinstance(self.partition_by, list):

tests/cli/test_cli.py

@@ -1,10 +1,8 @@
 import json
-import logging
 import os
 import pytest
 import string
 import time_machine
-from contextlib import contextmanager
 from os import getcwd, path, remove
 from pathlib import Path
 from shutil import rmtree
@@ -35,15 +33,6 @@ def runner() -> CliRunner:
     return CliRunner(env={"COLUMNS": "80"})
 
 
-@contextmanager
-def disable_logging():
-    logging.disable(logging.CRITICAL)
-    try:
-        yield
-    finally:
-        logging.disable(logging.NOTSET)
-
-
 def create_example_project(temp_dir, template=ProjectTemplate.DEFAULT) -> None:
     """
     Sets up CLI tests requiring a real SQLMesh project by:
@@ -795,8 +784,7 @@ def test_run_cron_not_elapsed(runner, tmp_path, caplog):
     init_prod_and_backfill(runner, tmp_path)
 
     # No error if `prod` environment exists and cron has not elapsed
-    with disable_logging():
-        result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run"])
+    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run"])
     assert result.exit_code == 0
 
     assert (
@@ -843,18 +831,17 @@ def test_table_name(runner, tmp_path):
     # Create and backfill `prod` environment
     create_example_project(tmp_path)
     init_prod_and_backfill(runner, tmp_path)
-    with disable_logging():
-        result = runner.invoke(
-            cli,
-            [
-                "--log-file-dir",
-                tmp_path,
-                "--paths",
-                tmp_path,
-                "table_name",
-                "sqlmesh_example.full_model",
-            ],
-        )
+    result = runner.invoke(
+        cli,
+        [
+            "--log-file-dir",
+            tmp_path,
+            "--paths",
+            tmp_path,
+            "table_name",
+            "sqlmesh_example.full_model",
+        ],
+    )
     assert result.exit_code == 0
     assert result.output.startswith("db.sqlmesh__sqlmesh_example.sqlmesh_example__full_model__")
 

tests/conftest.py

@@ -247,7 +247,7 @@ def rescope_duckdb_classvar(request):
     yield
 
 
-@pytest.fixture(scope="module", autouse=True)
+@pytest.fixture(scope="function", autouse=True)
 def rescope_log_handlers():
     logging.getLogger().handlers.clear()
     yield

tests/dbt/test_model.py

@@ -1,4 +1,6 @@
 import datetime
+import logging
+
 import pytest
 
 from pathlib import Path
@@ -274,7 +276,9 @@ def test_load_invalid_ref_audit_constraints(
     with open(model_schema_file, "w", encoding="utf-8") as f:
         yaml.dump(model_schema, f)
 
-    context = Context(paths=project_dir)
+    assert isinstance(get_console(), NoopConsole)
+    with caplog.at_level(logging.DEBUG):
+        context = Context(paths=project_dir)
     assert (
         "Skipping audit 'relationships_full_model_cola__cola__ref_not_real_model_' because model 'not_real_model' is not a valid ref"
         in caplog.text
@@ -539,7 +543,8 @@ def test_load_deprecated_incremental_time_column(
 
     snapshot_fqn = '"local"."main"."incremental_time_range"'
     assert isinstance(get_console(), NoopConsole)
-    context = Context(paths=project_dir)
+    with caplog.at_level(logging.DEBUG):
+        context = Context(paths=project_dir)
     model = context.snapshots[snapshot_fqn].model
     # Validate model-level attributes
     assert to_ds(model.start or "") == "2025-01-01"

tests/dbt/test_transformation.py

@@ -1885,7 +1885,7 @@ def test_parsetime_adapter_call(
 
 
 @pytest.mark.xdist_group("dbt_manifest")
-def test_partition_by(sushi_test_project: Project):
+def test_partition_by(sushi_test_project: Project, caplog):
     context = sushi_test_project.context
     context.target = BigQueryConfig(name="production", database="main", schema="sushi")
     model_config = ModelConfig(
@@ -1928,6 +1928,15 @@ def test_partition_by(sushi_test_project: Project):
     context.target = DuckDbConfig(name="target", schema="foo")
     assert model_config.to_sqlmesh(context).partitioned_by == []
 
+    context.target = SnowflakeConfig(
+        name="target", schema="test", database="test", account="foo", user="bar", password="baz"
+    )
+    assert model_config.to_sqlmesh(context).partitioned_by == []
+    assert (
+        "Ignoring partition_by config for model 'model' targeting snowflake. The partition_by config is not supported for Snowflake."
+        in caplog.text
+    )
+
     model_config = ModelConfig(
         name="model",
         alias="model",