test: make column-tag functional tests rerun-safe

CHANGELOG.md

@@ -17,6 +17,7 @@
 
 ### Under the Hood
 
+- Make the column-tag functional tests rerun-safe so a `pytest --reruns` retry no longer inherits mutated state (updated `schema.yml`, leftover column tags, a running streaming-table query) from the failed attempt (test-only, no runtime impact). ([#1499](https://github.com/databricks/dbt-databricks/pull/1499))
 - Raise the `dbt-tests-adapter` test-dependency floor to `>=1.20.0` to pick up its `persist_docs` fixture typo fix (test-only, no runtime impact) ([#1490](https://github.com/databricks/dbt-databricks/pull/1490))
 - Defer SDK `Config` construction to connection-open time so offline paths (`dbt parse`/`list`/`compile`) don't trigger the host-metadata probe introduced in `databricks-sdk>=0.103`; as a side effect, auth errors now surface at first connection rather than during profile parsing. ([#1474](https://github.com/databricks/dbt-databricks/pull/1474))
 - Bump ceilings on `databricks-sdk` (now `<0.105.0`) and `databricks-sql-connector[pyarrow]` (now `<4.3.0`) to admit newer releases; floors unchanged. ([#1474](https://github.com/databricks/dbt-databricks/pull/1474))

tests/functional/adapter/column_tags/test_column_tags.py

@@ -2,10 +2,14 @@
 from dbt.tests import util
 
 from tests.functional.adapter.column_tags import fixtures
-from tests.functional.adapter.fixtures import MaterializationV2Mixin
+from tests.functional.adapter.fixtures import MaterializationV2Mixin, RerunSafeMixin
 
 
-class ColumnTagsMixin(MaterializationV2Mixin):
+class ColumnTagsMixin(RerunSafeMixin, MaterializationV2Mixin):
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("base_model",)
+
     @pytest.fixture(scope="class")
     def models(self):
         return {
@@ -99,6 +103,13 @@ def models(self):
     def setup_streaming_table_seed(self, project):
         util.run_dbt(["seed"])
 
+    @pytest.fixture(autouse=True)
+    def stop_streaming_query(self, project):
+        # Drop base_model on teardown to stop its streaming query promptly, so it
+        # can't orphan and cascade failures into other tests on the same xdist worker.
+        yield
+        self._drop_relations(project, ("base_model",))
+
 
 @pytest.mark.skip_profile("databricks_cluster")
 class TestColumnTagsView(ColumnTagsMixin):

tests/functional/adapter/column_tags/test_snapshot_column_tags.py

@@ -2,10 +2,15 @@
 from dbt.tests import util
 
 from tests.functional.adapter.column_tags import fixtures
+from tests.functional.adapter.fixtures import RerunSafeMixin
 
 
 @pytest.mark.skip_profile("databricks_cluster")
-class TestSnapshotColumnTags:
+class TestSnapshotColumnTags(RerunSafeMixin):
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ("snapshot",)
+
     @pytest.fixture(scope="class")
     def snapshots(self):
         return {"snapshot.sql": fixtures.snapshot_column_tag_sql}

tests/functional/adapter/fixtures.py

@@ -26,6 +26,42 @@
 """
 
 
+class RerunSafeMixin:
+    """Make a stateful functional test safe to retry under `pytest --reruns`.
+
+    pytest-rerunfailures retries a failed test without tearing down the class-scoped
+    `project` fixture, so on-disk model files (often rewritten in place via
+    write_file) and previously-created relations leak into the retry. Subclasses
+    name the relations they build via the `relations_to_reset` fixture; this
+    function-scoped fixture re-runs on every attempt, restoring the initial model
+    files and dropping those relations so each attempt starts from a clean slate.
+    """
+
+    @pytest.fixture(scope="class")
+    def relations_to_reset(self):
+        return ()
+
+    @pytest.fixture(autouse=True)
+    def reset_project_state(self, project, models, relations_to_reset):
+        for name, contents in models.items():
+            util.write_file(contents, "models", name)
+        self._drop_relations(project, relations_to_reset)
+        yield
+
+    @staticmethod
+    def _drop_relations(project, identifiers):
+        """Drop each named relation in the test schema if it exists."""
+        with project.adapter.connection_named("__test_reset"):
+            for identifier in identifiers:
+                relation = project.adapter.get_relation(
+                    database=project.database,
+                    schema=project.test_schema,
+                    identifier=identifier,
+                )
+                if relation is not None:
+                    project.adapter.drop_relation(relation)
+
+
 class MaterializationV1Mixin:
     @pytest.fixture(scope="class")
     def project_config_update(self):