test: stop the cross-loop startup tests from racing on their own mock

Both cross-loop startup tests entered mock.patch.object on the shared
plugin instance from inside each worker thread. patch.object swaps and
restores one attribute on one object and is not thread safe: when two
threads read the original before either installs its mock, both record
that the attribute was absent from the instance, and both delete it on
exit. The second delete raises, so the test failed with

  AttributeError: object has no attribute "_lazy_setup"

which is the mock unwinding itself, not anything about coalescing.

Install the mock once from the test thread and let the worker threads
race only on _ensure_started, which is what these tests are for. The
behaviour under test is unchanged: both loops still call in concurrently
and setup still has to coalesce to a single run.

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 953746627

Author: GWeale

tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py

@@ -10460,20 +10460,21 @@ async def fake_lazy_setup(**kwargs):
 
     def run_in_fresh_loop():
       try:
-        with mock.patch.object(
-            plugin, "_lazy_setup", side_effect=fake_lazy_setup
-        ):
-          asyncio.run(plugin._ensure_started())
+        asyncio.run(plugin._ensure_started())
       except BaseException as e:  # noqa: BLE001 - collecting for assertion
         errors.append(e)
 
-    threads = [
-        platform_thread.create_thread(run_in_fresh_loop) for _ in range(2)
-    ]
-    for t in threads:
-      t.start()
-    for t in threads:
-      t.join(timeout=10)
+    # Patch from this thread only. patch.object swaps a single shared
+    # attribute and is not itself thread safe, so entering it from both
+    # threads raced on _lazy_setup instead of on the code under test.
+    with mock.patch.object(plugin, "_lazy_setup", side_effect=fake_lazy_setup):
+      threads = [
+          platform_thread.create_thread(run_in_fresh_loop) for _ in range(2)
+      ]
+      for t in threads:
+        t.start()
+      for t in threads:
+        t.join(timeout=10)
     assert not errors, f"cross-loop startup raised: {errors}"
 
   def test_concurrent_stale_cleanup_folds_once(
@@ -10593,24 +10594,27 @@ async def slow_setup(**kwargs):
 
     def run_in_fresh_loop():
       try:
-        with mock.patch.object(plugin, "_lazy_setup", side_effect=slow_setup):
-          barrier.wait(timeout=5)
-          asyncio.run(plugin._ensure_started())
+        barrier.wait(timeout=5)
+        asyncio.run(plugin._ensure_started())
       except BaseException as e:  # noqa: BLE001
         errors.append(e)
 
-    threads = [
-        platform_thread.create_thread(run_in_fresh_loop) for _ in range(2)
-    ]
-    for t in threads:
-      t.start()
-    # Deterministic rendezvous: hold the owner inside setup until BOTH
-    # threads have entered _ensure_started.
-    entered.wait(timeout=5)
-    release.set()
-    for t in threads:
-      t.join(timeout=10)
-      assert not t.is_alive(), "thread failed to terminate"
+    # Patch from this thread only. patch.object swaps a single shared
+    # attribute and is not itself thread safe, so entering it from both
+    # threads raced on _lazy_setup instead of on the code under test.
+    with mock.patch.object(plugin, "_lazy_setup", side_effect=slow_setup):
+      threads = [
+          platform_thread.create_thread(run_in_fresh_loop) for _ in range(2)
+      ]
+      for t in threads:
+        t.start()
+      # Deterministic rendezvous: hold the owner inside setup until BOTH
+      # threads have entered _ensure_started.
+      entered.wait(timeout=5)
+      release.set()
+      for t in threads:
+        t.join(timeout=10)
+        assert not t.is_alive(), "thread failed to terminate"
 
     assert not errors, f"cross-loop startup raised: {errors}"
     assert len(setup_calls) == 1, f"shared setup ran {len(setup_calls)} times"