test fixes

Author: GuyEshdat

integration_tests/tests/test_dbt_artifacts/test_microbatch_compiled_code.py

@@ -1,14 +1,13 @@
+from contextlib import contextmanager
+from pathlib import Path
+
 import pytest
 
 from dbt_project import DbtProject
 
 
-@pytest.mark.skip_targets(["vertica"])
-@pytest.mark.skip_for_dbt_fusion
-def test_microbatch_run_results_has_compiled_code(test_id: str, dbt_project: DbtProject):
-    dbt_project.dbt_runner.vars["disable_run_results"] = False
-
-    model_sql = """
+def _microbatch_model_sql() -> str:
+    return """
 {% set model_config = {
     "materialized": "incremental",
     "incremental_strategy": "microbatch",
@@ -28,15 +27,19 @@ def test_microbatch_run_results_has_compiled_code(test_id: str, dbt_project: Dbt
 {{ config(**model_config) }}
 
 select
-    cast({{ elementary.escape_reserved_keywords("one") }} as int) as order_id,
+    1 as order_id,
     1 as customer_id,
     42 as amount,
     {{ dbt.current_timestamp() }} as order_date
 from {{ ref('one') }}
 """
 
+
+def _run_microbatch_model_and_get_latest_success_result(
+    dbt_project: DbtProject, test_id: str
+):
     with dbt_project.create_temp_model_for_existing_table(
-        test_id, raw_code=model_sql
+        test_id, raw_code=_microbatch_model_sql()
     ) as model_path:
         dbt_project.dbt_runner.run(select=str(model_path))
 
@@ -47,7 +50,50 @@ def test_microbatch_run_results_has_compiled_code(test_id: str, dbt_project: Dbt
         order_by="generated_at desc",
         limit=1,
     )
+    return run_results
+
+
+@contextmanager
+def _without_microbatch_override_macro(dbt_project: DbtProject):
+    macro_path = (
+        dbt_project.project_dir_path / "macros" / "microbatch.sql"
+    )
+    backup_path = macro_path.with_suffix(".sql.bak")
+    if not macro_path.exists():
+        raise FileNotFoundError(f"Expected macro file at {macro_path}")
+
+    macro_path.rename(backup_path)
+    try:
+        yield
+    finally:
+        if backup_path.exists():
+            backup_path.rename(macro_path)
+
+
+@pytest.mark.skip_targets(["vertica"])
+@pytest.mark.skip_for_dbt_fusion
+def test_microbatch_run_results_has_compiled_code(test_id: str, dbt_project: DbtProject):
+    dbt_project.dbt_runner.vars["disable_run_results"] = False
+
+    run_results = _run_microbatch_model_and_get_latest_success_result(dbt_project, test_id)
     assert run_results, "Expected a successful run result row for microbatch model"
     assert run_results[0]["compiled_code"], (
         "Expected compiled_code to be populated for successful microbatch model run result"
     )
+
+
+@pytest.mark.skip_targets(["vertica"])
+@pytest.mark.skip_for_dbt_fusion
+def test_microbatch_run_results_without_override_has_empty_compiled_code(
+    test_id: str, dbt_project: DbtProject
+):
+    dbt_project.dbt_runner.vars["disable_run_results"] = False
+
+    with _without_microbatch_override_macro(dbt_project):
+        run_results = _run_microbatch_model_and_get_latest_success_result(
+            dbt_project, test_id
+        )
+    assert run_results, "Expected a successful run result row for microbatch model"
+    assert not run_results[0]["compiled_code"], (
+        "Expected compiled_code to stay empty when microbatch override macro is absent"
+    )