Add SQL model with python macro on new test

Author: VaggelisD

sqlmesh/core/test/definition.py

@@ -544,13 +544,9 @@ def _concurrent_render_context(self) -> t.Iterator[None]:
         with lock_ctx, time_ctx, dialect_patch_ctx:
             yield
 
-    def _execute(self, query: exp.Query) -> pd.DataFrame:
+    def _execute(self, query: exp.Query | str) -> pd.DataFrame:
         """Executes the given query using the testing engine adapter and returns a DataFrame."""
-
-        with self._concurrent_render_context():
-            sql = query.sql(self._test_adapter_dialect, pretty=self.engine_adapter._pretty_sql)
-
-        return self.engine_adapter.fetchdf(sql)
+        return self.engine_adapter.fetchdf(query)
 
     def _create_df(
         self,
@@ -611,7 +607,10 @@ def test_ctes(self, ctes: t.Dict[str, exp.Expression], recursive: bool = False)
                 self.assert_equal(expected, actual, sort=sort, partial=partial)
 
     def runTest(self) -> None:
-        query = self._render_model_query()
+        with self._concurrent_render_context():
+            query = self._render_model_query()
+            sql = query.sql(self._test_adapter_dialect, pretty=self.engine_adapter._pretty_sql)
+
         with_clause = query.args.get("with")
 
         if with_clause:
@@ -628,7 +627,7 @@ def runTest(self) -> None:
             partial = values.get("partial")
             sort = query.args.get("order") is None
 
-            actual = self._execute(query)
+            actual = self._execute(sql)
             expected = self._create_df(values, columns=self.model.columns_to_types, partial=partial)
 
             self.assert_equal(expected, actual, sort=sort, partial=partial)

tests/core/test_test.py

@@ -2376,7 +2376,37 @@ def test_freeze_time_concurrent(tmp_path: Path) -> None:
     tests_dir = tmp_path / "tests"
     tests_dir.mkdir()
 
-    for model_name in ["sql_model", "py_model"]:
+    macros_dir = tmp_path / "macros"
+    macros_dir.mkdir()
+
+    macro_file = macros_dir / "test_datetime_now.py"
+    macro_file.write_text(
+        """
+from sqlglot import exp
+import datetime
+from sqlmesh.core.macros import macro
+
+@macro()
+def test_datetime_now(evaluator):
+  return exp.cast(exp.Literal.string(datetime.datetime.now(tz=datetime.timezone.utc)), exp.DataType.Type.DATE)
+  
+@macro()
+def test_sqlglot_expr(evaluator):
+  return exp.CurrentDate().sql(evaluator.dialect)
+    """
+    )
+
+    models_dir = tmp_path / "models"
+    models_dir.mkdir()
+    sql_model1 = models_dir / "sql_model1.sql"
+    sql_model1.write_text(
+        """
+        MODEL(NAME sql_model1);
+        SELECT @test_datetime_now() AS col_exec_ds_time, @test_sqlglot_expr() AS col_current_date;
+        """
+    )
+
+    for model_name in ["sql_model1", "sql_model2", "py_model"]:
         for i in range(5):
             test_2019 = tmp_path / "tests" / f"test_2019_{model_name}_{i}.yaml"
             test_2019.write_text(
@@ -2428,16 +2458,15 @@ def execute(context, start, end, execution_time, **kwargs):
         )
 
     python_model = model.get_registry()["py_model"].model(module_path=Path("."), path=Path("."))
+    ctx.upsert_model(python_model)
 
     ctx.upsert_model(
         _create_model(
-            meta="MODEL(NAME sql_model)",
+            meta="MODEL(NAME sql_model2)",
             query="SELECT @execution_ds::timestamp_ntz AS col_exec_ds_time, current_date()::date AS col_current_date",
             default_catalog=ctx.default_catalog,
         )
     )
 
-    ctx.upsert_model(python_model)
-
     results = ctx.test()
-    assert len(results.successes) == 20
+    assert len(results.successes) == 30