[codex] Improve Jinja macro guardrails

macros/plugins/bigquery/create_external_table.sql

@@ -3,10 +3,11 @@
     {%- set external = source_node.external -%}
     {%- set partitions = external.partitions -%}
     {%- set options = external.options -%}
+    {%- set excluded_options = ['uris', 'connection_name'] %}
     {%- set non_string_options = ['max_staleness'] %}
 
     {% if options is mapping and options.get('connection_name', none) %}
-        {% set connection_name = options.pop('connection_name') %}
+        {% set connection_name = options.get('connection_name') %}
     {% endif %}
 
     {%- set uris = [] -%}
@@ -41,7 +42,7 @@
         options (
             uris = [{%- for uri in uris -%} '{{uri}}' {{- "," if not loop.last}} {%- endfor -%}]
             {%- if options is mapping -%}
-            {%- for key, value in options.items() if key != 'uris' %}
+            {%- for key, value in options.items() if key not in excluded_options %}
                 {%- if value is string and key not in non_string_options -%}
                 , {{key}} = '{{value}}'
                 {%- else -%}

macros/plugins/fabric/create_external_table.sql

@@ -9,7 +9,8 @@
     create external table {{source(source_node.source_name, source_node.name)}} (
         {% for column in columns %}
             {# TODO set nullity based on schema tests?? #}
-            {%- set nullity = 'NOT NULL' if 'not_null' in columns.tests else 'NULL'-%}
+            {%- set column_tests = column.tests if 'tests' in column else [] -%}
+            {%- set nullity = 'NOT NULL' if 'not_null' in column_tests else 'NULL'-%}
             {{adapter.quote(column.name)}} {{column.data_type}} {{nullity}}
             {{- ',' if not loop.last -}}
         {% endfor %}

macros/plugins/redshift/helpers/add_partitions.sql

@@ -20,12 +20,12 @@
 #}
 {% macro redshift_alter_table_add_partitions(source_node, partitions) %}
 
-  {{ log("Generating ADD PARTITION statement for partition set between " 
-         ~ partitions[0]['path'] ~ " and " ~ (partitions|last)['path']) }}
-
   {% set ddl = [] %}
 
   {% if partitions|length > 0 %}
+
+    {{ log("Generating ADD PARTITION statement for partition set between " 
+           ~ partitions[0]['path'] ~ " and " ~ (partitions|last)['path']) }}
 
     {% set alter_table_add %}
         alter table {{source(source_node.source_name, source_node.name)}} add if not exists 

macros/plugins/snowflake/refresh_external_table.sql

@@ -1,11 +1,8 @@
 {% macro snowflake__refresh_external_table(source_node) %}
 
     {% set external = source_node.external %}
-    {% set snowpipe = source_node.external.get('snowpipe', none) %}
 
     {% set auto_refresh = external.get('auto_refresh', false) %}
-    {% set partitions = external.get('partitions', none) %}
-    {% set delta_format = (external.table_format | lower == "delta") %}
 
     {% set manual_refresh = not auto_refresh %}
 

tests/test_jinja_macro_contracts.py

@@ -0,0 +1,37 @@
+from pathlib import Path
+from unittest import TestCase
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+
+def read_macro(path: str) -> str:
+    return (REPO_ROOT / path).read_text()
+
+
+class JinjaMacroContractsTest(TestCase):
+    def test_bigquery_connection_name_does_not_mutate_options(self):
+        macro = read_macro("macros/plugins/bigquery/create_external_table.sql")
+
+        self.assertNotIn(".pop(", macro)
+        self.assertIn("key not in excluded_options", macro)
+
+    def test_redshift_empty_partitions_checked_before_logging_range(self):
+        macro = read_macro("macros/plugins/redshift/helpers/add_partitions.sql")
+
+        empty_guard = macro.index("if partitions|length > 0")
+        first_partition_lookup = macro.index("partitions[0]")
+        self.assertLess(empty_guard, first_partition_lookup)
+
+    def test_fabric_nullity_uses_current_column_tests(self):
+        macro = read_macro("macros/plugins/fabric/create_external_table.sql")
+
+        self.assertNotIn("columns.tests", macro)
+        self.assertIn("column_tests", macro)
+
+    def test_snowflake_refresh_has_no_unused_branch_variables(self):
+        macro = read_macro("macros/plugins/snowflake/refresh_external_table.sql")
+
+        self.assertNotIn("set snowpipe", macro)
+        self.assertNotIn("set partitions", macro)
+        self.assertNotIn("set delta_format", macro)