Contract enforcement on temporary tables (#8889)

* add test

* fix test

* first pass with constraint error

* add back column checks for temp tables

* changelog

* Update .changes/unreleased/Fixes-20231024-145504.yaml

(cherry picked from commit 98310b6)

.changes/unreleased/Fixes-20231024-145504.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Add back contract enforcement for temporary tables on postgres
+time: 2023-10-24T14:55:04.051683-05:00
+custom:
+  Author: emmyoop
+  Issue: "8857"

plugins/postgres/dbt/include/postgres/macros/adapters.sql

@@ -10,9 +10,11 @@
     unlogged
   {%- endif %} table {{ relation }}
   {% set contract_config = config.get('contract') %}
-  {% if contract_config.enforced and (not temporary) %}
+  {% if contract_config.enforced %}
     {{ get_assert_columns_equivalent(sql) }}
-    {{ get_table_columns_and_constraints() }} ;
+  {% endif -%}
+  {% if contract_config.enforced and (not temporary) -%}
+      {{ get_table_columns_and_constraints() }} ;
     insert into {{ relation }} (
       {{ adapter.dispatch('get_column_names', 'dbt')() }}
     )

tests/functional/contracts/test_contract_enforcement.py

@@ -0,0 +1,44 @@
+import pytest
+from dbt.tests.util import run_dbt, write_file
+
+
+my_model_sql = """
+select 'some string' as string_column
+"""
+
+my_model_int_sql = """
+select 123 as int_column
+"""
+
+model_schema_yml = """
+models:
+  - name: my_model
+    config:
+      materialized: incremental
+      on_schema_change: append_new_columns
+      contract: {enforced: true}
+    columns:
+      - name: string_column
+        data_type: text
+"""
+
+
+class TestIncrementalModelContractEnforcement:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "my_model.sql": my_model_sql,
+            "schema.yml": model_schema_yml,
+        }
+
+    def test_contracted_incremental(self, project):
+        results = run_dbt()
+        assert len(results) == 1
+        # now update the column type in the model to break the contract
+        write_file(my_model_int_sql, project.project_root, "models", "my_model.sql")
+
+        expected_msg = "This model has an enforced contract that failed."
+        results = run_dbt(expect_pass=False)
+        assert len(results) == 1
+        msg = results[0].message
+        assert expected_msg in msg