This is an automated email from the ASF dual-hosted git repository. ccy 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 5d630dc [BEAM-6619] [BEAM-6593] Add gcsio integration tests to postcommit new dd6b3d0 Merge pull request #7946 from Juta/it-tests 5d630dc is described below commit 5d630dc6a2bc07eedfec45e9b968b041b4be9075 Author: Juta <juta_st...@hotmail.com> AuthorDate: Tue Feb 26 10:12:25 2019 +0100 [BEAM-6619] [BEAM-6593] Add gcsio integration tests to postcommit --- .../python/apache_beam/examples/streaming_wordcount_it_test.py | 2 +- sdks/python/apache_beam/io/gcp/tests/pubsub_matcher.py | 2 ++ sdks/python/apache_beam/io/gcp/tests/pubsub_matcher_test.py | 4 ++-- sdks/python/test-suites/dataflow/py3/build.gradle | 8 ++++++++ sdks/python/test-suites/direct/py3/build.gradle | 10 ++++++++++ 5 files changed, 23 insertions(+), 3 deletions(-) diff --git a/sdks/python/apache_beam/examples/streaming_wordcount_it_test.py b/sdks/python/apache_beam/examples/streaming_wordcount_it_test.py index 78e89a1..281dc69 100644 --- a/sdks/python/apache_beam/examples/streaming_wordcount_it_test.py +++ b/sdks/python/apache_beam/examples/streaming_wordcount_it_test.py @@ -70,7 +70,7 @@ class StreamingWordCountIT(unittest.TestCase): """Inject numbers as test data to PubSub.""" logging.debug('Injecting %d numbers to topic %s', num_messages, topic.name) for n in range(num_messages): - self.pub_client.publish(self.input_topic.name, str(n)) + self.pub_client.publish(self.input_topic.name, str(n).encode('utf-8')) def tearDown(self): test_utils.cleanup_subscriptions(self.sub_client, diff --git a/sdks/python/apache_beam/io/gcp/tests/pubsub_matcher.py b/sdks/python/apache_beam/io/gcp/tests/pubsub_matcher.py index 7a0b5c8..8b8e515 100644 --- a/sdks/python/apache_beam/io/gcp/tests/pubsub_matcher.py +++ b/sdks/python/apache_beam/io/gcp/tests/pubsub_matcher.py @@ -103,6 +103,8 @@ class PubSubMessageMatcher(BaseMatcher): for rm in response.received_messages: msg = PubsubMessage._from_message(rm.message) if not self.with_attributes: + if isinstance(msg.data, bytes): + msg.data = msg.data.decode('utf-8') total_messages.append(msg.data) continue diff --git a/sdks/python/apache_beam/io/gcp/tests/pubsub_matcher_test.py b/sdks/python/apache_beam/io/gcp/tests/pubsub_matcher_test.py index bfed329..6a58ddf 100644 --- a/sdks/python/apache_beam/io/gcp/tests/pubsub_matcher_test.py +++ b/sdks/python/apache_beam/io/gcp/tests/pubsub_matcher_test.py @@ -59,7 +59,7 @@ class PubSubMatcherTest(unittest.TestCase): def test_message_matcher_success(self, mock_get_sub, unsued_mock): self.init_matcher() - self.pubsub_matcher.expected_msg = [b'a', b'b'] + self.pubsub_matcher.expected_msg = ['a', 'b'] mock_sub = mock_get_sub.return_value mock_sub.pull.side_effect = [ create_pull_response([PullResponseMessage(b'a', {})]), @@ -130,7 +130,7 @@ class PubSubMatcherTest(unittest.TestCase): with self.assertRaises(AssertionError) as error: hc_assert_that(self.mock_presult, self.pubsub_matcher) self.assertEqual(mock_sub.pull.call_count, 1) - self.assertCountEqual([b'c', b'd'], self.pubsub_matcher.messages) + self.assertCountEqual(['c', 'd'], self.pubsub_matcher.messages) self.assertTrue( '\nExpected: Expected 1 messages.\n but: Got 2 messages.' in str(error.exception.args[0])) diff --git a/sdks/python/test-suites/dataflow/py3/build.gradle b/sdks/python/test-suites/dataflow/py3/build.gradle index f6465b7..b7a2ec0 100644 --- a/sdks/python/test-suites/dataflow/py3/build.gradle +++ b/sdks/python/test-suites/dataflow/py3/build.gradle @@ -36,6 +36,14 @@ task postCommitIT(dependsOn: ['sdist', 'installGcpTest']) { def tests = [ "apache_beam.examples.wordcount_it_test:WordCountIT.test_wordcount_it", "apache_beam.examples.cookbook.bigquery_tornadoes_it_test:BigqueryTornadoesIT.test_bigquery_tornadoes_it", + "apache_beam.examples.streaming_wordcount_it_test:StreamingWordCountIT.test_streaming_wordcount_it", + "apache_beam.examples.wordcount_it_test:WordCountIT.test_wordcount_fnapi_it", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_batch", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_batch_kms", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_batch_rewrite_token", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_kms", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_rewrite_token", ] def testOpts = basicTestOpts + ["--tests=${tests.join(',')}"] def cmdArgs = project.mapToArgString([ diff --git a/sdks/python/test-suites/direct/py3/build.gradle b/sdks/python/test-suites/direct/py3/build.gradle index f8b58ea..4e68446 100644 --- a/sdks/python/test-suites/direct/py3/build.gradle +++ b/sdks/python/test-suites/direct/py3/build.gradle @@ -30,10 +30,20 @@ task postCommitIT(dependsOn: 'installGcpTest') { def batchTests = [ "apache_beam.examples.wordcount_it_test:WordCountIT.test_wordcount_it", "apache_beam.examples.cookbook.bigquery_tornadoes_it_test:BigqueryTornadoesIT.test_bigquery_tornadoes_it", + "apache_beam.examples.streaming_wordcount_it_test:StreamingWordCountIT.test_streaming_wordcount_it", + "apache_beam.examples.wordcount_it_test:WordCountIT.test_wordcount_fnapi_it", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_batch", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_batch_kms", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_batch_rewrite_token", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_kms", + "apache_beam.io.gcp.gcsio_integration_test:GcsIOIntegrationTest.test_copy_rewrite_token", ] def testOpts = [ "--tests=${batchTests.join(',')}", "--nocapture", // Print stdout instantly + "--processes=4", // run tests in parallel + "--process-timeout=4500", // timeout of whole command execution ] def argMap = ["runner": "TestDirectRunner", "test_opts": testOpts]