@@ -0,0 +1,3137 @@
+# Copyright 2018 Google LLC%0A#%0A# Licensed under the Apache License, Version 2.0 (the %22License%22);%0A# you may not use this file except in compliance with the License.%0A# You may obtain a copy of the License at%0A#%0A#     https://www.apache.org/licenses/LICENSE-2.0%0A#%0A# Unless required by applicable law or agreed to in writing, software%0A# distributed under the License is distributed on an %22AS IS%22 BASIS,%0A# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A# See the License for the specific language governing permissions and%0A# limitations under the License.%0A%22%22%22Integration tests for proto_task_queue's worker and requestor.%22%22%22%0A%0Afrom __future__ import absolute_import%0Afrom __future__ import division%0Afrom __future__ import print_function%0A%0Aimport os%0Aimport time%0Aimport unittest%0Afrom unittest import mock%0Aimport uuid%0A%0Afrom absl.testing import absltest%0Afrom proto_task_queue import requestor%0Afrom proto_task_queue import test_task_pb2%0Afrom proto_task_queue import worker%0A%0Afrom google.cloud.pubsub_v1.publisher import client as publisher_client%0Afrom google.cloud.pubsub_v1.subscriber import client as subscriber_client%0A%0A_PROJECT_ID_ENV_VAR = 'PTQ_PROJECT'%0A_PROJECT_ID = os.environ.get(_PROJECT_ID_ENV_VAR)%0A_TOPIC_ID = 'ptq-test-%7B%7D'.format(uuid.uuid4())%0A_TOPIC_NAME = 'projects/%7B%7D/topics/%7B%7D'.format(_PROJECT_ID, _TOPIC_ID)%0A_SUBSCRIPTION_ID = _TOPIC_ID%0A_SUBSCRIPTION_NAME = 'projects/%7B%7D/subscriptions/%7B%7D'.format(%0A    _PROJECT_ID, _SUBSCRIPTION_ID)%0A%0A%0A@unittest.skipUnless(%0A    _PROJECT_ID,%0A    ('To run integration tests, set %7B%7D to a GCP project ID that you can create '%0A     'topics and subscriptions in.').format(_PROJECT_ID_ENV_VAR))%0Aclass ProtoTaskQueueTest(absltest.TestCase):%0A%0A  def setUp(self):%0A    self._publisher_client = publisher_client.Client()%0A    self._publisher_client.create_topic(_TOPIC_NAME)%0A    self._subscriber_client = subscriber_client.Client()%0A    self._subscriber_client.create_subscription(_SUBSCRIPTION_NAME, _TOPIC_NAME)%0A%0A    self._requestor = requestor.Requestor()%0A    self._worker = worker.Worker()%0A%0A  def tearDown(self):%0A    self._subscriber_client.delete_subscription(_SUBSCRIPTION_NAME)%0A    self._publisher_client.delete_topic(_TOPIC_NAME)%0A%0A  def test_retry_failure_then_succeed(self):%0A    # Fail once, then succeed.%0A    foo_task_processor = mock.Mock(%0A        side_effect=%5BRuntimeError('retry this'), None%5D)%0A    self._worker.register(test_task_pb2.FooTaskArgs, foo_task_processor)%0A    subscriber_future = self._worker.subscribe(_SUBSCRIPTION_NAME)%0A%0A    # Request that foo be worked on.%0A    foo_task_args = test_task_pb2.FooTaskArgs()%0A    foo_task_args.widget = 'widget maker maker'%0A    request_future = self._requestor.request(_TOPIC_NAME, foo_task_args)%0A    request_future.result(timeout=10)%0A%0A    # Wait up to a minute for the messages to go through, then stop waiting for%0A    # more.%0A    for _ in range(60):%0A      time.sleep(1)%0A      if foo_task_processor.call_count %3E= 2:%0A        break%0A    subscriber_future.cancel()%0A%0A    self.assertEqual(2, foo_task_processor.call_count)%0A    foo_task_processor.assert_has_calls(%5Bmock.call(foo_task_args)%5D * 2)%0A%0A%0Aif __name__ == '__main__':%0A  absltest.main()%0A

Add integration tests.