fix tests

Author: izeigerman

examples/multi/repo_1/models/a.sql

@@ -1,5 +1,6 @@
 MODEL (
-  name bronze.a
+  name bronze.a,
+  kind FULL
 );
 
 SELECT

examples/multi/repo_1/models/b.sql

@@ -1,5 +1,6 @@
 MODEL (
-  name bronze.b
+  name bronze.b,
+  kind FULL
 );
 
 SELECT

examples/multi/repo_2/models/c.sql

@@ -1,5 +1,6 @@
 MODEL (
-  name silver.c
+  name silver.c,
+  kind FULL
 );
 
 SELECT DISTINCT col_a

examples/multi/repo_2/models/d.sql

@@ -1,5 +1,6 @@
 MODEL (
-  name silver.d
+  name silver.d,
+  kind FULL
 );
 
 SELECT

examples/multi/repo_2/models/e.sql

@@ -1,5 +1,6 @@
 MODEL (
-  name silver.e
+  name silver.e,
+  kind FULL
 );
 
 SELECT

examples/sushi/config.py

@@ -192,3 +192,7 @@
     },
     before_all=before_all,
 )
+
+hash_md5_naming_config = config.copy(update={"physical_table_naming_convention": "hash_md5"})
+
+table_only_naming_config = config.copy(update={"physical_table_naming_convention": "table_only"})

sqlmesh/core/console.py

@@ -4074,8 +4074,8 @@ def _format_node_error(ex: NodeExecutionFailedError) -> str:
         node_name = ""
         if isinstance(error.node, SnapshotId):
             node_name = error.node.name
-        elif isinstance(error.node, tuple):
-            node_name = error.node[0]
+        elif hasattr(error.node, "snapshot_name"):
+            node_name = error.node.snapshot_name
 
         msg = _format_node_error(error)
         msg = "  " + msg.replace("\n", "\n  ")

sqlmesh/core/context_diff.py

@@ -222,7 +222,9 @@ def create(
             infer_python_dependencies=infer_python_dependencies,
         )
 
-        previous_environment_statements = state_reader.get_environment_statements(environment)
+        previous_environment_statements = (
+            state_reader.get_environment_statements(env.name) if env else []
+        )
 
         if existing_env and always_recreate_environment:
             previous_plan_id: t.Optional[str] = existing_env.plan_id
@@ -288,7 +290,7 @@ def create_no_diff(cls, environment: str, state_reader: StateReader) -> ContextD
             previous_finalized_snapshots=env.previous_finalized_snapshots,
             previous_requirements=env.requirements,
             requirements=env.requirements,
-            previous_environment_statements=[],
+            previous_environment_statements=environment_statements,
             environment_statements=environment_statements,
             previous_gateway_managed_virtual_layer=env.gateway_managed,
             gateway_managed_virtual_layer=env.gateway_managed,

sqlmesh/core/plan/builder.py

@@ -16,7 +16,7 @@
 )
 from sqlmesh.core.context_diff import ContextDiff
 from sqlmesh.core.environment import EnvironmentNamingInfo
-from sqlmesh.core.plan.common import should_force_rebuild
+from sqlmesh.core.plan.common import should_force_rebuild, is_breaking_kind_change
 from sqlmesh.core.plan.definition import (
     Plan,
     SnapshotMapping,
@@ -597,7 +597,7 @@ def _categorize_snapshots(
             forward_only = self._forward_only or self._is_forward_only_change(s_id)
             if forward_only and s_id.name in self._context_diff.modified_snapshots:
                 new, old = self._context_diff.modified_snapshots[s_id.name]
-                if should_force_rebuild(old, new) or snapshot.is_seed:
+                if is_breaking_kind_change(old, new) or snapshot.is_seed:
                     # Breaking kind changes and seed changes can't be forward-only.
                     forward_only = False
 
@@ -622,7 +622,7 @@ def _categorize_snapshot(
         if self._context_diff.directly_modified(s_id.name):
             if self._auto_categorization_enabled:
                 new, old = self._context_diff.modified_snapshots[s_id.name]
-                if should_force_rebuild(old, new):
+                if is_breaking_kind_change(old, new):
                     snapshot.categorize_as(SnapshotChangeCategory.BREAKING, False)
                     return
 
@@ -780,7 +780,7 @@ def _is_forward_only_change(self, s_id: SnapshotId) -> bool:
         if snapshot.name in self._context_diff.modified_snapshots:
             _, old = self._context_diff.modified_snapshots[snapshot.name]
             # If the model kind has changed in a breaking way, then we can't consider this to be a forward-only change.
-            if snapshot.is_model and should_force_rebuild(old, snapshot):
+            if snapshot.is_model and is_breaking_kind_change(old, snapshot):
                 return False
         return (
             snapshot.is_model and snapshot.model.forward_only and bool(snapshot.previous_versions)

sqlmesh/core/plan/common.py

@@ -4,9 +4,13 @@
 
 
 def should_force_rebuild(old: Snapshot, new: Snapshot) -> bool:
-    if new.is_view:
+    if new.is_view and new.is_indirect_non_breaking and not new.is_forward_only:
         # View models always need to be rebuilt to reflect updated upstream dependencies.
         return True
+    return is_breaking_kind_change(old, new)
+
+
+def is_breaking_kind_change(old: Snapshot, new: Snapshot) -> bool:
     if old.virtual_environment_mode != new.virtual_environment_mode:
         # If the virtual environment mode has changed, then we need to rebuild
         return True