chore(bigtable): prevent test leaks (#17350)

The bigtable system tests were leaving left-over instances, which could
cause future tests to fail until they were manually cleaned up

This PR improves the test logic to make sure resources are properly
cleaned up

Also added a fixture to remove test instances created more than a day
ago

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: daniel-sanche

packages/google-cloud-bigtable/tests/system/admin_overlay/test_system_async.py

@@ -17,7 +17,6 @@
 from typing import Tuple
 
 import pytest
-from google.api_core import exceptions
 from google.cloud.environment_vars import BIGTABLE_EMULATOR
 
 from google.cloud import bigtable_admin_v2 as admin_v2
@@ -89,33 +88,49 @@ async def instance_admin_client(admin_overlay_project_id):
 
 
 @CrossSync.convert
-@CrossSync.pytest_fixture(scope="session")
+@CrossSync.pytest_fixture(scope="session", autouse=True)
+async def cleanup_old_instances(admin_overlay_project_id):
+    """
+    Automatically deletes any test instances older than 1 day.
+
+    This fixture runs once per test session and helps prevent resource leakage
+    by cleaning up instances that failed to be deleted during previous test runs.
+    """
+    from tests.system.utils import clear_stale_instances
+
+    from .conftest import INSTANCE_PREFIX
+
+    clear_stale_instances(admin_overlay_project_id, INSTANCE_PREFIX, older_than_days=1)
+
+
+@CrossSync.convert
+@CrossSync.pytest_fixture(scope="function")
 async def instances_to_delete(instance_admin_client):
     instances = []
 
     try:
         yield instances
     finally:
-        for instance in instances:
+        for instance in reversed(instances):
             try:
                 await instance_admin_client.delete_instance(name=instance.name)
-            except exceptions.NotFound:
-                pass
+            except Exception as e:
+                print(f"Failed to delete instance {instance.name}: {e}")
 
 
 @CrossSync.convert
-@CrossSync.pytest_fixture(scope="session")
+@CrossSync.pytest_fixture(scope="function")
 async def backups_to_delete(table_admin_client):
     backups = []
 
     try:
         yield backups
     finally:
-        for backup in backups:
+        for backup in reversed(backups):
             try:
                 await table_admin_client.delete_backup(name=backup.name)
-            except exceptions.NotFound:
-                pass
+            except Exception as e:
+                print(f"Failed to delete backup {backup.name}: {e}")
 
 
 @CrossSync.convert
@@ -169,7 +184,8 @@ async def create_instance(
 
         # add to cleanup list before waiting for result, in case of timeout
         instance_name = instance_admin_client.instance_path(project_id, instance_id)
-        instances_to_delete.append(admin_v2.Instance(name=instance_name))
+        instance_placeholder = admin_v2.Instance(name=instance_name)
+        instances_to_delete.append(instance_placeholder)
 
         instance = await operation.result()
 
@@ -260,9 +276,9 @@ async def create_backup(
     )
 
     # add to cleanup list before waiting for result, in case of timeout
-    backups_to_delete.append(
-        admin_v2.Backup(name=f"{cluster_name}/backups/{backup_id}")
-    )
+    backup_name = f"{cluster_name}/backups/{backup_id}"
+    backup_placeholder = admin_v2.Backup(name=backup_name)
+    backups_to_delete.append(backup_placeholder)
 
     backup = await operation.result()
 

packages/google-cloud-bigtable/tests/system/admin_overlay/test_system_autogen.py

@@ -20,7 +20,6 @@
 from typing import Tuple
 
 import pytest
-from google.api_core import exceptions
 from google.api_core import operation as api_core_operation
 from google.cloud.environment_vars import BIGTABLE_EMULATOR
 
@@ -73,30 +72,43 @@ def instance_admin_client(admin_overlay_project_id):
         yield client
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="session", autouse=True)
+def cleanup_old_instances(admin_overlay_project_id):
+    """Automatically deletes any test instances older than 1 day.
+
+    This fixture runs once per test session and helps prevent resource leakage
+    by cleaning up instances that failed to be deleted during previous test runs."""
+    from tests.system.utils import clear_stale_instances
+
+    from .conftest import INSTANCE_PREFIX
+
+    clear_stale_instances(admin_overlay_project_id, INSTANCE_PREFIX, older_than_days=1)
+
+
+@pytest.fixture(scope="function")
 def instances_to_delete(instance_admin_client):
     instances = []
     try:
         yield instances
     finally:
-        for instance in instances:
+        for instance in reversed(instances):
             try:
                 instance_admin_client.delete_instance(name=instance.name)
-            except exceptions.NotFound:
-                pass
+            except Exception as e:
+                print(f"Failed to delete instance {instance.name}: {e}")
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="function")
 def backups_to_delete(table_admin_client):
     backups = []
     try:
         yield backups
     finally:
-        for backup in backups:
+        for backup in reversed(backups):
             try:
                 table_admin_client.delete_backup(name=backup.name)
-            except exceptions.NotFound:
-                pass
+            except Exception as e:
+                print(f"Failed to delete backup {backup.name}: {e}")
 
 
 def create_instance(
@@ -135,7 +147,8 @@ def create_instance(
         )
         operation = instance_admin_client.create_instance(create_instance_request)
         instance_name = instance_admin_client.instance_path(project_id, instance_id)
-        instances_to_delete.append(admin_v2.Instance(name=instance_name))
+        instance_placeholder = admin_v2.Instance(name=instance_name)
+        instances_to_delete.append(instance_placeholder)
         instance = operation.result()
         instances_to_delete[-1] = instance
     create_table_request = admin_v2.CreateTableRequest(
@@ -198,9 +211,9 @@ def create_backup(
             ),
         )
     )
-    backups_to_delete.append(
-        admin_v2.Backup(name=f"{cluster_name}/backups/{backup_id}")
-    )
+    backup_name = f"{cluster_name}/backups/{backup_id}"
+    backup_placeholder = admin_v2.Backup(name=backup_name)
+    backups_to_delete.append(backup_placeholder)
     backup = operation.result()
     backups_to_delete[-1] = backup
     return backup

packages/google-cloud-bigtable/tests/system/data/__init__.py

@@ -34,6 +34,15 @@ class SystemTestRunner:
     used by standard system tests, and metrics tests
     """
 
+    @pytest.fixture(scope="session", autouse=True)
+    def cleanup_old_instances(self, project_id):
+        """
+        Automatically deletes any test instances older than 1 day.
+        """
+        from tests.system.utils import clear_stale_instances
+
+        clear_stale_instances(project_id, "python-bigtable-tests", older_than_days=1)
+
     @pytest.fixture(scope="session")
     def init_table_id(self):
         """
@@ -128,8 +137,8 @@ def instance_id(self, admin_client, project_id, cluster_config):
                     admin_client.instance_admin_client.delete_instance(
                         name=f"projects/{project_id}/instances/{instance_id}"
                     )
-                except exceptions.NotFound:
-                    pass
+                except Exception as e:
+                    print(f"Failed to delete instance {instance_id}: {e}")
 
     @pytest.fixture(scope="session")
     def column_split_config(self):
@@ -195,8 +204,8 @@ def table_id(
                 admin_client.table_admin_client.delete_table(
                     name=f"{parent_path}/tables/{init_table_id}"
                 )
-            except exceptions.NotFound:
-                print(f"Table {init_table_id} not found, skipping deletion")
+            except Exception as e:
+                print(f"Failed to delete table {init_table_id}: {e}")
 
     @pytest.fixture(scope="session")
     def authorized_view_id(
@@ -256,8 +265,8 @@ def authorized_view_id(
                     admin_client.table_admin_client.delete_authorized_view(
                         name=new_path
                     )
-                except exceptions.NotFound:
-                    print(f"View {new_view_id} not found, skipping deletion")
+                except Exception as e:
+                    print(f"Failed to delete view {new_view_id}: {e}")
 
     @pytest.fixture(scope="session")
     def project_id(self, client):

packages/google-cloud-bigtable/tests/system/utils.py

@@ -0,0 +1,58 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from datetime import datetime, timedelta, timezone
+
+from google.api_core.exceptions import NotFound
+
+from google.cloud import bigtable_admin_v2 as admin_v2
+
+
+def clear_stale_instances(project_id: str, prefix: str, older_than_days: int = 1):
+    """
+    Synchronously deletes any instances in the given project that are older
+    than older_than_days and whose name or display name matches the given prefix.
+    """
+    client = admin_v2.BigtableInstanceAdminClient(
+        client_options={"quota_project_id": project_id}
+    )
+    parent = client.common_project_path(project_id)
+    next_page_token = ""
+
+    while True:
+        try:
+            response = client.list_instances(
+                request={"parent": parent, "page_token": next_page_token}
+            )
+        except Exception:
+            # Cannot list instances, skip cleanup
+            break
+
+        for instance in response.instances:
+            # Check if instance matches the prefix
+            display_name_matches = instance.display_name.startswith(prefix)
+            name_matches = instance.name.split("/")[-1].startswith(prefix)
+
+            if display_name_matches or name_matches:
+                if instance.create_time:
+                    now = datetime.now(timezone.utc)
+                    if now - instance.create_time > timedelta(days=older_than_days):
+                        try:
+                            client.delete_instance(name=instance.name)
+                        except NotFound:
+                            pass
+
+        next_page_token = response.next_page_token
+        if not next_page_token:
+            break