From 7580e8abdb2a82a1d1a5bdb4165c67c9c8c886b8 Mon Sep 17 00:00:00 2001
From: Mike Alfare <mike.alfare@dbtlabs.com>
Date: Fri, 20 Sep 2024 15:56:20 -0400
Subject: [PATCH] move dynamic table tests down into the relation tests folder

---
 .../test_configuration_changes.py             | 103 ++++++++++++++++++
 1 file changed, 103 insertions(+)
 create mode 100644 tests/functional/relation_tests/dynamic_table_tests/test_configuration_changes.py

diff --git a/tests/functional/relation_tests/dynamic_table_tests/test_configuration_changes.py b/tests/functional/relation_tests/dynamic_table_tests/test_configuration_changes.py
new file mode 100644
index 000000000..3c4f65a87
--- /dev/null
+++ b/tests/functional/relation_tests/dynamic_table_tests/test_configuration_changes.py
@@ -0,0 +1,103 @@
+import pytest
+
+from dbt.tests.util import run_dbt
+
+from tests.functional.relation_tests.dynamic_table_tests import models
+from tests.functional.utils import describe_dynamic_table, update_model
+
+
+class Changes:
+
+    @pytest.fixture(scope="class", autouse=True)
+    def seeds(self):
+        yield {"my_seed.csv": models.SEED}
+
+    @pytest.fixture(scope="class", autouse=True)
+    def models(self):
+        yield {
+            "dynamic_table_alter.sql": models.DYNAMIC_TABLE,
+            "dynamic_table_replace.sql": models.DYNAMIC_TABLE,
+        }
+
+    @pytest.fixture(scope="function", autouse=True)
+    def setup_class(self, project):
+        run_dbt(["seed"])
+        yield
+        project.run_sql(f"drop schema if exists {project.test_schema} cascade")
+
+    @pytest.fixture(scope="function", autouse=True)
+    def setup_method(self, project, setup_class):
+        # make sure the model in the data reflects the files each time
+        run_dbt(["run", "--full-refresh"])
+        self.assert_changes_are_not_applied(project)
+
+        update_model(project, "dynamic_table_alter", models.DYNAMIC_TABLE_ALTER)
+        update_model(project, "dynamic_table_replace", models.DYNAMIC_TABLE_REPLACE)
+
+        yield
+
+        update_model(project, "dynamic_table_alter", models.DYNAMIC_TABLE)
+        update_model(project, "dynamic_table_replace", models.DYNAMIC_TABLE)
+
+    @staticmethod
+    def assert_changes_are_applied(project):
+        altered = describe_dynamic_table(project, "dynamic_table_alter")
+        assert altered.snowflake_warehouse == "DBT_TESTING"
+        assert altered.target_lag == "5 minutes"  # this updated
+        assert altered.refresh_mode == "INCREMENTAL"
+
+        replaced = describe_dynamic_table(project, "dynamic_table_replace")
+        assert replaced.snowflake_warehouse == "DBT_TESTING"
+        assert replaced.target_lag == "2 minutes"
+        assert replaced.refresh_mode == "FULL"  # this updated
+
+    @staticmethod
+    def assert_changes_are_not_applied(project):
+        altered = describe_dynamic_table(project, "dynamic_table_alter")
+        assert altered.snowflake_warehouse == "DBT_TESTING"
+        assert altered.target_lag == "2 minutes"  # this would have updated, but didn't
+        assert altered.refresh_mode == "INCREMENTAL"
+
+        replaced = describe_dynamic_table(project, "dynamic_table_replace")
+        assert replaced.snowflake_warehouse == "DBT_TESTING"
+        assert replaced.target_lag == "2 minutes"
+        assert replaced.refresh_mode == "INCREMENTAL"  # this would have updated, but didn't
+
+    def test_full_refresh_is_always_successful(self, project):
+        # this always passes and always changes the configuration, regardless of on_configuration_change
+        # and regardless of whether the changes require a replace versus an alter
+        run_dbt(["run", "--full-refresh"])
+        self.assert_changes_are_applied(project)
+
+
+class TestChangesApply(Changes):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"models": {"on_configuration_change": "apply"}}
+
+    def test_changes_are_applied(self, project):
+        # this passes and changes the configuration
+        run_dbt(["run"])
+        self.assert_changes_are_applied(project)
+
+
+class TestChangesContinue(Changes):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"models": {"on_configuration_change": "continue"}}
+
+    def test_changes_are_not_applied(self, project):
+        # this passes but does not change the configuration
+        run_dbt(["run"])
+        self.assert_changes_are_not_applied(project)
+
+
+class TestChangesFail(Changes):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {"models": {"on_configuration_change": "fail"}}
+
+    def test_changes_are_not_applied(self, project):
+        # this fails and does not change the configuration
+        run_dbt(["run"], expect_pass=False)
+        self.assert_changes_are_not_applied(project)