This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 1f76c9c36d6 Fix Python unit tests broken by google-api-core arriving
via cloud-sql-python-connector 1.23.0 (#40623)
1f76c9c36d6 is described below
commit 1f76c9c36d642f58ca81b6b1fb757411771323cb
Author: Shunping Huang <[email protected]>
AuthorDate: Fri Oct 16 10:45:22 2026 -0400
Fix Python unit tests broken by google-api-core arriving via
cloud-sql-python-connector 1.23.0 (#40623)
---
.../beam_PostCommit_Python_Versions.json | 2 +-
sdks/python/apache_beam/dataframe/io_it_test.py | 7 ++--
sdks/python/apache_beam/dataframe/io_test.py | 11 ++-----
.../apache_beam/io/requestresponse_it_test.py | 5 +++
sdks/python/apache_beam/testing/test_utils.py | 25 ++++++++++++++
sdks/python/apache_beam/testing/test_utils_test.py | 37 +++++++++++++++++++++
.../enrichment_handlers/cloudsql_it_test.py | 3 ++
sdks/python/conftest.py | 38 +++++++++++++++++-----
8 files changed, 106 insertions(+), 22 deletions(-)
diff --git a/.github/trigger_files/beam_PostCommit_Python_Versions.json
b/.github/trigger_files/beam_PostCommit_Python_Versions.json
index 8b2c8c445c1..b5704c67ef1 100644
--- a/.github/trigger_files/beam_PostCommit_Python_Versions.json
+++ b/.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
}
diff --git a/sdks/python/apache_beam/dataframe/io_it_test.py
b/sdks/python/apache_beam/dataframe/io_it_test.py
index da88ffb5476..9f750e2ff58 100644
--- a/sdks/python/apache_beam/dataframe/io_it_test.py
+++ b/sdks/python/apache_beam/dataframe/io_it_test.py
@@ -33,13 +33,12 @@ from apache_beam.testing.util import equal_to
_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
[email protected](
- GoogleAPICallError is None, 'GCP dependencies are not installed')
[email protected](HttpError is None, 'GCP dependencies are not installed')
class ReadUsingReadGbqTests(unittest.TestCase):
@pytest.mark.it_postcommit
def test_ReadGbq(self):
diff --git a/sdks/python/apache_beam/dataframe/io_test.py
b/sdks/python/apache_beam/dataframe/io_test.py
index 051a85b379f..d9723e1e5bb 100644
--- a/sdks/python/apache_beam/dataframe/io_test.py
+++ b/sdks/python/apache_beam/dataframe/io_test.py
@@ -46,9 +46,9 @@ from apache_beam.testing.util import assert_that
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 @@ X , c1, c2
set(self.read_all_lines(output + 'out2.csv*')))
[email protected](
- GoogleAPICallError is None, 'GCP dependencies are not installed')
[email protected](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"),
diff --git a/sdks/python/apache_beam/io/requestresponse_it_test.py
b/sdks/python/apache_beam/io/requestresponse_it_test.py
index 072a1dc27b9..4e4c2b63844 100644
--- a/sdks/python/apache_beam/io/requestresponse_it_test.py
+++ b/sdks/python/apache_beam/io/requestresponse_it_test.py
@@ -30,6 +30,7 @@ import apache_beam as beam
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 @@ class FakeCallerForCache(Caller[str, str]):
@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()
diff --git a/sdks/python/apache_beam/testing/test_utils.py
b/sdks/python/apache_beam/testing/test_utils.py
index 425b4c6dfac..ee953a902a0 100644
--- a/sdks/python/apache_beam/testing/test_utils.py
+++ b/sdks/python/apache_beam/testing/test_utils.py
@@ -31,6 +31,7 @@ import shutil
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):
diff --git a/sdks/python/apache_beam/testing/test_utils_test.py
b/sdks/python/apache_beam/testing/test_utils_test.py
index 9c379cf35aa..3b99687c115 100644
--- a/sdks/python/apache_beam/testing/test_utils_test.py
+++ b/sdks/python/apache_beam/testing/test_utils_test.py
@@ -126,6 +126,43 @@ class BlockImportsTest(unittest.TestCase):
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
diff --git
a/sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py
b/sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py
index 04db85a75c2..644dd56978b 100644
--- a/sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py
+++ b/sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py
@@ -28,6 +28,7 @@ import pytest
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 @@ class BaseExternalSQLDBEnrichment(BaseTestSQLEnrichment):
# 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
diff --git a/sdks/python/conftest.py b/sdks/python/conftest.py
index 92d90769fb1..0b2db2649a9 100644
--- a/sdks/python/conftest.py
+++ b/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(
