test(spanner): add pytest-xdist parallel execution with state isolation (#17344)

This PR adds `pytest-xdist` to parallelize unit tests and the
`core_deps_from_source` nox sessions for the `google-cloud-spanner`
package.

**By running tests in parallel using `-n auto`, the execution time of
the Spanner unit tests are reduced from ~8 minutes to 4 minutes.**

State isolation and test reliability are achieved by:

* **Simplifying Subtests**: We pass simple strings/names to `subTest()`
instead of complex objects. This keeps subtests lightweight and prevents
serialization errors.
* **Cleaning Up Global Singletons**: We reset telemetry singleton states
on test teardown (using pytest's idiomatic `monkeypatch` fixture). This
ensures metric counters don't leak from one test into another.
* **Fixing Concurrent Mock Conflicts**: We return fresh mocked iterators
for concurrent calls (using `side_effect` instead of `return_value`).
This prevents one thread from exhausting a mock's results before another
thread can read them.
* **Robust Assertion Checks**: We added a helper method
(`_assert_concurrent_transaction_invariants`) that verifies the
*behavior* of concurrent threads (ensuring exactly one thread starts the
transaction while others wait/reuse it), rather than checking fragile
call logs or counting sequential request IDs. This allowed us to safely
run four concurrent tests.

> [!note]
> The long pole in the tent is still the `system` tests, which require
about 30 minutes. It is not as simple as just adding xdist because there
are other factors that limit velocity including the fact that system
tests actually interact with live systems.

Author: chalmerlowe

packages/google-cloud-spanner/noxfile.py

@@ -43,6 +43,7 @@
     "pytest",
     "pytest-cov",
     "pytest-asyncio",
+    "pytest-xdist",
 ]
 MOCK_SERVER_ADDITIONAL_DEPENDENCIES = [
     "google-cloud-testutils",
@@ -240,6 +241,8 @@ def unit(session, protobuf_implementation):
     # Run py.test against the unit tests.
     args = [
         "py.test",
+        "-n",
+        "auto",
         "-s",
         f"--junitxml=unit_{session.python}_sponge_log.xml",
         "--cov=google",
@@ -754,7 +757,6 @@ def prerelease_deps(session, protobuf_implementation, database_dialect):
 def mypy(session):
     """Run the type checker."""
     session.skip("Mypy is not yet supported")
-
     # TODO(https://github.com/googleapis/gapic-generator-python/issues/2579):
     # use the latest version of mypy
     session.install(
@@ -832,12 +834,15 @@ def core_deps_from_source(session, protobuf_implementation):
     dep_paths = [str(deps_dir / dep) for dep in core_dependencies_from_source]
 
     session.install(*dep_paths, "--no-deps", "--ignore-installed")
+    session.install("pytest-xdist")
     print(
         f"Installed {', '.join(core_dependencies_from_source)} locally from {deps_dir}"
     )
 
     session.run(
         "py.test",
+        "-n",
+        "auto",
         "tests/unit",
         env={
             "PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION": protobuf_implementation,

packages/google-cloud-spanner/setup.py

@@ -60,7 +60,17 @@
     "google-cloud-monitoring >= 2.16.0",
     "mmh3 >= 4.1.0",
 ]
-extras = {"libcst": "libcst >= 0.2.5"}
+extras = {
+    "libcst": "libcst >= 0.2.5",
+    "test": [
+        "pytest",
+        "mock",
+        "asyncmock",
+        "pytest-cov",
+        "pytest-asyncio",
+        "pytest-xdist",
+    ],
+}
 
 url = "https://github.com/googleapis/google-cloud-python/tree/main/packages/google-cloud-spanner"
 

packages/google-cloud-spanner/tests/system/_async/test_database_api.py

@@ -179,7 +179,10 @@ async def _unit_of_work(transaction):
         transaction.insert_or_update(sd.TABLE, sd.COLUMNS, sd.ROW_DATA)
 
     await shared_database.run_in_transaction(_unit_of_work)
-    assert attempts == 2
+    # Expect at least 2 attempts due to our simulated manual abort on first try.
+    # We use >= 2 rather than == 2 because the live Spanner server can also
+    # trigger transient abort retries depending on real-world GCP resource contention.
+    assert attempts >= 2
 
 
 @pytest.mark.asyncio

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

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import datetime
+import os
 import time
 
 import pytest
@@ -25,6 +26,10 @@
 
 from . import _helpers
 
+# Disable builtin metrics for system tests by default to avoid 401 errors
+# from the background thread exporting to Cloud Monitoring without permissions.
+os.environ["SPANNER_DISABLE_BUILTIN_METRICS"] = "true"
+
 
 @pytest.fixture(scope="function")
 def if_create_instance():

packages/google-cloud-spanner/tests/unit/_async/test_client.py

@@ -799,7 +799,9 @@ async def test_constructor_logs_options_disabled_by_default(self):
             info_logger.assert_not_called()
 
         # Also test when the environment variable is not set at all
-        with mock.patch.dict(os.environ, {}, clear=True):
+        with mock.patch.dict(
+            os.environ, {"SPANNER_DISABLE_BUILTIN_METRICS": "true"}, clear=True
+        ):
             with mock.patch.object(logger, "info") as info_logger:
                 client = self._make_one(project=self.PROJECT, credentials=creds)
                 self.assertIsNotNone(client)

packages/google-cloud-spanner/tests/unit/_async/test_client_extra.py

@@ -131,7 +131,21 @@ async def test_sync_branches_admin_apis(self):
                                 self.assertIsNotNone(ia_api)
                                 self.assertIsNotNone(da_api)
 
-    def test_initialize_metrics_double_check(self):
+    # Safety shield mocks: We intercept the OpenTelemetry metric classes at the client module namespace level
+    # to prevent instantiating real exporter objects. This prevents spawning live background worker threads
+    # that periodically wake up and trigger 401 credential errors inside unauthenticated unit test runs.
+    @mock.patch("google.cloud.spanner_v1._async.client.CloudMonitoringMetricsExporter")
+    @mock.patch("google.cloud.spanner_v1._async.client.PeriodicExportingMetricReader")
+    @mock.patch("google.cloud.spanner_v1._async.client.MeterProvider")
+    # Global state reset: Temporarily override the module's process-wide global boolean _metrics_monitor_initialized
+    # to False so that the client enters the initialization logic instead of returning early.
+    @mock.patch(
+        "google.cloud.spanner_v1._async.client._metrics_monitor_initialized",
+        False,
+    )
+    def test_initialize_metrics_double_check(
+        self, mock_provider, mock_reader, mock_exporter
+    ):
         # coverage for line 143->exit
         from google.cloud.spanner_v1._async import client as MUT
 
@@ -147,15 +161,17 @@ def __enter__(self):
             def __exit__(self, *args):
                 return original_lock.__exit__(*args)
 
+        # Concurrency race condition simulator: Replace the process synchronization lock with our custom SettingLock.
+        # When this lock enters, it toggles _metrics_monitor_initialized to True to simulate another thread
+        # completing metrics setup while this thread was waiting for the lock.
         with mock.patch(
-            "google.cloud.spanner_v1._async.client._metrics_monitor_initialized", False
+            "google.cloud.spanner_v1._async.client._metrics_monitor_lock",
+            SettingLock(),
         ):
-            with mock.patch(
-                "google.cloud.spanner_v1._async.client._metrics_monitor_lock",
-                SettingLock(),
-            ):
-                MUT._initialize_metrics("project", self.credentials)
-                self.assertTrue(MUT._metrics_monitor_initialized)
+            # Trigger the initialization function and verify Spanner's double-checked lock safely
+            # checks the flag again and aborts cleanly to prevent dual-registration.
+            MUT._initialize_metrics("project", self.credentials)
+            self.assertTrue(MUT._metrics_monitor_initialized)
 
     def test_default_transaction_options_validation(self):
         # coverage for line 344

packages/google-cloud-spanner/tests/unit/_async/test_session.py

@@ -1,4 +1,5 @@
 import datetime
+import threading
 from datetime import timezone
 
 import google.api_core.gapic_v1.method
@@ -1800,11 +1801,16 @@ async def unit_of_work(txn, *args, **kw):
             called_with.append((txn, args, kw))
             txn.insert(TABLE_NAME, COLUMNS, VALUES)
 
+        main_thread = threading.current_thread()
+        _results = [1, 1.5]
+
         # retry once w/ timeout_secs=1
-        def _time(_results=[1, 1.5]):
-            if len(_results) > 1:
-                return _results.pop(0)
-            return _results[0]
+        def _time():
+            if threading.current_thread() is main_thread:
+                if len(_results) > 1:
+                    return _results.pop(0)
+                return _results[0]
+            return 1.0
 
         with mock.patch("time.time", _time):
             with mock.patch(
@@ -1877,9 +1883,14 @@ async def unit_of_work(txn, *args, **kw):
             called_with.append((txn, args, kw))
             txn.insert(TABLE_NAME, COLUMNS, VALUES)
 
+        main_thread = threading.current_thread()
+        _results = [1] * 100
+
         # retry several times to check backoff
-        def _time(_results=[1] * 100):
-            return _results.pop(0)
+        def _time():
+            if threading.current_thread() is main_thread:
+                return _results.pop(0)
+            return 1.0
 
         with (
             mock.patch("time.time", _time),

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

@@ -14,5 +14,23 @@
 
 import os
 
+import pytest
+
+from google.cloud.spanner_v1.metrics.spanner_metrics_tracer_factory import (
+    SpannerMetricsTracerFactory,
+)
+
 # Disable builtin metrics to avoid background thread noise and 401 errors in unit tests
 os.environ["SPANNER_DISABLE_BUILTIN_METRICS"] = "true"
+
+
+@pytest.fixture(autouse=True)
+def reset_metrics_singletons(monkeypatch):
+    # Reset singletons and env var before test to avoid state pollution
+    monkeypatch.setenv("SPANNER_DISABLE_BUILTIN_METRICS", "true")
+    SpannerMetricsTracerFactory._metrics_tracer_factory = None
+    SpannerMetricsTracerFactory._current_metrics_tracer_ctx.set(None)
+    yield
+    # Reset singletons after test to ensure no leakage
+    SpannerMetricsTracerFactory._metrics_tracer_factory = None
+    SpannerMetricsTracerFactory._current_metrics_tracer_ctx.set(None)

packages/google-cloud-spanner/tests/unit/spanner_dbapi/test_partition_helper.py

@@ -192,7 +192,7 @@ def collect_protobufs(val):
 
         registered_classes = set(partition_helper._PROTO_CLASS_MAP.values())
         for cls in discovered_protobuf_classes:
-            with self.subTest(cls=cls):
+            with self.subTest(cls_name=cls.__name__):
                 self.assertIn(
                     cls,
                     registered_classes,

packages/google-cloud-spanner/tests/unit/test__helpers.py

@@ -329,7 +329,7 @@ def test_w_numeric_precision_and_scale_valid(self):
             decimal.Decimal("1E-9"),
         ]
         for value in cases:
-            with self.subTest(value=value):
+            with self.subTest(value=str(value)):
                 value_pb = self._callFUT(value)
                 self.assertIsInstance(value_pb, Value)
                 self.assertEqual(value_pb.string_value, str(value))
@@ -371,7 +371,7 @@ def test_w_numeric_precision_and_scale_invalid(self):
         ]
 
         for value, err_msg in cases:
-            with self.subTest(value=value, err_msg=err_msg):
+            with self.subTest(value=str(value), err_msg=err_msg):
                 self.assertRaisesRegex(
                     ValueError,
                     err_msg,