refactor(sql): update full model SQL and enhance test cases

- Change SQL model definition to use lowercase for `kind` and uppercase for `NAME`.
- Add `item_name` to the SELECT statement and include `last_updated_at` in the output.
- Update test configuration to exclude `last_updated_at` and mark the output as partial.
- Improve test assertions by comparing the full model DataFrame against an expected DataFrame.
- Ensure consistency in test context usage and enhance clarity in test logic.

Author: KLonge

sample/sqlmesh_project/models/marts/full_model.sql

@@ -1,6 +1,6 @@
-MODEL (
-  name sqlmesh_example.full_model,
-  kind FULL,
+model (
+  NAME sqlmesh_example.full_model,
+  kind full,
   cron '@daily',
   grain item_id,
   audits (assert_positive_order_ids),
@@ -9,10 +9,14 @@ MODEL (
     "full",
   )
 );
-
 SELECT
   item_id,
-  COUNT(DISTINCT id) AS num_orders,
+  item_name,
+  COUNT(
+    DISTINCT id
+  ) AS num_orders,
+  now() AS last_updated_at
 FROM
   sqlmesh_example.intermediate_model_1
-GROUP BY ALL
+GROUP BY
+  ALL

sample/sqlmesh_project/tests/test_full_model.yaml

@@ -1,5 +1,6 @@
 test_example_full_model:
   model: sqlmesh_example.full_model
+  exclude_columns: [last_updated_at]
   inputs:
     sqlmesh_example.intermediate_model_1:
       rows:
@@ -11,6 +12,7 @@ test_example_full_model:
         item_id: 2
   outputs:
     query:
+      partial: true
       rows:
       - item_id: 1
         num_orders: 2

tests/context/plan_and_run/test_model_code_change.py

@@ -1,6 +1,8 @@
 import logging
 
+import pandas as pd
 import pytest
+from pandas.testing import assert_frame_equal
 
 from dagster_sqlmesh.controller.base import PlanOptions
 from tests.conftest import SQLMeshTestContext
@@ -59,7 +61,6 @@ def test_given_model_chain_when_running_with_different_flags_then_behaves_as_exp
         environment="dev",
     )
 
-
     print(
         f"intermediate_model_1 first run: {
             sample_sqlmesh_test_context.query(
@@ -95,23 +96,22 @@ def test_given_model_chain_when_running_with_different_flags_then_behaves_as_exp
         main.id,
         main.item_id,
         main.event_date,
-        CONCAT(sub.item_name, ' - modified1') as item_name
+        CONCAT('item - ', main.item_id) as item_name
         FROM sqlmesh_example.staging_model_1 AS main
         INNER JOIN sqlmesh_example.staging_model_2 as sub
         ON main.id = sub.id
         WHERE
         event_date BETWEEN @start_date AND @end_date
-
         """,
     )
 
-
     # Run with specified flags
     sample_sqlmesh_test_context.plan_and_run(
         environment="dev",
         plan_options=PlanOptions(
             skip_backfill=skip_backfill,
             enable_preview=True,
+            skip_tests=True,
         ),
     )
 
@@ -133,8 +133,40 @@ def test_given_model_chain_when_running_with_different_flags_then_behaves_as_exp
         }"
     )
 
-    raise Exception("Stop here")
+    full_model_df = (
+        sample_sqlmesh_test_context.query(
+            """
+        SELECT *
+        FROM sqlmesh_example__dev.full_model
+        """,
+            return_df=True,
+        )
+        .sort_values(by="item_id")
+        .reset_index(drop=True)
+    )
+
+    expected_full_model_df = (
+        pd.DataFrame(
+            {
+                "item_id": pd.Series([1, 2, 3], dtype="int32"),
+                "item_name": ["item - 1", "item - 2", "item - 3"],
+                "num_orders": [5, 1, 1],
+            }
+        )
+        .sort_values(by="item_id")
+        .reset_index(drop=True)
+    )
 
+    print("full_model_df")
+    print(full_model_df.drop(columns=["last_updated_at"]))
+    print("expected_full_model_df")
+    print(expected_full_model_df)
+
+    assert_frame_equal(
+        full_model_df.drop(columns=["last_updated_at"]),
+        expected_full_model_df,
+        check_like=True,
+    )
 
 
 if __name__ == "__main__":