From b34b69a694916485fbde99ba9e66415d78268d26 Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Mon, 23 Oct 2023 16:14:35 -0700 Subject: [PATCH 01/11] [YAML] Basic integration testing framework. The tests themselves are defined in yaml as a series of pipelines, possibly with some setup code. One of the key features of this framework is that if multiple providers vend the same transform each will be tested to ensure they have consistent behavior. --- sdks/python/apache_beam/testing/util.py | 4 +- .../apache_beam/yaml/integration_tests.py | 173 ++++++++++++++++++ sdks/python/apache_beam/yaml/tests/csv.yaml | 33 ++++ sdks/python/apache_beam/yaml/tests/json.yaml | 33 ++++ sdks/python/apache_beam/yaml/yaml_provider.py | 12 ++ 5 files changed, 253 insertions(+), 2 deletions(-) create mode 100644 sdks/python/apache_beam/yaml/integration_tests.py create mode 100644 sdks/python/apache_beam/yaml/tests/csv.yaml create mode 100644 sdks/python/apache_beam/yaml/tests/json.yaml diff --git a/sdks/python/apache_beam/testing/util.py b/sdks/python/apache_beam/testing/util.py index 10a7a8e86f94..cffafa6c0740 100644 --- a/sdks/python/apache_beam/testing/util.py +++ b/sdks/python/apache_beam/testing/util.py @@ -301,12 +301,12 @@ def expand(self, pcoll): if not use_global_window: plain_actual = plain_actual | 'AddWindow' >> ParDo(AddWindow()) - plain_actual = plain_actual | 'Match' >> Map(matcher) + return plain_actual | 'Match' >> Map(matcher) def default_label(self): return label - actual | AssertThat() # pylint: disable=expression-not-assigned + return actual | AssertThat() @ptransform_fn diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py new file mode 100644 index 000000000000..97843a34d126 --- /dev/null +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -0,0 +1,173 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Runs integration tests in the tests directory.""" + +import argparse +import contextlib +import copy +import glob +import itertools +import logging +import mock +import os +import random +import re +import sys +import tempfile +import uuid +import unittest + +import yaml +from yaml.loader import SafeLoader + +import apache_beam as beam +from apache_beam.io import filesystems +from apache_beam.options.pipeline_options import PipelineOptions +from apache_beam.typehints import trivial_inference +from apache_beam.yaml import yaml_mapping +from apache_beam.yaml import yaml_provider +from apache_beam.yaml import yaml_transform + +from apache_beam.utils import python_callable + + +@contextlib.contextmanager +def gcs_temp_dir(bucket): + gcs_tempdir = bucket + '/yaml-' + str(uuid.uuid4()) + yield gcs_tempdir + filesystems.FileSystems.delete([gcs_tempdir]) + + +def replace_recursive(spec, vars): + if isinstance(spec, dict): + return { + key: replace_recursive(value, vars) + for (key, value) in spec.items() + } + elif isinstance(spec, list): + return [replace_recursive(value, vars) for value in spec] + elif isinstance(spec, str) and '{' in spec: + try: + return eval('f' + repr(spec), vars) + except Exception as exn: + raise ValueError(f"Error evaluating {spec}: {exn}") from exn + else: + return spec + + +def transform_types(spec): + if spec.get('type', None) in (None, 'composite', 'chain'): + if 'source' in spec: + yield from transform_types(spec['source']) + for t in spec.get('transforms', []): + yield from transform_types(t) + if 'sink' in spec: + yield from transform_types(spec['sink']) + else: + yield spec['type'] + + +def provider_sets(spec, require_available=False): + """For transforms that are vended by multiple providers, yields all possible + combinations of providers to use. + """ + all_transform_types = set.union( + *( + set( + transform_types( + yaml_transform.preprocess(copy.deepcopy(p['pipeline'])))) + for p in spec['pipelines'])) + + def filter_to_available(t, providers): + if require_available: + for p in providers: + if not p.available(): + raise ValueError("Provider {p} required for {t} is not available.") + return providers + else: + return [p for p in providers if p.available()] + + standard_providers = yaml_provider.standard_providers() + multiple_providers = { + t: filter_to_available(t, standard_providers[t]) + for t in all_transform_types + if len(filter_to_available(t, standard_providers[t])) > 1 + } + if not multiple_providers: + return None, standard_providers + else: + names, provider_lists = zip(*sorted(multiple_providers.items())) + for ix, c in enumerate(itertools.product(*provider_lists)): + yield ( + '_'.join( + n + '_' + type(p.underlying_provider()).__name__ + for (n, p) in zip(names, c)) + f'_{ix}', + dict(standard_providers, **{n: [p] + for (n, p) in zip(names, c)})) + + +def create_test_methods(spec): + for suffix, providers in provider_sets(spec): + + def test(self, providers=providers): # default arg to capture loop value + vars = {} + with contextlib.ExitStack() as stack: + stack.enter_context( + mock.patch( + 'apache_beam.yaml.yaml_provider.standard_providers', + lambda: providers)) + for fixture in spec.get('fixtures', []): + vars[fixture['name']] = stack.enter_context( + python_callable.PythonCallableWithSource. + load_from_fully_qualified_name(fixture['type'])( + **yaml_transform.SafeLineLoader.strip_metadata( + fixture.get('config', {})))) + for pipeline_spec in spec['pipelines']: + with beam.Pipeline(options=PipelineOptions( + pickle_library='cloudpickle', + **yaml_transform.SafeLineLoader.strip_metadata(pipeline_spec.get( + 'options', {})))) as p: + yaml_transform.expand_pipeline( + p, replace_recursive(pipeline_spec, vars)) + + yield suffix, test + + +def parse_test_files(filepattern): + for path in glob.glob(filepattern): + with open(path) as fin: + base_name = f'test_{os.path.basename(path)}'.replace('.', '_') + for suffix, func in create_test_methods( + yaml.load(fin, Loader=yaml_transform.SafeLineLoader)): + if suffix: + yield f'{base_name}_{suffix}', func + else: + yield base_name, func + + +def createTestSuite(name, filepattern): + return type(name, (unittest.TestCase, ), dict(parse_test_files(filepattern))) + + +IntegrationTests = createTestSuite( + 'IntegrationTests', + os.path.join(os.path.dirname(__file__), 'tests', '*.yaml')) + +if __name__ == '__main__': + logging.getLogger().setLevel(logging.INFO) + unittest.main(argv=unknown_args) diff --git a/sdks/python/apache_beam/yaml/tests/csv.yaml b/sdks/python/apache_beam/yaml/tests/csv.yaml new file mode 100644 index 000000000000..ea7b6542e6a0 --- /dev/null +++ b/sdks/python/apache_beam/yaml/tests/csv.yaml @@ -0,0 +1,33 @@ +fixtures: + - name: TEMP_DIR + # Need distributed filesystem to be able to read and write from a container. + type: "apache_beam.yaml.integration_tests.gcs_temp_dir" + config: + bucket: "gs://temp-storage-for-end-to-end-tests/temp-it" + +pipelines: + - pipeline: + type: chain + transforms: + - type: Create + config: + elements: + - {label: "11a", rank: 0} + - {label: "37a", rank: 1} + - {label: "389a", rank: 2} + - type: WriteToCsv + config: + path: "{TEMP_DIR}/out.csv" + + - pipeline: + type: chain + transforms: + - type: ReadFromCsv + config: + path: "{TEMP_DIR}/out.csv*" + - type: AssertEqual + config: + elements: + - {label: "11a", rank: 0} + - {label: "37a", rank: 1} + - {label: "389a", rank: 2} diff --git a/sdks/python/apache_beam/yaml/tests/json.yaml b/sdks/python/apache_beam/yaml/tests/json.yaml new file mode 100644 index 000000000000..4d8584c12689 --- /dev/null +++ b/sdks/python/apache_beam/yaml/tests/json.yaml @@ -0,0 +1,33 @@ +fixtures: + - name: TEMP_DIR + # Need distributed filesystem to be able to read and write from a container. + type: "apache_beam.yaml.integration_tests.gcs_temp_dir" + config: + bucket: "gs://temp-storage-for-end-to-end-tests/temp-it" + +pipelines: + - pipeline: + type: chain + transforms: + - type: Create + config: + elements: + - {label: "11a", rank: 0} + - {label: "37a", rank: 1} + - {label: "389a", rank: 2} + - type: WriteToJson + config: + path: "{TEMP_DIR}/out.json" + + - pipeline: + type: chain + transforms: + - type: ReadFromJson + config: + path: "{TEMP_DIR}/out.json*" + - type: AssertEqual + config: + elements: + - {label: "11a", rank: 0} + - {label: "37a", rank: 1} + - {label: "389a", rank: 2} diff --git a/sdks/python/apache_beam/yaml/yaml_provider.py b/sdks/python/apache_beam/yaml/yaml_provider.py index 84399cd597b2..4797e76e1f2b 100644 --- a/sdks/python/apache_beam/yaml/yaml_provider.py +++ b/sdks/python/apache_beam/yaml/yaml_provider.py @@ -44,6 +44,8 @@ import apache_beam.io import apache_beam.transforms.util from apache_beam.portability.api import schema_pb2 +from apache_beam.testing.util import assert_that +from apache_beam.testing.util import equal_to from apache_beam.transforms import external from apache_beam.transforms import window from apache_beam.transforms.fully_qualified_named_transform import FullyQualifiedNamedTransform @@ -585,6 +587,15 @@ def log_and_return(x): logging.info(x) return x + class AssertEqual(beam.PTransform): + def __init__(self, elements): + self._elements = elements + + def expand(self, pcoll): + return assert_that( + pcoll | beam.Map(lambda row: beam.Row(**row._asdict())), + equal_to(dicts_to_rows(self._elements))) + return InlineProvider({ 'Create': create, 'LogForTesting': lambda: beam.Map(log_and_return), @@ -592,6 +603,7 @@ def log_and_return(x): 'WithSchemaExperimental': with_schema, 'Flatten': Flatten, 'WindowInto': WindowInto, + 'AssertEqual': AssertEqual, }, no_input_transforms=('Create', )) From 772fba2827e9e907f0daf0e672500dac942e33ba Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Mon, 23 Oct 2023 17:07:15 -0700 Subject: [PATCH 02/11] Add BQ integration test. --- .../apache_beam/yaml/integration_tests.py | 41 ++++++++----- .../apache_beam/yaml/tests/bigquery.yaml | 60 +++++++++++++++++++ 2 files changed, 85 insertions(+), 16 deletions(-) create mode 100644 sdks/python/apache_beam/yaml/tests/bigquery.yaml diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 97843a34d126..ebf484d22626 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -37,6 +37,8 @@ import apache_beam as beam from apache_beam.io import filesystems +from apache_beam.io.gcp.bigquery_tools import BigQueryWrapper +from apache_beam.io.gcp.internal.clients import bigquery from apache_beam.options.pipeline_options import PipelineOptions from apache_beam.typehints import trivial_inference from apache_beam.yaml import yaml_mapping @@ -53,6 +55,19 @@ def gcs_temp_dir(bucket): filesystems.FileSystems.delete([gcs_tempdir]) +@contextlib.contextmanager +def temp_bigquery_table(project, prefix='yaml_bq_it_'): + bigquery_client = BigQueryWrapper() + dataset_id = '%s_%s' % (prefix, uuid.uuid4().hex) + bigquery_client.get_or_create_dataset(project, dataset_id) + logging.info("Created dataset %s in project %s", dataset_id, project) + yield f'{project}:{dataset_id}.tmp_table' + request = bigquery.BigqueryDatasetsDeleteRequest( + projectId=project, datasetId=dataset_id, deleteContents=True) + logging.info("Deleting dataset %s in project %s", dataset_id, project) + bigquery_client.client.datasets.Delete(request) + + def replace_recursive(spec, vars): if isinstance(spec, dict): return { @@ -109,7 +124,7 @@ def filter_to_available(t, providers): if len(filter_to_available(t, standard_providers[t])) > 1 } if not multiple_providers: - return None, standard_providers + return 'only', standard_providers else: names, provider_lists = zip(*sorted(multiple_providers.items())) for ix, c in enumerate(itertools.product(*provider_lists)): @@ -145,28 +160,22 @@ def test(self, providers=providers): # default arg to capture loop value yaml_transform.expand_pipeline( p, replace_recursive(pipeline_spec, vars)) - yield suffix, test + yield f'test_{suffix}', test def parse_test_files(filepattern): for path in glob.glob(filepattern): with open(path) as fin: - base_name = f'test_{os.path.basename(path)}'.replace('.', '_') - for suffix, func in create_test_methods( - yaml.load(fin, Loader=yaml_transform.SafeLineLoader)): - if suffix: - yield f'{base_name}_{suffix}', func - else: - yield base_name, func - - -def createTestSuite(name, filepattern): - return type(name, (unittest.TestCase, ), dict(parse_test_files(filepattern))) + suite_name = os.path.splitext(os.path.basename(path))[0].title() + 'Test' + print(path, suite_name) + methods = dict( + create_test_methods( + yaml.load(fin, Loader=yaml_transform.SafeLineLoader))) + globals()[suite_name] = type(suite_name, (unittest.TestCase, ), methods) -IntegrationTests = createTestSuite( - 'IntegrationTests', - os.path.join(os.path.dirname(__file__), 'tests', '*.yaml')) +logging.getLogger().setLevel(logging.INFO) +parse_test_files(os.path.join(os.path.dirname(__file__), 'tests', '*.yaml')) if __name__ == '__main__': logging.getLogger().setLevel(logging.INFO) diff --git a/sdks/python/apache_beam/yaml/tests/bigquery.yaml b/sdks/python/apache_beam/yaml/tests/bigquery.yaml new file mode 100644 index 000000000000..ceaa3515e647 --- /dev/null +++ b/sdks/python/apache_beam/yaml/tests/bigquery.yaml @@ -0,0 +1,60 @@ +fixtures: + - name: BQ_TABLE + type: "apache_beam.yaml.integration_tests.temp_bigquery_table" + config: + project: "apache-beam-testing" + - name: TEMP_DIR + # Need distributed filesystem to be able to read and write from a container. + type: "apache_beam.yaml.integration_tests.gcs_temp_dir" + config: + bucket: "gs://temp-storage-for-end-to-end-tests/temp-it" + +pipelines: + - pipeline: + type: chain + transforms: + - type: Create + config: + elements: + - {label: "11a", rank: 0} + - {label: "37a", rank: 1} + - {label: "389a", rank: 2} + - type: WriteToBigQuery + config: + table: "{BQ_TABLE}" + options: + project: "apache-beam-testing" + temp_location: "{TEMP_DIR}" + + - pipeline: + type: chain + transforms: + - type: ReadFromBigQuery + config: + table: "{BQ_TABLE}" + - type: AssertEqual + config: + elements: + - {label: "11a", rank: 0} + - {label: "37a", rank: 1} + - {label: "389a", rank: 2} + options: + project: "apache-beam-testing" + temp_location: "{TEMP_DIR}" + + - pipeline: + type: chain + transforms: + - type: ReadFromBigQuery + config: + table: "{BQ_TABLE}" + fields: ["label"] + row_restriction: "rank > 0" + - type: AssertEqual + config: + elements: + - {label: "37a"} + - {label: "389a"} + options: + project: "apache-beam-testing" + temp_location: "{TEMP_DIR}" From 84283a9ddad268253f0225b439ca5c358b5a838f Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Wed, 27 Mar 2024 12:44:09 -0700 Subject: [PATCH 03/11] Fix merge issue. --- sdks/python/apache_beam/yaml/yaml_provider.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/yaml/yaml_provider.py b/sdks/python/apache_beam/yaml/yaml_provider.py index 253cceb476b3..bf1238bd84f9 100755 --- a/sdks/python/apache_beam/yaml/yaml_provider.py +++ b/sdks/python/apache_beam/yaml/yaml_provider.py @@ -822,7 +822,7 @@ def log_and_return(x): @staticmethod def create_builtin_provider(): return InlineProvider({ - 'AssertEqual': AssertEqual, + 'AssertEqual': YamlProviders.AssertEqual, 'Create': YamlProviders.create, 'LogForTesting': YamlProviders.log_for_testing, 'PyTransform': YamlProviders.fully_qualified_named_transform, From ddc42c377f1cf387a1189a59da357273db53d4bc Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Wed, 27 Mar 2024 12:44:36 -0700 Subject: [PATCH 04/11] Apache licence headers. --- .../python/apache_beam/yaml/tests/bigquery.yaml | 17 +++++++++++++++++ sdks/python/apache_beam/yaml/tests/csv.yaml | 17 +++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/sdks/python/apache_beam/yaml/tests/bigquery.yaml b/sdks/python/apache_beam/yaml/tests/bigquery.yaml index ceaa3515e647..208f71ddc119 100644 --- a/sdks/python/apache_beam/yaml/tests/bigquery.yaml +++ b/sdks/python/apache_beam/yaml/tests/bigquery.yaml @@ -1,3 +1,20 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the# Row(word='License'); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an# Row(word='AS IS' BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + fixtures: - name: BQ_TABLE type: "apache_beam.yaml.integration_tests.temp_bigquery_table" diff --git a/sdks/python/apache_beam/yaml/tests/csv.yaml b/sdks/python/apache_beam/yaml/tests/csv.yaml index ea7b6542e6a0..e37b13d2ee13 100644 --- a/sdks/python/apache_beam/yaml/tests/csv.yaml +++ b/sdks/python/apache_beam/yaml/tests/csv.yaml @@ -1,3 +1,20 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the# Row(word='License'); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an# Row(word='AS IS' BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + fixtures: - name: TEMP_DIR # Need distributed filesystem to be able to read and write from a container. From fa4fd1e486db2958cfd81edef847e8c2c6eb3776 Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Wed, 27 Mar 2024 12:54:43 -0700 Subject: [PATCH 05/11] Apache licence headers. --- sdks/python/apache_beam/yaml/tests/json.yaml | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/sdks/python/apache_beam/yaml/tests/json.yaml b/sdks/python/apache_beam/yaml/tests/json.yaml index 4d8584c12689..aa2bc479467e 100644 --- a/sdks/python/apache_beam/yaml/tests/json.yaml +++ b/sdks/python/apache_beam/yaml/tests/json.yaml @@ -1,3 +1,20 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the# Row(word='License'); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an# Row(word='AS IS' BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + fixtures: - name: TEMP_DIR # Need distributed filesystem to be able to read and write from a container. From c87ea71d587a47fda80aa0edd275f73fb229396e Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Wed, 27 Mar 2024 12:55:42 -0700 Subject: [PATCH 06/11] yapf --- sdks/python/apache_beam/yaml/yaml_provider.py | 1 - 1 file changed, 1 deletion(-) diff --git a/sdks/python/apache_beam/yaml/yaml_provider.py b/sdks/python/apache_beam/yaml/yaml_provider.py index bf1238bd84f9..605d4d0eb618 100755 --- a/sdks/python/apache_beam/yaml/yaml_provider.py +++ b/sdks/python/apache_beam/yaml/yaml_provider.py @@ -565,7 +565,6 @@ def expand(self, pcoll): pcoll | beam.Map(lambda row: beam.Row(**row._asdict())), equal_to(dicts_to_rows(self._elements))) - @staticmethod def create(elements: Iterable[Any], reshuffle: Optional[bool] = True): """Creates a collection containing a specified set of elements. From 672c85be9e064089d7669acc618d7f7fdcbda148 Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Wed, 27 Mar 2024 13:41:48 -0700 Subject: [PATCH 07/11] lint --- sdks/python/apache_beam/yaml/integration_tests.py | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index ebf484d22626..238d2295ea8e 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -17,7 +17,6 @@ """Runs integration tests in the tests directory.""" -import argparse import contextlib import copy import glob @@ -25,23 +24,17 @@ import logging import mock import os -import random -import re -import sys import tempfile import uuid import unittest import yaml -from yaml.loader import SafeLoader import apache_beam as beam from apache_beam.io import filesystems from apache_beam.io.gcp.bigquery_tools import BigQueryWrapper from apache_beam.io.gcp.internal.clients import bigquery from apache_beam.options.pipeline_options import PipelineOptions -from apache_beam.typehints import trivial_inference -from apache_beam.yaml import yaml_mapping from apache_beam.yaml import yaml_provider from apache_beam.yaml import yaml_transform @@ -78,7 +71,7 @@ def replace_recursive(spec, vars): return [replace_recursive(value, vars) for value in spec] elif isinstance(spec, str) and '{' in spec: try: - return eval('f' + repr(spec), vars) + return spec.format(**vars) except Exception as exn: raise ValueError(f"Error evaluating {spec}: {exn}") from exn else: From 767beceafd4761f7bfe9686bcbeefa0ae65b22d3 Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Wed, 27 Mar 2024 15:17:34 -0700 Subject: [PATCH 08/11] use local tempdir --- sdks/python/apache_beam/yaml/integration_tests.py | 4 +--- sdks/python/apache_beam/yaml/tests/csv.yaml | 5 +---- sdks/python/apache_beam/yaml/tests/json.yaml | 5 +---- 3 files changed, 3 insertions(+), 11 deletions(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 238d2295ea8e..66777d251db6 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -24,7 +24,6 @@ import logging import mock import os -import tempfile import uuid import unittest @@ -35,11 +34,10 @@ from apache_beam.io.gcp.bigquery_tools import BigQueryWrapper from apache_beam.io.gcp.internal.clients import bigquery from apache_beam.options.pipeline_options import PipelineOptions +from apache_beam.utils import python_callable from apache_beam.yaml import yaml_provider from apache_beam.yaml import yaml_transform -from apache_beam.utils import python_callable - @contextlib.contextmanager def gcs_temp_dir(bucket): diff --git a/sdks/python/apache_beam/yaml/tests/csv.yaml b/sdks/python/apache_beam/yaml/tests/csv.yaml index e37b13d2ee13..3607abb236b7 100644 --- a/sdks/python/apache_beam/yaml/tests/csv.yaml +++ b/sdks/python/apache_beam/yaml/tests/csv.yaml @@ -17,10 +17,7 @@ fixtures: - name: TEMP_DIR - # Need distributed filesystem to be able to read and write from a container. - type: "apache_beam.yaml.integration_tests.gcs_temp_dir" - config: - bucket: "gs://temp-storage-for-end-to-end-tests/temp-it" + type: "tempfile.TemporaryDirectory" pipelines: - pipeline: diff --git a/sdks/python/apache_beam/yaml/tests/json.yaml b/sdks/python/apache_beam/yaml/tests/json.yaml index aa2bc479467e..dbf039f4735d 100644 --- a/sdks/python/apache_beam/yaml/tests/json.yaml +++ b/sdks/python/apache_beam/yaml/tests/json.yaml @@ -17,10 +17,7 @@ fixtures: - name: TEMP_DIR - # Need distributed filesystem to be able to read and write from a container. - type: "apache_beam.yaml.integration_tests.gcs_temp_dir" - config: - bucket: "gs://temp-storage-for-end-to-end-tests/temp-it" + type: "tempfile.TemporaryDirectory" pipelines: - pipeline: From 6eca0192905f73b864b2bb518b1c936d935241ae Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Wed, 27 Mar 2024 15:19:34 -0700 Subject: [PATCH 09/11] mypy --- sdks/python/apache_beam/yaml/integration_tests.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 66777d251db6..a73cd693ae72 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -170,4 +170,4 @@ def parse_test_files(filepattern): if __name__ == '__main__': logging.getLogger().setLevel(logging.INFO) - unittest.main(argv=unknown_args) + unittest.main() From 31599c262553377da6a90056d7e4059814e1d85f Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Thu, 28 Mar 2024 11:16:13 -0700 Subject: [PATCH 10/11] isort --- sdks/python/apache_beam/yaml/integration_tests.py | 2 +- sdks/python/apache_beam/yaml/yaml_provider.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index a73cd693ae72..bdf1b83263fd 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -22,11 +22,11 @@ import glob import itertools import logging -import mock import os import uuid import unittest +import mock import yaml import apache_beam as beam diff --git a/sdks/python/apache_beam/yaml/yaml_provider.py b/sdks/python/apache_beam/yaml/yaml_provider.py index 605d4d0eb618..127d532b54bb 100755 --- a/sdks/python/apache_beam/yaml/yaml_provider.py +++ b/sdks/python/apache_beam/yaml/yaml_provider.py @@ -46,9 +46,9 @@ import apache_beam.io import apache_beam.transforms.util from apache_beam.portability.api import schema_pb2 +from apache_beam.runners import pipeline_context from apache_beam.testing.util import assert_that from apache_beam.testing.util import equal_to -from apache_beam.runners import pipeline_context from apache_beam.transforms import external from apache_beam.transforms import window from apache_beam.transforms.fully_qualified_named_transform import FullyQualifiedNamedTransform From 10e05356dc43301a3ac32ac21f705871876049a2 Mon Sep 17 00:00:00 2001 From: Robert Bradshaw Date: Thu, 28 Mar 2024 16:12:43 -0700 Subject: [PATCH 11/11] isort --- sdks/python/apache_beam/yaml/integration_tests.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index bdf1b83263fd..19c22d1c6d84 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -23,8 +23,8 @@ import itertools import logging import os -import uuid import unittest +import uuid import mock import yaml