@@ -78,6 +78,20 @@ def test_given_model_chain_when_running_with_different_flags_then_behaves_as_exp
78
78
} "
79
79
)
80
80
81
+ # Store initial timestamp before making changes
82
+ full_model_df_initial = (
83
+ sample_sqlmesh_test_context .query (
84
+ """
85
+ SELECT *
86
+ FROM sqlmesh_example__dev.full_model
87
+ """ ,
88
+ return_df = True ,
89
+ )
90
+ .sort_values (by = "item_id" )
91
+ .reset_index (drop = True )
92
+ )
93
+ last_updated_initial = full_model_df_initial ["last_updated_at" ].iloc [0 ]
94
+
81
95
# # Modify intermediate_model_1 sql to cause breaking change
82
96
sample_sqlmesh_test_context .modify_model_file (
83
97
"intermediate_model_1.sql" ,
@@ -145,14 +159,12 @@ def test_given_model_chain_when_running_with_different_flags_then_behaves_as_exp
145
159
.reset_index (drop = True )
146
160
)
147
161
148
- expected_full_model_df = (
149
- pd .DataFrame (
150
- {
151
- "item_id" : pd .Series ([1 , 2 , 3 ], dtype = "int32" ),
152
- "item_name" : ["item - 1" , "item - 2" , "item - 3" ],
153
- "num_orders" : [5 , 1 , 1 ],
154
- }
155
- )
162
+ expected_full_model_df = pd .DataFrame (
163
+ {
164
+ "item_id" : pd .Series ([1 , 2 , 3 ], dtype = "int32" ),
165
+ "item_name" : ["item - 1" , "item - 2" , "item - 3" ],
166
+ "num_orders" : [5 , 1 , 1 ],
167
+ }
156
168
)
157
169
158
170
print ("full_model_df" )
@@ -166,6 +178,43 @@ def test_given_model_chain_when_running_with_different_flags_then_behaves_as_exp
166
178
check_like = True ,
167
179
)
168
180
181
+ # Store the last_updated_at timestamps after the model change
182
+ last_updated_after_change = full_model_df ["last_updated_at" ].iloc [0 ]
183
+
184
+ # Verify the timestamp changed after the model modification
185
+ assert last_updated_after_change > last_updated_initial , (
186
+ "Expected last_updated_at timestamp to change after model modification"
187
+ )
188
+
189
+ # Run plan and run again with no changes
190
+ sample_sqlmesh_test_context .plan_and_run (
191
+ environment = "dev" ,
192
+ plan_options = PlanOptions (
193
+ skip_backfill = skip_backfill ,
194
+ enable_preview = True ,
195
+ skip_tests = True ,
196
+ ),
197
+ )
198
+
199
+ # Get the new timestamps
200
+ full_model_df_no_changes = (
201
+ sample_sqlmesh_test_context .query (
202
+ """
203
+ SELECT *
204
+ FROM sqlmesh_example__dev.full_model
205
+ """ ,
206
+ return_df = True ,
207
+ )
208
+ .sort_values (by = "item_id" )
209
+ .reset_index (drop = True )
210
+ )
211
+ last_updated_no_changes = full_model_df_no_changes ["last_updated_at" ].iloc [0 ]
212
+
213
+ # Verify timestamps haven't changed
214
+ assert last_updated_after_change == last_updated_no_changes , (
215
+ "Expected last_updated_at timestamps to remain unchanged when no model changes were made"
216
+ )
217
+
169
218
170
219
if __name__ == "__main__" :
171
220
pytest .main ([__file__ ])
0 commit comments