Add functional tests

tests/functional/adapter/test_incremental_iceberg.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+from dbt.contracts.results import RunStatus
 from dbt.tests.adapter.incremental.test_incremental_merge_exclude_columns import (
     BaseMergeExcludeColumns,
 )
@@ -30,6 +31,7 @@
     seeds__duplicate_insert_sql,
     seeds__seed_csv,
 )
+from dbt.tests.util import check_relations_equal, run_dbt
 
 seeds__expected_incremental_predicates_csv = """id,msg,color
 3,anyway,purple
@@ -78,13 +80,50 @@
 {% endif %}
 """
 
-
 seeds__expected_merge_exclude_all_columns_csv = """id,msg,color
 1,hello,blue
 2,goodbye,red
 3,anyway,purple
 """
 
+models__update_condition_sql = """
+{{ config(
+        table_type='iceberg',
+        materialized='incremental',
+        incremental_strategy='merge',
+        unique_key=['id'],
+        update_condition='target.id > 1'
+    )
+}}
+
+{% if is_incremental() %}
+
+select * from (
+    values
+    (1, 'v1-updated')
+    , (2, 'v2-updated')
+) as t (id, value)
+
+{% else %}
+
+select * from (
+    values
+    (-1, 'v-1')
+    , (0, 'v0')
+    , (1, 'v1')
+    , (2, 'v2')
+) as t (id, value)
+
+{% endif %}
+"""
+
+seeds__expected_update_condition_csv = """id,value
+-1,v-1
+0,v0
+1,v1
+2,v2-updated
+"""
+
 
 class TestIcebergIncrementalUniqueKey(BaseIncrementalUniqueKey):
     @pytest.fixture(scope="class")
@@ -254,6 +293,33 @@ def seeds(self):
         return {"expected_merge_exclude_columns.csv": seeds__expected_merge_exclude_all_columns_csv}
 
 
+class TestIcebergUpdateCondition:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"merge_update_condition.sql": models__update_condition_sql}
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {"expected_merge_update_condition.csv": seeds__expected_update_condition_csv}
+
+    def test__merge_update_condition(self, project):
+        """Seed should match the model after incremental run"""
+
+        expected_seed_name = "expected_merge_update_condition"
+        run_dbt(["seed", "--select", expected_seed_name, "--full-refresh"])
+
+        relation_name = "merge_update_condition"
+        model_run = run_dbt(["run", "--select", relation_name])
+        model_run_result = model_run.results[0]
+        assert model_run_result.status == RunStatus.Success
+
+        model_update = run_dbt(["run", "--select", relation_name])
+        model_update_result = model_update.results[0]
+        assert model_update_result.status == RunStatus.Success
+
+        check_relations_equal(project.adapter, [relation_name, expected_seed_name])
+
+
 def replace_cast_date(model: str) -> str:
     """Wrap all date strings with a cast date function"""