fix(sqlalchemy-spanner): isolate test database names and extend stale cleanup cutoff

packages/sqlalchemy-spanner/create_test_config.py

@@ -15,6 +15,7 @@
 # limitations under the License.
 
 import configparser
+import os
 import sys
 
 
@@ -41,7 +42,8 @@ def set_test_config(
     config.add_section("db")
     config["db"]["default"] = url
 
-    with open("test.cfg", "w") as configfile:
+    config_filename = os.getenv("SQLALCHEMY_SPANNER_CONFIG", "test.cfg")
+    with open(config_filename, "w") as configfile:
         config.write(configfile)
 
 

packages/sqlalchemy-spanner/create_test_database.py

@@ -14,18 +14,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import os
+import pathlib
 import re
 import time
+import uuid
 
+from create_test_config import set_test_config
 from google.api_core import datetime_helpers
 from google.api_core.exceptions import AlreadyExists, ResourceExhausted
 from google.cloud.spanner_v1 import Client
 from google.cloud.spanner_v1.database import Database
 from google.cloud.spanner_v1.instance import Instance
 
-from create_test_config import set_test_config
-
 USE_EMULATOR = os.getenv("SPANNER_EMULATOR_HOST") is not None
 
 PROJECT = os.getenv(
@@ -69,26 +71,39 @@ def delete_stale_test_instances():
 
 
 def delete_stale_test_databases():
-    """Delete test databases that are older than 10 minutes.
-
-    In this test suite, active databases typically finish running in ~5 minutes.
-    To prevent concurrent Kokoro runs from accidentally deleting each other's
-    active databases we use a 10-minute safety threshold. Without an aggressive
-    cutoff we quickly bump up against Cloud Spanner's limit of 100 databases per instance.
+    """Delete test databases that are older than 4 hours.
+
+    Uses a .stale_cleanup_done sentinel file gate to ensure this global sweep
+    runs exactly once at the start of a test run across parallel/parametrized sessions,
+    preventing concurrent sessions from interfering with each other.
     """
-    cutoff = (int(time.time()) - 10 * 60) * 1000
+    marker = ".stale_cleanup_done"
+    if os.path.exists(marker):
+        return
+
+    try:
+        pathlib.Path(marker).touch(exist_ok=False)
+    except FileExistsError:
+        return  # Another parallel process already performed cleanup
+
+    cutoff = (int(time.time()) - 4 * 60 * 60) * 1000
     instance = CLIENT.instance("sqlalchemy-dialect-test")
     if not instance.exists():
         return
     database_pbs = instance.list_databases()
     for database_pb in database_pbs:
         database = Database.from_pb(database_pb, instance)
-        # Parse creation time from database ID first (e.g. "sqlalchemy-test-1779989493809")
-        # to be 100% independent of emulator metadata or GCP Client API create_time gaps!
+
+        # Parse creation time from database ID first (e.g. "sp_test_1787069488_a3f")
         create_time = None
-        match = re.match(r"sqlalchemy-test-(\d+)", database.database_id)
+        match = re.match(r"sp_test_(\d+)", database.database_id)
         if match:
-            create_time = int(match.group(1))
+            ts_str = match.group(1)
+            ts_val = int(ts_str)
+            if len(ts_str) == 10:
+                create_time = ts_val * 1000
+            else:
+                create_time = ts_val
         elif database_pb.create_time is not None:
             create_time = datetime_helpers.to_milliseconds(database_pb.create_time)
 
@@ -123,8 +138,12 @@ def create_test_instance():
         except AlreadyExists:
             pass  # instance was already created
 
-    unique_resource_id = "%s%d" % ("-", 1000 * time.time())
-    database_id = "sqlalchemy-test" + unique_resource_id
+    # Generate a session-isolated unique database ID within Spanner 30-char limit
+    # Format: sp_test_{timestamp_in_seconds}_{rand_hex3} (compliant with Spanner naming: ^[a-z][a-z0-9_]{1,29}$)
+    creation_timestamp = time.time()
+    timestamp_part = str(int(creation_timestamp))
+    rand_part = uuid.uuid4().hex[:8]
+    database_id = f"sp_test_{timestamp_part}_{rand_part}"
 
     try:
         database = instance.database(database_id)
@@ -135,6 +154,44 @@ def create_test_instance():
 
     set_test_config(PROJECT, instance_id, database_id)
 
+    # Record metadata for duration tracking on teardown
+    meta_path = os.path.join(os.path.dirname(__file__), ".db_session_info.json")
+    with open(meta_path, "w") as f:
+        json.dump({"database_id": database_id, "creation_time": creation_timestamp}, f)
+
+
+def main(argv):
+    config_filename = argv[0] if argv else os.getenv("SQLALCHEMY_SPANNER_CONFIG", "test.cfg")
+    os.environ["SQLALCHEMY_SPANNER_CONFIG"] = config_filename
+
+    delete_stale_test_databases()
+    create_test_instance()
+
+    instance_id = "sqlalchemy-dialect-test"
+    instance = CLIENT.instance(instance_id)
+
+    # Generate a session-isolated unique database ID within Spanner 30-char limit
+    # Format: sp_test_{timestamp_in_seconds}_{rand_hex8} (compliant with Spanner naming: ^[a-z][a-z0-9_]{1,29}$)
+    creation_timestamp = time.time()
+    timestamp_part = str(int(creation_timestamp))
+    rand_part = uuid.uuid4().hex[:8]
+    database_id = f"sp_test_{timestamp_part}_{rand_part}"
+
+    try:
+        database = instance.database(database_id)
+        created_op = database.create()
+        created_op.result(1800)
+    except AlreadyExists:
+        pass  # database was already created
+
+    set_test_config(PROJECT, instance_id, database_id)
+
+    # Record metadata for duration tracking on teardown
+    meta_path = os.path.join(os.path.dirname(__file__), ".db_session_info.json")
+    with open(meta_path, "w") as f:
+        json.dump({"database_id": database_id, "creation_time": creation_timestamp}, f)
+
 
-delete_stale_test_databases()
-create_test_instance()
+if __name__ == "__main__":
+    import sys
+    main(sys.argv[1:])

packages/sqlalchemy-spanner/drop_test_database.py

@@ -15,18 +15,18 @@
 # limitations under the License.
 
 import configparser
+import json
 import os
 import re
 import time
 
+from create_test_config import set_test_config
 from google.api_core import datetime_helpers
 from google.api_core.exceptions import AlreadyExists, ResourceExhausted
 from google.cloud.spanner_v1 import Client
 from google.cloud.spanner_v1.database import Database
 from google.cloud.spanner_v1.instance import Instance
 
-from create_test_config import set_test_config
-
 USE_EMULATOR = os.getenv("SPANNER_EMULATOR_HOST") is not None
 
 PROJECT = os.getenv(
@@ -43,21 +43,76 @@
     CLIENT = Client(project=PROJECT)
 
 
+def format_duration(seconds):
+    mins = int(seconds // 60)
+    secs = int(seconds % 60)
+    if mins > 0:
+        return f"{mins} minutes and {secs} seconds"
+    else:
+        return f"{secs} seconds"
+
+
 def delete_test_database():
     """Delete the currently configured test database."""
     config = configparser.ConfigParser()
-    if os.path.exists("test.cfg"):
-        config.read("test.cfg")
+    config_env_val = os.getenv("SQLALCHEMY_SPANNER_CONFIG")
+    if config_env_val:
+        config_filename = config_env_val
+        if not os.path.exists(config_filename):
+            print(f"[Spanner DB] Config file {config_filename} specified in SQLALCHEMY_SPANNER_CONFIG does not exist. Skipping database drop.")
+            return
+    elif os.path.exists("test.cfg"):
+        config_filename = "test.cfg"
     else:
-        config.read("setup.cfg")
+        config_filename = "setup.cfg"
+
+    config.read(config_filename)
+
     db_url = config.get("db", "default")
+    if not db_url.startswith("spanner"):
+        print(f"[Spanner DB] Database URL {db_url} is not a Spanner URL. Skipping database drop.")
+        return
 
     instance_id = re.findall(r"instances(.*?)databases", db_url)
     database_id = re.findall(r"databases(.*?)$", db_url)
 
     instance = CLIENT.instance(instance_id="".join(instance_id).replace("/", ""))
-    database = instance.database("".join(database_id).replace("/", ""))
+    database_id_str = "".join(database_id).replace("/", "")
+    database = instance.database(database_id_str)
     database.drop()
 
+    # Calculate and report active duration with type-validation for compliance
+    meta_path = os.path.join(os.path.dirname(__file__), ".db_session_info.json")
+    if os.path.exists(meta_path):
+        try:
+            with open(meta_path, "r") as f:
+                meta = json.load(f)
+            if isinstance(meta, dict):
+                creation_time = meta.get("creation_time", time.time())
+                db_name = meta.get("database_id", database_id_str)
+                elapsed_seconds = time.time() - creation_time
+                duration_str = format_duration(elapsed_seconds)
+                print(f"[Spanner DB] Database {db_name} was active for {duration_str} before teardown.")
+        except Exception:
+            pass
+        finally:
+            if os.path.exists(meta_path):
+                os.remove(meta_path)
+
+    # Clean up session-specific config file
+    if os.path.exists(config_filename) and config_filename != "setup.cfg":
+        try:
+            os.remove(config_filename)
+        except Exception:
+            pass
+
+
+def main(argv):
+    config_filename = argv[0] if argv else os.getenv("SQLALCHEMY_SPANNER_CONFIG", "test.cfg")
+    os.environ["SQLALCHEMY_SPANNER_CONFIG"] = config_filename
+    delete_test_database()
+
 
-delete_test_database()
+if __name__ == "__main__":
+    import sys
+    main(sys.argv[1:])