You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2025/10/09 19:00:02 UTC
(beam) branch users/damccorm/gbek-xlang-py-tests created (now 5064168fb1d)
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a change to branch users/damccorm/gbek-xlang-py-tests
in repository https://gitbox.apache.org/repos/asf/beam.git
at 5064168fb1d Add some x-lang gbek tests
This branch includes the following new commits:
new 5064168fb1d Add some x-lang gbek tests
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(beam) 01/01: Add some x-lang gbek tests
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch users/damccorm/gbek-xlang-py-tests
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 5064168fb1d491c3aab8265ee865e3a782dd775f
Author: Danny Mccormick <da...@google.com>
AuthorDate: Thu Oct 9 14:59:44 2025 -0400
Add some x-lang gbek tests
---
...am_PostCommit_XVR_PythonUsingJava_Dataflow.json | 4 +
.../transforms/validate_runner_xlang_test.py | 93 ++++++++++++++++++++++
2 files changed, 97 insertions(+)
diff --git a/.github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json b/.github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json
new file mode 100644
index 00000000000..b73af5e61a4
--- /dev/null
+++ b/.github/trigger_files/beam_PostCommit_XVR_PythonUsingJava_Dataflow.json
@@ -0,0 +1,4 @@
+{
+ "comment": "Modify this file in a trivial way to cause this test suite to run.",
+ "modification": 1
+}
diff --git a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
index 8e8e7964825..40f7478fd5d 100644
--- a/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
+++ b/sdks/python/apache_beam/transforms/validate_runner_xlang_test.py
@@ -52,17 +52,27 @@ https://docs.google.com/document/d/1xQp0ElIV84b8OCVz8CD2hvbiWdR8w4BvWxPTZJZA6NA
import logging
import os
+import random
+import string
import typing
import unittest
import pytest
import apache_beam as beam
+from apache_beam.options.pipeline_options import SetupOptions
from apache_beam.testing.test_pipeline import TestPipeline
from apache_beam.testing.util import assert_that
from apache_beam.testing.util import equal_to
+from apache_beam.transforms.util import GcpSecret
+from apache_beam.transforms.util import Secret
from apache_beam.transforms.external import ImplicitSchemaPayloadBuilder
+try:
+ from google.cloud import secretmanager
+except ImportError:
+ secretmanager = None # type: ignore[assignment]
+
TEST_PREFIX_URN = "beam:transforms:xlang:test:prefix"
TEST_MULTI_URN = "beam:transforms:xlang:test:multi"
TEST_GBK_URN = "beam:transforms:xlang:test:gbk"
@@ -140,6 +150,24 @@ class CrossLanguageTestPipelines(object):
| beam.Map(lambda x: "{}:{}".format(x[0], ','.join(sorted(x[1])))))
assert_that(res, equal_to(['0:1,2', '1:3']))
+ def run_group_by_key_no_assert(self, pipeline):
+ """
+ Target transform - GroupByKey, with no assertion for checking errors
+ (https://beam.apache.org/documentation/programming-guide/#groupbykey)
+ Test scenario - Grouping a collection of KV<K,V> to a collection of
+ KV<K, Iterable<V>> by key
+ Boundary conditions checked -
+ - PCollection<KV<?, ?>> to external transforms
+ - PCollection<KV<?, Iterable<?>>> from external transforms
+ """
+ with pipeline as p:
+ (
+ p
+ | beam.Create([(0, "1"), (0, "2"),
+ (1, "3")], reshuffle=False).with_output_types(
+ typing.Tuple[int, str])
+ | beam.ExternalTransform(TEST_GBK_URN, None, self.expansion_service))
+
def run_cogroup_by_key(self, pipeline):
"""
Target transform - CoGroupByKey
@@ -298,6 +326,71 @@ class ValidateRunnerXlangTest(unittest.TestCase):
test_pipeline or self.create_pipeline())
+@unittest.skipUnless(
+ os.environ.get('EXPANSION_PORT'),
+ "EXPANSION_PORT environment var is not provided.")
+@unittest.skipIf(secretmanager is None, 'secretmanager not installed')
+class ValidateRunnerGBEKTest(unittest.TestCase):
+ def setUp(self):
+ if secretmanager is not None:
+ self.project_id = 'apache-beam-testing'
+ secret_postfix = ''.join(random.choice(string.digits) for _ in range(6))
+ self.secret_id = 'gbek_secret_tests_' + secret_postfix
+ self.client = secretmanager.SecretManagerServiceClient()
+ self.project_path = f'projects/{self.project_id}'
+ self.secret_path = f'{self.project_path}/secrets/{self.secret_id}'
+ try:
+ self.client.get_secret(request={'name': self.secret_path})
+ except Exception:
+ self.client.create_secret(
+ request={
+ 'parent': self.project_path,
+ 'secret_id': self.secret_id,
+ 'secret': {
+ 'replication': {
+ 'automatic': {}
+ }
+ }
+ })
+ self.client.add_secret_version(
+ request={
+ 'parent': self.secret_path,
+ 'payload': {
+ 'data': Secret.generate_secret_bytes()
+ }
+ })
+ version_name = f'{self.secret_path}/versions/latest'
+ self.gcp_secret = GcpSecret(version_name)
+ self.secret_option = f'type:GcpSecret;version_name:{version_name}'
+
+ def tearDown(self):
+ if secretmanager is not None:
+ self.client.delete_secret(request={'name': self.secret_path})
+
+ def create_pipeline(self):
+ test_pipeline = TestPipeline()
+ test_pipeline.not_use_test_runner_api = True
+ return test_pipeline
+
+ # This test and test_group_by_key_gbek_bad_secret validate that the gbek
+ # pipeline option is correctly passed through
+ @pytest.mark.uses_java_expansion_service
+ @pytest.mark.uses_python_expansion_service
+ def test_group_by_key_gbek(self, test_pipeline=None):
+ test_pipeline = test_pipeline or self.create_pipeline()
+ good_secret = self.secret_option
+ test_pipeline.options.view_as(SetupOptions).gbek = good_secret
+ CrossLanguageTestPipelines().run_group_by_key(test_pipeline)
+
+ # Verify actually using secret manager
+ test_pipeline = self.create_pipeline()
+ nonexistent_secret = 'version_name:nonexistent_secret'
+ test_pipeline.options.view_as(SetupOptions).gbek = nonexistent_secret
+ with self.assertRaisesRegex(
+ Exception, 'Secret string must contain a valid type parameter'):
+ CrossLanguageTestPipelines().run_group_by_key_no_assert(test_pipeline)
+
+
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
unittest.main()