test: make unit contracts platform neutral

Pre-emptive: every CI job is ubuntu-latest, so none of these tests fail today.
No assertion is weakened - each replacement is equivalent or stricter on Linux.

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

Author: GWeale

tests/unittests/artifacts/test_artifact_service.py

@@ -26,8 +26,8 @@
 from typing import Union
 from unittest import mock
 from unittest.mock import patch
-from urllib.parse import unquote
 from urllib.parse import urlparse
+from urllib.request import url2pathname
 
 from google.adk.artifacts import file_artifact_service
 from google.adk.artifacts.base_artifact_service import ArtifactVersion
@@ -892,7 +892,7 @@ async def test_file_metadata_camelcase(tmp_path, artifact_service_factory):
       "customMetadata": {},
   }
   parsed_canonical = urlparse(metadata["canonicalUri"])
-  canonical_path = Path(unquote(parsed_canonical.path))
+  canonical_path = Path(url2pathname(parsed_canonical.path))
   assert canonical_path.name == "report.txt"
   assert canonical_path.read_bytes() == b"binary-content"
 
@@ -942,7 +942,7 @@ async def test_file_list_artifact_versions(tmp_path, artifact_service_factory):
   assert version_meta.canonical_uri == version_payload_path.as_uri()
   assert version_meta.custom_metadata == custom_metadata
   parsed_version_uri = urlparse(version_meta.canonical_uri)
-  version_uri_path = Path(unquote(parsed_version_uri.path))
+  version_uri_path = Path(url2pathname(parsed_version_uri.path))
   assert version_uri_path.read_bytes() == b"binary-content"
 
   fetched = await artifact_service.get_artifact_version(

tests/unittests/cli/test_adk_web_server_tests.py

@@ -127,7 +127,8 @@ def test_rebuild_single_test(test_client):
     assert response.json() == {"status": "success"}
     mock_to_thread.assert_called_once()
     args, kwargs = mock_to_thread.call_args
-    assert args[1].endswith("tests/my_test.json")
+    test_dir, test_name = os.path.split(args[1])
+    assert (os.path.basename(test_dir), test_name) == ("tests", "my_test.json")
 
 
 def test_run_tests(test_client):

tests/unittests/code_executors/test_code_execution_utils.py

@@ -12,11 +12,53 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import signal
+import multiprocessing
+import time
+import traceback
 
 from google.adk.code_executors import code_execution_utils
 from google.genai import types
 
+# The extraction itself must finish promptly. The join budget is far looser
+# because it also covers spawning the child and importing this module there.
+_REDOS_DEADLINE_SECONDS = 2.0
+_CHILD_JOIN_TIMEOUT_SECONDS = 120.0
+
+
+def _exercise_redos_candidate(result_conn) -> None:
+  """Runs the ReDoS regression payload in an independently stoppable process."""
+  failure = None
+  try:
+    ticks = "`" * 3
+    long_invalid_payload = (
+        ticks + "python\n" + "x = 1\n" * 5000 + "not_matching"
+    )
+    content = types.Content(
+        role="model",
+        parts=[types.Part(text=long_invalid_payload)],
+    )
+    delimiters = [(ticks + "python\n", "\n" + ticks)]
+
+    started = time.perf_counter()
+    code = code_execution_utils.CodeExecutionUtils.extract_code_and_truncate_content(
+        content, delimiters
+    )
+    elapsed = time.perf_counter() - started
+
+    if code is not None:
+      failure = f"expected no code to be extracted, got {code!r}"
+    elif elapsed > _REDOS_DEADLINE_SECONDS:
+      failure = (
+          f"extraction took {elapsed:.3f}s, over the"
+          f" {_REDOS_DEADLINE_SECONDS}s deadline (possible ReDoS regression)"
+      )
+  except BaseException:  # pylint: disable=broad-except
+    # Without this the parent only sees a bare exit code and has to dig the
+    # traceback out of the child's captured stderr.
+    failure = f"extraction raised in the child:\n{traceback.format_exc()}"
+  result_conn.send(failure)
+  result_conn.close()
+
 
 def test_extract_code_and_truncate_content_basic():
   """Tests basic code extraction and content truncation."""
@@ -94,29 +136,30 @@ def test_extract_code_and_truncate_content_no_delimiter():
 
 def test_extract_code_and_truncate_content_redos_vulnerability():
   """Tests that a string that would cause ReDoS behaves reasonably."""
-  # Construct a long string that contains repeating patterns without matching delimiters.
-  # The old regex pattern would backtrack exponentially.
-  ticks = "`" * 3
-  long_invalid_payload = ticks + "python\n" + "x = 1\n" * 5000 + "not_matching"
-  content = types.Content(
-      role="model",
-      parts=[types.Part(text=long_invalid_payload)],
-  )
-  delimiters = [(ticks + "python\n", "\n" + ticks)]
-
-  def handler(_signum, _frame):
-    raise TimeoutError("Test timed out (possible ReDoS regression)")
-
-  signal.signal(signal.SIGALRM, handler)
-  signal.alarm(2)
+  context = multiprocessing.get_context("spawn")
+  receiver, sender = context.Pipe(duplex=False)
+  process = context.Process(target=_exercise_redos_candidate, args=(sender,))
+  process.start()
+  sender.close()
+  process.join(timeout=_CHILD_JOIN_TIMEOUT_SECONDS)
+  hung = process.is_alive()
+  if hung:
+    process.kill()
+    process.join()
+  exitcode = process.exitcode
   try:
-    # If ReDoS vulnerability exists, this call will hang or take a very long time.
-    code = code_execution_utils.CodeExecutionUtils.extract_code_and_truncate_content(
-        content, delimiters
-    )
+    # poll() is also true at EOF, so recv() has to carry the "child died
+    # without reporting" case rather than a poll() guard.
+    failure = receiver.recv()
+  except EOFError:
+    failure = "child exited without reporting a result"
   finally:
-    signal.alarm(0)
-  assert code is None
+    receiver.close()
+    process.close()
+
+  assert not hung, "extraction never returned (possible ReDoS regression)"
+  assert failure is None, failure
+  assert exitcode == 0, f"extraction process exited with {exitcode}"
 
 
 def test_extract_code_and_truncate_content_multiple_delimiter_pairs():

tests/unittests/sessions/migration/test_migration.py

@@ -15,6 +15,7 @@
 
 from __future__ import annotations
 
+import contextlib
 from datetime import datetime
 from datetime import timezone
 import os
@@ -368,24 +369,69 @@ def test_migrate_from_sqlalchemy_pickle_ignores_non_object_json_fields():
   assert event.content is None
 
 
-def test_migrate_from_sqlalchemy_pickle_reads_naive_timestamp_as_local(
-    monkeypatch,
-):
+@contextlib.contextmanager
+def _pinned_local_timezone(name: str):
+  """Pins the process timezone for the duration of the block.
+
+  ``time.tzset`` is POSIX-only, so on other platforms the block runs in the
+  host zone instead. Restoring ``TZ`` without a second ``tzset`` would leave
+  the C library pinned for the rest of the session, so both are undone.
+  """
+  if not hasattr(time, "tzset"):
+    yield
+    return
+  previous = os.environ.get("TZ")
+  os.environ["TZ"] = name
+  time.tzset()
+  try:
+    yield
+  finally:
+    if previous is None:
+      os.environ.pop("TZ", None)
+    else:
+      os.environ["TZ"] = previous
+    time.tzset()
+
+
+def test_migrate_from_sqlalchemy_pickle_reads_naive_timestamp_as_local():
   """Naive v0 event timestamps must migrate as local time, not UTC.
 
   The v0 schema stored the event ``timestamp`` column as a naive datetime in
   local time (``StorageEvent.from_event`` uses ``datetime.fromtimestamp`` and
   ``to_event`` reads it back with naive ``.timestamp()``). Forcing UTC on that
-  naive value shifted every migrated timestamp by the host's UTC offset. Pin a
-  fixed non-UTC zone so the round trip is exact regardless of the host.
+  naive value shifted every migrated timestamp by the host's UTC offset.
   """
-  monkeypatch.setenv("TZ", "Asia/Kolkata")
-  time.tzset()
-  try:
-    original_epoch = 1000000.0
+  original_epoch = 1000000.0
+
+  class NaiveLocalDatetime(datetime):
+    """Local naive datetime that rejects a timezone being forced onto it.
+
+    ``replace`` and ``astimezone`` return instances of this subclass, so a
+    migration that pins a timezone before reading the epoch back trips the
+    guard even on a host whose local zone is already UTC and where the
+    resulting epoch would be unchanged.
+    """
+
+    def timestamp(self) -> float:
+      assert (
+          self.tzinfo is None
+      ), f"migration forced {self.tzinfo} onto a naive v0 timestamp"
+      return super().timestamp()
+
+  # The pinned zone is what catches a UTC recomputation that arrives by some
+  # other route, e.g. calendar.timegm(), which the guard above cannot see.
+  with _pinned_local_timezone("Asia/Kolkata"):
     # Exactly what v0.StorageEvent.from_event persisted: naive local time.
-    naive_local_timestamp = datetime.fromtimestamp(original_epoch)
-    assert naive_local_timestamp.tzinfo is None
+    local = datetime.fromtimestamp(original_epoch)
+    naive_local_timestamp = NaiveLocalDatetime(
+        local.year,
+        local.month,
+        local.day,
+        local.hour,
+        local.minute,
+        local.second,
+        local.microsecond,
+    )
 
     event = mfsp._row_to_event({
         "id": "event-naive-timestamp",
@@ -396,8 +442,6 @@ def test_migrate_from_sqlalchemy_pickle_reads_naive_timestamp_as_local(
     })
 
     assert event.timestamp == original_epoch
-  finally:
-    time.tzset()
 
 
 def test_migrate_from_sqlalchemy_pickle_blocks_unsafe_actions_pickle(
@@ -528,7 +572,7 @@ def __reduce__(self):
 
 
 def test_migrate_from_sqlalchemy_pickle_with_async_driver_urls(tmp_path):
-  """Tests that migration works with async driver URLs (fixes issue #4176).
+  """Tests that migration works with async driver URLs.
 
   Users often provide async driver URLs (e.g., postgresql+asyncpg://) since
   that's what ADK requires at runtime. The migration tool should handle these
@@ -564,7 +608,7 @@ def test_migrate_from_sqlalchemy_pickle_with_async_driver_urls(tmp_path):
   source_session.commit()
   source_session.close()
 
-  # This should NOT raise an error about async drivers (the fix for #4176)
+  # This should NOT raise an error about async drivers.
   mfsp.migrate(source_db_url, dest_db_url)
 
   # Verify destination DB

tests/unittests/workflow/test_dynamic_node_scheduler.py

@@ -29,7 +29,6 @@
 from google.adk.workflow._dynamic_node_scheduler import DynamicNodeScheduler
 from google.adk.workflow._dynamic_node_scheduler import DynamicNodeState
 from google.adk.workflow._node_state import NodeState
-from google.adk.workflow._node_status import NodeStatus
 from google.adk.workflow._workflow import _LoopState
 from pydantic import BaseModel
 from pydantic import ValidationError
@@ -568,20 +567,21 @@ async def _run_impl(self, *, ctx, node_input):
 
 
 def test_get_dynamic_tasks_excludes_done_tasks():
-  """get_dynamic_tasks should not return completed tasks (regression for #6082)."""
+  """get_dynamic_tasks should not return completed tasks."""
   import asyncio
 
   loop = asyncio.new_event_loop()
+  running_task = None
   try:
 
     async def _done():
       return None
 
-    done_task = loop.run_until_complete(
-        asyncio.ensure_future(_done(), loop=loop)
-    )
-    running_coro = asyncio.sleep(9999)
-    running_task = loop.create_task(running_coro)
+    # run_until_complete returns the coroutine's result, so the run entry has
+    # to hold the task itself for the done-task filter to be exercised at all.
+    done_task = loop.create_task(_done())
+    loop.run_until_complete(done_task)
+    running_task = loop.create_task(asyncio.sleep(9999))
 
     state = DynamicNodeState()
     state.runs['path/done@r-1'] = DynamicNodeRun(
@@ -600,8 +600,14 @@ async def _done():
     tasks = state.get_dynamic_tasks()
 
     assert tasks == [running_task]
-    running_task.cancel()
   finally:
+    # Cancelling without draining leaves the task pending at close() and the
+    # sleep coroutine unawaited, which surfaces as a warning in later tests.
+    if running_task is not None:
+      running_task.cancel()
+      loop.run_until_complete(
+          asyncio.gather(running_task, return_exceptions=True)
+      )
     loop.close()