Merge pull request apache#7946 from Juta/it-tests

[BEAM-6619] [BEAM-6593] Add gcsio integration tests to postcommit
grtheod · Mar 7, 2019 · dd6b3d0 · dd6b3d0
2 parents e835a05 + 5d630dc
commit dd6b3d0
Show file tree

Hide file tree

Showing 5 changed files with 23 additions and 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
@@ -70,7 +70,7 @@ def _inject_numbers(self, topic, num_messages):
     """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
@@ -103,6 +103,8 @@ def _wait_for_messages(self, expected_num, timeout):
       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
@@ -59,7 +59,7 @@ def init_matcher(self, with_attributes=False, strip_attributes=None):
 
   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 @@ def test_message_matcher_mismatch(self, mock_get_sub, unused_mock):
     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
@@ -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
@@ -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]