tests: cleanup stale databases in spanner system tests (#17668)

### Problem
Spanner system tests can sometimes leave behind test databases if a run
is interrupted or fails to clean up properly. Over time, these stale
databases can accumulate, potentially hitting quota limits or cluttering
the test environment.

### Solution
Introduced a proactive cleanup mechanism to remove stale databases
during test setup.

* **Added `cleanup_stale_databases` helper**: A new function in
`_helpers.py` that lists databases in a given instance and drops any
that are older than a specified cutoff (defaulting to 10 minutes/600
seconds).
* **Drop Protection Handling**: The helper explicitly disables
`enable_drop_protection` if it's set, ensuring that even protected test
databases can be cleaned up automatically.
* **Fixture Integration**: Integrated this cleanup call into the
`shared_instance` fixture in `conftest.py`, ensuring that every test
session starts by cleaning up residue from previous runs.

### Impact
*   Reduces the risk of test failures due to database quota exhaustion.
*   Keeps the Spanner instance used for testing clean.
* Resilient against partial failures by handling `NotFound` and other
API errors during cleanup.

Author: chalmerlowe

packages/google-cloud-spanner/tests/system/_helpers.py

@@ -17,10 +17,10 @@
 import time
 import uuid
 
-from google.api_core import exceptions
+from google.api_core import datetime_helpers, exceptions
+from google.cloud.spanner_v1 import instance as instance_mod
 from test_utils import retry
 
-from google.cloud.spanner_v1 import instance as instance_mod
 from tests import _fixtures
 
 CREATE_INSTANCE_ENVVAR = "GOOGLE_CLOUD_TESTS_CREATE_SPANNER_INSTANCE"
@@ -159,6 +159,30 @@ def cleanup_old_instances(spanner_client):
                 scrub_instance_ignore_not_found(instance)
 
 
+def cleanup_stale_databases(instance, cutoff_seconds=600):
+    """Delete stale databases in the given instance older than cutoff_seconds."""
+    cutoff_ms = (int(time.time()) - cutoff_seconds) * 1000
+
+    for database_pb in instance.list_databases():
+        if database_pb.create_time is not None:
+            create_time_ms = datetime_helpers.to_milliseconds(database_pb.create_time)
+
+            if create_time_ms < cutoff_ms:
+                db = instance.database(database_pb.name.split("/")[-1])
+                try:
+                    db.reload()
+                    if db.enable_drop_protection:
+                        db.enable_drop_protection = False
+                        operation = db.update(["enable_drop_protection"])
+                        operation.result(DATABASE_OPERATION_TIMEOUT_IN_SECONDS)
+                    db.drop()
+                except exceptions.NotFound:
+                    pass
+                except exceptions.GoogleAPIError:
+                    # Ignore other API errors during cleanup
+                    pass
+
+
 def unique_id(prefix, separator="-"):
     # Database name size: Spanner database names are limited to 30 characters.
     # See: https://docs.cloud.google.com/spanner/docs/reference/rpc/google.spanner.admin.database.v1#createdatabaserequest

packages/google-cloud-spanner/tests/system/conftest.py

@@ -17,7 +17,6 @@
 import time
 
 import pytest
-
 from google.cloud import spanner_v1
 from google.cloud.spanner_admin_database_v1 import DatabaseDialect
 from google.cloud.spanner_admin_database_v1.types.backup import (
@@ -218,6 +217,8 @@ def shared_instance(
         instance = spanner_client.instance(shared_instance_id)
         instance.reload()
 
+    _helpers.cleanup_stale_databases(instance)
+
     yield instance
 
     if _helpers.CREATE_INSTANCE: