diff --git a/.github/trigger_files/beam_PostCommit_Python_Versions.json b/.github/trigger_files/beam_PostCommit_Python_Versions.json index 8b2c8c445c1f..b5704c67ef1c 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 da88ffb54760..9f750e2ff58c 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 @@ _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): diff --git a/sdks/python/apache_beam/dataframe/io_test.py b/sdks/python/apache_beam/dataframe/io_test.py index 051a85b379f7..d9723e1e5bb5 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 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"), diff --git a/sdks/python/apache_beam/io/requestresponse_it_test.py b/sdks/python/apache_beam/io/requestresponse_it_test.py index 072a1dc27b9f..4e4c2b638441 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 @@ 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() diff --git a/sdks/python/apache_beam/testing/test_utils.py b/sdks/python/apache_beam/testing/test_utils.py index 425b4c6dfac4..ee953a902a0d 100644 --- a/sdks/python/apache_beam/testing/test_utils.py +++ b/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): diff --git a/sdks/python/apache_beam/testing/test_utils_test.py b/sdks/python/apache_beam/testing/test_utils_test.py index 9c379cf35aab..3b99687c1158 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 @@ 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 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 04db85a75c29..644dd56978bc 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 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 diff --git a/sdks/python/conftest.py b/sdks/python/conftest.py index 92d90769fb15..0b2db2649a99 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(