Fix Python unit tests broken by google-api-core arriving via cloud-sql-python-connector 1.23.0

.github/trigger_files/beam_PostCommit_Python_Versions.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "revision": 5
+  "revision": 6
 }

sdks/python/apache_beam/dataframe/io_it_test.py

@@ -33,13 +33,12 @@
 _LOGGER = logging.getLogger(__name__)
 
 try:
-  from google.api_core.exceptions import GoogleAPICallError
+  from apitools.base.py.exceptions import HttpError
 except ImportError:
-  GoogleAPICallError = None
+  HttpError = None
 
 
-@unittest.skipIf(
-    GoogleAPICallError is None, 'GCP dependencies are not installed')
+@unittest.skipIf(HttpError is None, 'GCP dependencies are not installed')
 class ReadUsingReadGbqTests(unittest.TestCase):
   @pytest.mark.it_postcommit
   def test_ReadGbq(self):

sdks/python/apache_beam/dataframe/io_test.py

@@ -46,9 +46,9 @@
 from apache_beam.testing.util import equal_to
 
 try:
-  from google.api_core.exceptions import GoogleAPICallError
+  from apitools.base.py.exceptions import HttpError
 except ImportError:
-  GoogleAPICallError = None
+  HttpError = None
 
 # Get major, minor version
 PD_VERSION = tuple(map(int, pd.__version__.split('.')[0:2]))
@@ -492,15 +492,10 @@ def test_double_write(self):
                           set(self.read_all_lines(output + 'out2.csv*')))
 
 
-@unittest.skipIf(
-    GoogleAPICallError is None, 'GCP dependencies are not installed')
+@unittest.skipIf(HttpError is None, 'GCP dependencies are not installed')
 class ReadGbqTransformTests(unittest.TestCase):
   @mock.patch.object(BigQueryWrapper, 'get_table')
   def test_bad_schema_public_api_direct_read(self, get_table):
-    try:
-      bigquery.TableFieldSchema
-    except AttributeError:
-      raise ValueError('Please install GCP Dependencies.')
     fields = [
         bigquery.TableFieldSchema(name='stn', type='DOUBLE', mode="NULLABLE"),
         bigquery.TableFieldSchema(name='temp', type='FLOAT64', mode="REPEATED"),

sdks/python/apache_beam/io/requestresponse_it_test.py

@@ -30,6 +30,7 @@
 from apache_beam.coders import coders
 from apache_beam.options.pipeline_options import PipelineOptions
 from apache_beam.testing.test_pipeline import TestPipeline
+from apache_beam.testing.test_utils import require_linux_docker
 
 # pylint: disable=ungrouped-imports
 try:
@@ -206,6 +207,10 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
 @pytest.mark.uses_testcontainer
 class TestRedisCache(unittest.TestCase):
+  @classmethod
+  def setUpClass(cls) -> None:
+    require_linux_docker()
+
   def setUp(self) -> None:
     self.retries = 5
     self._start_container()

sdks/python/apache_beam/testing/test_utils.py

@@ -31,6 +31,7 @@
 import struct
 import sys
 import tempfile
+from unittest import SkipTest
 
 import numpy as np
 
@@ -186,6 +187,30 @@ def block_imports(*packages):
     sys.modules.update(hidden_modules)
 
 
+def require_linux_docker():
+  """Raises ``unittest.SkipTest`` unless Docker can run Linux containers.
+
+  Intended for tests that start containers via testcontainers, e.g. from
+  ``setUpClass``. Such tests need a reachable Docker daemon running Linux
+  containers. Some environments have no daemon at all (e.g. GitHub macOS
+  runners), while others only run Windows containers (e.g. GitHub Windows
+  runners).
+  """
+  try:
+    import docker
+    client = docker.from_env()
+    try:
+      os_type = client.info().get('OSType')
+    finally:
+      client.close()
+  except Exception as e:  # pylint: disable=broad-except
+    raise SkipTest(f'Docker is not available: {e}')
+  if os_type != 'linux':
+    raise SkipTest(
+        f'Docker daemon runs {os_type} containers; '
+        'Linux containers are required.')
+
+
 @retry.with_exponential_backoff(
     num_retries=3, retry_filter=retry.retry_on_beam_io_error_filter)
 def delete_files(file_paths):

sdks/python/apache_beam/testing/test_utils_test.py

@@ -126,6 +126,43 @@ def test_block_imports_only_blocks_given_packages(self):
     self.assertEqual(attempted_imports, [])
 
 
+class RequireLinuxDockerTest(unittest.TestCase):
+  def _fake_docker(self, os_type=None, error=None):
+    fake_docker = mock.MagicMock()
+    client = fake_docker.from_env.return_value
+    if error:
+      client.info.side_effect = error
+    else:
+      client.info.return_value = {'OSType': os_type}
+    return fake_docker, client
+
+  def test_linux_docker_does_not_skip(self):
+    fake_docker, client = self._fake_docker(os_type='linux')
+    with mock.patch.dict(sys.modules, {'docker': fake_docker}):
+      utils.require_linux_docker()
+    client.close.assert_called_once()
+
+  def test_windows_docker_skips(self):
+    fake_docker, client = self._fake_docker(os_type='windows')
+    with mock.patch.dict(sys.modules, {'docker': fake_docker}):
+      with self.assertRaisesRegex(unittest.SkipTest, 'windows containers'):
+        utils.require_linux_docker()
+    client.close.assert_called_once()
+
+  def test_unreachable_docker_skips(self):
+    fake_docker, client = self._fake_docker(
+        error=ConnectionError('no such file'))
+    with mock.patch.dict(sys.modules, {'docker': fake_docker}):
+      with self.assertRaisesRegex(unittest.SkipTest, 'Docker is not available'):
+        utils.require_linux_docker()
+    client.close.assert_called_once()
+
+  def test_missing_docker_sdk_skips(self):
+    with utils.block_imports('docker'):
+      with self.assertRaisesRegex(unittest.SkipTest, 'Docker is not available'):
+        utils.require_linux_docker()
+
+
 class LCGeneratorTest(unittest.TestCase):
   Generator = utils.LCGenerator
 

sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py

@@ -28,6 +28,7 @@
 import apache_beam as beam
 from apache_beam.coders import coders
 from apache_beam.testing.test_pipeline import TestPipeline
+from apache_beam.testing.test_utils import require_linux_docker
 from apache_beam.testing.util import assert_that
 from apache_beam.testing.util import equal_to
 
@@ -589,6 +590,8 @@ def setUpClass(cls):
       # Skip setup for the base class.
       raise unittest.SkipTest("Base class - no db_adapter defined")
 
+    require_linux_docker()
+
     # Type hint data from subclasses.
     cls._db_adapter: DatabaseTypeAdapter
 

sdks/python/conftest.py

@@ -18,11 +18,11 @@
 """Pytest configuration and custom hooks."""
 
 import gc
+import math
 import os
 import sys
 import threading
 import time
-from types import SimpleNamespace
 
 import pytest
 
@@ -150,6 +150,32 @@ def enhance_mock_stability():
   time.sleep(0.05)
 
 
+def _configure_testcontainers_timeouts():
+  """Applies TC_* environment settings to testcontainers' global config.
+
+  testcontainers 4.x computes its wait timeout as ``max_tries * sleep_time``
+  from ``testcontainers_config``. It reads ``TC_MAX_TRIES`` and
+  ``TC_POOLING_INTERVAL`` itself, but not ``TC_TIMEOUT`` or ``TC_SLEEP_TIME``,
+  which Beam's CI workflows set. Translate them here so that the effective
+  timeout is at least ``TC_TIMEOUT`` seconds.
+  """
+  from testcontainers.core.config import testcontainers_config
+
+  sleep_time = float(
+      os.getenv("TC_SLEEP_TIME") or os.getenv("TC_POOLING_INTERVAL") or "1")
+  timeout = float(os.getenv("TC_TIMEOUT", "120"))
+  max_tries = int(os.getenv("TC_MAX_TRIES", "0"))
+
+  testcontainers_config.sleep_time = sleep_time
+  testcontainers_config.max_tries = max(
+      max_tries, math.ceil(timeout / sleep_time))
+  print(
+      "Set testcontainers max_tries=%d, sleep_time=%s (timeout=%ss)" % (
+          testcontainers_config.max_tries,
+          testcontainers_config.sleep_time,
+          testcontainers_config.timeout))
+
+
 def pytest_configure(config):
   """Saves options added in pytest_addoption for later use.
   This is necessary since pytest-xdist workers do not have the
@@ -159,14 +185,8 @@ def pytest_configure(config):
   # for the entire test session.
   print("\n--- Applying global testcontainers timeout configuration ---")
   try:
-    from testcontainers.core import waiting_utils
-    waiting_utils.config = SimpleNamespace(
-        timeout=int(os.getenv("TC_TIMEOUT", "120")),
-        max_tries=int(os.getenv("TC_MAX_TRIES", "120")),
-        sleep_time=float(os.getenv("TC_SLEEP_TIME", "1")),
-    )
-    print("Successfully set waiting utils config")
-  except ModuleNotFoundError:
+    _configure_testcontainers_timeouts()
+  except ImportError:
     print("The testcontainers library is not installed.")
 
   TestPipeline.pytest_test_pipeline_options = config.getoption(