From cca033b72242c6d04e5ad301508fd2aea9fe2469 Mon Sep 17 00:00:00 2001 From: Danny Hermes Date: Thu, 22 Sep 2016 10:14:18 -0700 Subject: [PATCH 1/3] Revert "Making alias for grpc _Rendezvous." This reverts commit b185ee8420a6170322beb50eae14f6a082b82b4f. --- google/cloud/bigtable/row_data.py | 2 +- google/cloud/bigtable/table.py | 2 +- google/cloud/datastore/connection.py | 5 +-- google/cloud/exceptions.py | 11 ------ google/cloud/logging/_gax.py | 20 +++++------ google/cloud/pubsub/_gax.py | 26 +++++++------- system_tests/bigtable.py | 4 +-- system_tests/logging_.py | 4 +-- system_tests/pubsub.py | 4 +-- unit_tests/_testing.py | 4 +-- unit_tests/datastore/test_connection.py | 10 +++--- unit_tests/logging/test__gax.py | 36 +++++++++---------- unit_tests/pubsub/test__gax.py | 48 ++++++++++++------------- 13 files changed, 83 insertions(+), 93 deletions(-) diff --git a/google/cloud/bigtable/row_data.py b/google/cloud/bigtable/row_data.py index 91d457319973..add93113f6ad 100644 --- a/google/cloud/bigtable/row_data.py +++ b/google/cloud/bigtable/row_data.py @@ -183,7 +183,7 @@ class InvalidChunk(RuntimeError): class PartialRowsData(object): """Convenience wrapper for consuming a ``ReadRows`` streaming response. - :type response_iterator: :class:`~google.cloud.exceptions.GrpcRendezvous` + :type response_iterator: :class:`grpc._channel._Rendezvous` :param response_iterator: A streaming iterator returned from a ``ReadRows`` request. """ diff --git a/google/cloud/bigtable/table.py b/google/cloud/bigtable/table.py index 6e3ac7eef370..665491d7d0a7 100644 --- a/google/cloud/bigtable/table.py +++ b/google/cloud/bigtable/table.py @@ -302,7 +302,7 @@ def sample_row_keys(self): samples would require space roughly equal to the difference in their ``offset_bytes`` fields. - :rtype: :class:`~google.cloud.exceptions.GrpcRendezvous` + :rtype: :class:`grpc._channel._Rendezvous` :returns: A cancel-able iterator. Can be consumed by calling ``next()`` or by casting to a :class:`list` and can be cancelled by calling ``cancel()``. diff --git a/google/cloud/datastore/connection.py b/google/cloud/datastore/connection.py index 9382eb533176..1f6416a1d39b 100644 --- a/google/cloud/datastore/connection.py +++ b/google/cloud/datastore/connection.py @@ -24,17 +24,18 @@ from google.cloud.environment_vars import DISABLE_GRPC from google.cloud.environment_vars import GCD_HOST from google.cloud.exceptions import Conflict -from google.cloud.exceptions import GrpcRendezvous from google.cloud.exceptions import make_exception from google.cloud.datastore._generated import datastore_pb2 as _datastore_pb2 # pylint: disable=ungrouped-imports try: from grpc import StatusCode + from grpc._channel import _Rendezvous from google.cloud.datastore._generated import datastore_grpc_pb2 except ImportError: # pragma: NO COVER _HAVE_GRPC = False datastore_grpc_pb2 = None StatusCode = None + _Rendezvous = Exception else: _HAVE_GRPC = True # pylint: enable=ungrouped-imports @@ -312,7 +313,7 @@ def commit(self, project, request_pb): request_pb.project_id = project try: return self._stub.Commit(request_pb) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.ABORTED: raise Conflict(exc.details()) raise diff --git a/google/cloud/exceptions.py b/google/cloud/exceptions.py index b21abad5cf84..ae5e44dbe1fb 100644 --- a/google/cloud/exceptions.py +++ b/google/cloud/exceptions.py @@ -25,17 +25,6 @@ _HTTP_CODE_TO_EXCEPTION = {} # populated at end of module -try: - from grpc._channel import _Rendezvous -except ImportError: # pragma: NO COVER - _Rendezvous = None - - -# pylint: disable=invalid-name -GrpcRendezvous = _Rendezvous -"""Exception class raised by gRPC stable.""" -# pylint: enable=invalid-name - class GoogleCloudError(Exception): """Base error class for Google Cloud errors (abstract). diff --git a/google/cloud/logging/_gax.py b/google/cloud/logging/_gax.py index f84ef53b6f42..ec4f70d42cc8 100644 --- a/google/cloud/logging/_gax.py +++ b/google/cloud/logging/_gax.py @@ -24,12 +24,12 @@ from google.logging.v2.log_entry_pb2 import LogEntry from google.protobuf.json_format import Parse from grpc import StatusCode +from grpc._channel import _Rendezvous # pylint: disable=ungrouped-imports from google.cloud._helpers import _datetime_to_pb_timestamp from google.cloud._helpers import _pb_timestamp_to_rfc3339 from google.cloud.exceptions import Conflict -from google.cloud.exceptions import GrpcRendezvous from google.cloud.exceptions import NotFound # pylint: enable=ungrouped-imports @@ -122,7 +122,7 @@ def logger_delete(self, project, logger_name): path = 'projects/%s/logs/%s' % (project, logger_name) try: self._gax_api.delete_log(path, options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(path) raise @@ -194,7 +194,7 @@ def sink_create(self, project, sink_name, filter_, destination): destination=destination) try: self._gax_api.create_sink(parent, sink_pb, options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.FAILED_PRECONDITION: path = 'projects/%s/sinks/%s' % (project, sink_name) raise Conflict(path) @@ -217,7 +217,7 @@ def sink_get(self, project, sink_name): path = 'projects/%s/sinks/%s' % (project, sink_name) try: sink_pb = self._gax_api.get_sink(path, options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(path) raise @@ -249,7 +249,7 @@ def sink_update(self, project, sink_name, filter_, destination): sink_pb = LogSink(name=path, filter=filter_, destination=destination) try: self._gax_api.update_sink(path, sink_pb, options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(path) raise @@ -268,7 +268,7 @@ def sink_delete(self, project, sink_name): path = 'projects/%s/sinks/%s' % (project, sink_name) try: self._gax_api.delete_sink(path, options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(path) raise @@ -339,7 +339,7 @@ def metric_create(self, project, metric_name, filter_, description): description=description) try: self._gax_api.create_log_metric(parent, metric_pb, options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.FAILED_PRECONDITION: path = 'projects/%s/metrics/%s' % (project, metric_name) raise Conflict(path) @@ -362,7 +362,7 @@ def metric_get(self, project, metric_name): path = 'projects/%s/metrics/%s' % (project, metric_name) try: metric_pb = self._gax_api.get_log_metric(path, options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(path) raise @@ -394,7 +394,7 @@ def metric_update(self, project, metric_name, filter_, description): description=description) try: self._gax_api.update_log_metric(path, metric_pb, options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(path) raise @@ -413,7 +413,7 @@ def metric_delete(self, project, metric_name): path = 'projects/%s/metrics/%s' % (project, metric_name) try: self._gax_api.delete_log_metric(path, options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(path) raise diff --git a/google/cloud/pubsub/_gax.py b/google/cloud/pubsub/_gax.py index c05d1f352518..4e397bef2758 100644 --- a/google/cloud/pubsub/_gax.py +++ b/google/cloud/pubsub/_gax.py @@ -22,12 +22,12 @@ from google.pubsub.v1.pubsub_pb2 import PushConfig from grpc import insecure_channel from grpc import StatusCode +from grpc._channel import _Rendezvous # pylint: disable=ungrouped-imports from google.cloud._helpers import _to_bytes from google.cloud._helpers import _pb_timestamp_to_rfc3339 from google.cloud.exceptions import Conflict -from google.cloud.exceptions import GrpcRendezvous from google.cloud.exceptions import NotFound # pylint: enable=ungrouped-imports @@ -92,7 +92,7 @@ def topic_create(self, topic_path): """ try: topic_pb = self._gax_api.create_topic(topic_path) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.FAILED_PRECONDITION: raise Conflict(topic_path) raise @@ -115,7 +115,7 @@ def topic_get(self, topic_path): """ try: topic_pb = self._gax_api.get_topic(topic_path) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(topic_path) raise @@ -133,7 +133,7 @@ def topic_delete(self, topic_path): """ try: self._gax_api.delete_topic(topic_path) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(topic_path) raise @@ -162,7 +162,7 @@ def topic_publish(self, topic_path, messages): try: result = self._gax_api.publish(topic_path, message_pbs, options=options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(topic_path) raise @@ -200,7 +200,7 @@ def topic_list_subscriptions(self, topic_path, page_size=0, try: page_iter = self._gax_api.list_topic_subscriptions( topic_path, page_size=page_size, options=options) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(topic_path) raise @@ -293,7 +293,7 @@ def subscription_create(self, subscription_path, topic_path, try: sub_pb = self._gax_api.create_subscription( subscription_path, topic_path, push_config, ack_deadline) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.FAILED_PRECONDITION: raise Conflict(topic_path) raise @@ -315,7 +315,7 @@ def subscription_get(self, subscription_path): """ try: sub_pb = self._gax_api.get_subscription(subscription_path) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise @@ -334,7 +334,7 @@ def subscription_delete(self, subscription_path): """ try: self._gax_api.delete_subscription(subscription_path) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise @@ -359,7 +359,7 @@ def subscription_modify_push_config(self, subscription_path, push_config = PushConfig(push_endpoint=push_endpoint) try: self._gax_api.modify_push_config(subscription_path, push_config) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise @@ -391,7 +391,7 @@ def subscription_pull(self, subscription_path, return_immediately=False, try: response_pb = self._gax_api.pull( subscription_path, max_messages, return_immediately) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise @@ -414,7 +414,7 @@ def subscription_acknowledge(self, subscription_path, ack_ids): """ try: self._gax_api.acknowledge(subscription_path, ack_ids) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise @@ -441,7 +441,7 @@ def subscription_modify_ack_deadline(self, subscription_path, ack_ids, try: self._gax_api.modify_ack_deadline( subscription_path, ack_ids, ack_deadline) - except GrpcRendezvous as exc: + except _Rendezvous as exc: if exc.code() == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise diff --git a/system_tests/bigtable.py b/system_tests/bigtable.py index bcb4c41965c2..af8668031a87 100644 --- a/system_tests/bigtable.py +++ b/system_tests/bigtable.py @@ -93,7 +93,7 @@ def _retry_on_unavailable(exc): def setUpModule(): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous Config.IN_EMULATOR = os.getenv(BIGTABLE_EMULATOR) is not None @@ -106,7 +106,7 @@ def setUpModule(): Config.INSTANCE = Config.CLIENT.instance(INSTANCE_ID, LOCATION_ID) if not Config.IN_EMULATOR: - retry = RetryErrors(GrpcRendezvous, + retry = RetryErrors(_Rendezvous, error_predicate=_retry_on_unavailable) instances, failed_locations = retry(Config.CLIENT.list_instances)() diff --git a/system_tests/logging_.py b/system_tests/logging_.py index 5292c0afce64..3ba39edc8912 100644 --- a/system_tests/logging_.py +++ b/system_tests/logging_.py @@ -73,9 +73,9 @@ def _logger_name(): return 'system-tests-logger' + unique_resource_id('-') def _list_entries(self, logger): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous inner = RetryResult(_has_entries)(logger.list_entries) - outer = RetryErrors(GrpcRendezvous, _retry_on_unavailable)(inner) + outer = RetryErrors(_Rendezvous, _retry_on_unavailable)(inner) return outer() def test_log_text(self): diff --git a/system_tests/pubsub.py b/system_tests/pubsub.py index 7da8600df946..cd5ffaacbe3f 100644 --- a/system_tests/pubsub.py +++ b/system_tests/pubsub.py @@ -16,11 +16,11 @@ import unittest from grpc import StatusCode +from grpc._channel import _Rendezvous import httplib2 # pylint: disable=ungrouped-imports from google.cloud.environment_vars import PUBSUB_EMULATOR -from google.cloud.exceptions import GrpcRendezvous from google.cloud.pubsub import client # pylint: enable=ungrouped-imports @@ -35,7 +35,7 @@ def _unavailable(exc): return exc.code() == StatusCode.UNAVAILABLE -retry_unavailable = RetryErrors(GrpcRendezvous, _unavailable) +retry_unavailable = RetryErrors(_Rendezvous, _unavailable) class Config(object): diff --git a/unit_tests/_testing.py b/unit_tests/_testing.py index f181b7c94a6b..5436145073fd 100644 --- a/unit_tests/_testing.py +++ b/unit_tests/_testing.py @@ -58,16 +58,16 @@ def __init__(self, **kw): self.__dict__.update(kw) def _make_grpc_error(self, status_code=None): + from grpc._channel import _Rendezvous from grpc._channel import _RPCState from grpc import StatusCode - from google.cloud.exceptions import GrpcRendezvous if status_code is None: status_code = StatusCode.UNKNOWN details = 'Some error details.' exc_state = _RPCState((), None, None, status_code, details) - return GrpcRendezvous(exc_state, None, None, None) + return _Rendezvous(exc_state, None, None, None) def _make_grpc_not_found(self): from grpc import StatusCode diff --git a/unit_tests/datastore/test_connection.py b/unit_tests/datastore/test_connection.py index 5df1e08c762f..44b829367b9f 100644 --- a/unit_tests/datastore/test_connection.py +++ b/unit_tests/datastore/test_connection.py @@ -243,24 +243,24 @@ def _commit_failure_helper(self, exc, err_class): @unittest.skipUnless(_HAVE_GRPC, 'No gRPC') def test_commit_failure_aborted(self): from grpc import StatusCode + from grpc._channel import _Rendezvous from grpc._channel import _RPCState from google.cloud.exceptions import Conflict - from google.cloud.exceptions import GrpcRendezvous details = 'Bad things.' exc_state = _RPCState((), None, None, StatusCode.ABORTED, details) - exc = GrpcRendezvous(exc_state, None, None, None) + exc = _Rendezvous(exc_state, None, None, None) self._commit_failure_helper(exc, Conflict) @unittest.skipUnless(_HAVE_GRPC, 'No gRPC') def test_commit_failure_cancelled(self): from grpc import StatusCode + from grpc._channel import _Rendezvous from grpc._channel import _RPCState - from google.cloud.exceptions import GrpcRendezvous exc_state = _RPCState((), None, None, StatusCode.CANCELLED, None) - exc = GrpcRendezvous(exc_state, None, None, None) - self._commit_failure_helper(exc, GrpcRendezvous) + exc = _Rendezvous(exc_state, None, None, None) + self._commit_failure_helper(exc, _Rendezvous) @unittest.skipUnless(_HAVE_GRPC, 'No gRPC') def test_commit_failure_non_grpc_err(self): diff --git a/unit_tests/logging/test__gax.py b/unit_tests/logging/test__gax.py index c27068c2d770..bed668cfa0c9 100644 --- a/unit_tests/logging/test__gax.py +++ b/unit_tests/logging/test__gax.py @@ -423,13 +423,13 @@ def test_logger_delete_not_found(self): self.assertIsNone(options) def test_logger_delete_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous LOG_PATH = 'projects/%s/logs/%s' % (self.PROJECT, self.LOG_NAME) gax_api = _GAXLoggingAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.logger_delete(self.PROJECT, self.LOG_NAME) log_name, options = gax_api._delete_log_called_with @@ -504,12 +504,12 @@ def test_list_sinks_w_paging(self): self.assertEqual(options.page_token, TOKEN) def test_sink_create_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXSinksAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.sink_create( self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI) @@ -550,12 +550,12 @@ def test_sink_get_error(self): api.sink_get(self.PROJECT, self.SINK_NAME) def test_sink_get_miss(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXSinksAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.sink_get(self.PROJECT, self.SINK_NAME) def test_sink_get_hit(self): @@ -578,12 +578,12 @@ def test_sink_get_hit(self): self.assertIsNone(options) def test_sink_update_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXSinksAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.sink_update( self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI) @@ -618,12 +618,12 @@ def test_sink_update_hit(self): self.assertIsNone(options) def test_sink_delete_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXSinksAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.sink_delete(self.PROJECT, self.SINK_NAME) def test_sink_delete_miss(self): @@ -712,12 +712,12 @@ def test_list_metrics_w_paging(self): self.assertEqual(options.page_token, TOKEN) def test_metric_create_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXMetricsAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.metric_create( self.PROJECT, self.METRIC_NAME, self.FILTER, self.DESCRIPTION) @@ -758,12 +758,12 @@ def test_metric_get_error(self): api.metric_get(self.PROJECT, self.METRIC_NAME) def test_metric_get_miss(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXMetricsAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.metric_get(self.PROJECT, self.METRIC_NAME) def test_metric_get_hit(self): @@ -786,12 +786,12 @@ def test_metric_get_hit(self): self.assertIsNone(options) def test_metric_update_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXMetricsAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.metric_update( self.PROJECT, self.METRIC_NAME, self.FILTER, self.DESCRIPTION) @@ -826,12 +826,12 @@ def test_metric_update_hit(self): self.assertIsNone(options) def test_metric_delete_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXMetricsAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.metric_delete(self.PROJECT, self.METRIC_NAME) def test_metric_delete_miss(self): diff --git a/unit_tests/pubsub/test__gax.py b/unit_tests/pubsub/test__gax.py index ad7752c17aa6..2ea8beb84c4a 100644 --- a/unit_tests/pubsub/test__gax.py +++ b/unit_tests/pubsub/test__gax.py @@ -123,12 +123,12 @@ def test_topic_create_already_exists(self): self.assertIsNone(options) def test_topic_create_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.topic_create(self.TOPIC_PATH) topic_path, options = gax_api._create_topic_called_with @@ -160,12 +160,12 @@ def test_topic_get_miss(self): self.assertIsNone(options) def test_topic_get_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.topic_get(self.TOPIC_PATH) topic_path, options = gax_api._get_topic_called_with @@ -195,12 +195,12 @@ def test_topic_delete_miss(self): self.assertIsNone(options) def test_topic_delete_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.topic_delete(self.TOPIC_PATH) topic_path, options = gax_api._delete_topic_called_with @@ -248,7 +248,7 @@ def test_topic_publish_miss_w_attrs_w_bytes_payload(self): def test_topic_publish_error(self): import base64 - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous PAYLOAD = b'This is the message text' B64 = base64.b64encode(PAYLOAD).decode('ascii') @@ -256,7 +256,7 @@ def test_topic_publish_error(self): gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.topic_publish(self.TOPIC_PATH, [MESSAGE]) topic_path, message_pbs, options = gax_api._publish_called_with @@ -333,12 +333,12 @@ def test_topic_list_subscriptions_miss(self): def test_topic_list_subscriptions_error(self): from google.gax import INITIAL_PAGE - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.topic_list_subscriptions(self.TOPIC_PATH) topic_path, page_size, options = ( @@ -455,12 +455,12 @@ def test_subscription_create_already_exists(self): self.assertIsNone(options) def test_subscription_create_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.subscription_create(self.SUB_PATH, self.TOPIC_PATH) name, topic, push_config, ack_deadline, options = ( @@ -505,12 +505,12 @@ def test_subscription_get_miss(self): self.assertIsNone(options) def test_subscription_get_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.subscription_get(self.SUB_PATH) sub_path, options = gax_api._get_subscription_called_with @@ -540,12 +540,12 @@ def test_subscription_delete_miss(self): self.assertIsNone(options) def test_subscription_delete_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.subscription_delete(self.TOPIC_PATH) sub_path, options = gax_api._delete_subscription_called_with @@ -578,12 +578,12 @@ def test_subscription_modify_push_config_miss(self): self.assertIsNone(options) def test_subscription_modify_push_config_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.subscription_modify_push_config( self.SUB_PATH, self.PUSH_ENDPOINT) @@ -645,12 +645,12 @@ def test_subscription_pull_defaults_miss(self): self.assertIsNone(options) def test_subscription_pull_defaults_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.subscription_pull(self.SUB_PATH) sub_path, max_messages, return_immediately, options = ( @@ -689,14 +689,14 @@ def test_subscription_acknowledge_miss(self): self.assertIsNone(options) def test_subscription_acknowledge_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous ACK_ID1 = 'DEADBEEF' ACK_ID2 = 'BEADCAFE' gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.subscription_acknowledge(self.SUB_PATH, [ACK_ID1, ACK_ID2]) sub_path, ack_ids, options = gax_api._acknowledge_called_with @@ -741,7 +741,7 @@ def test_subscription_modify_ack_deadline_miss(self): self.assertIsNone(options) def test_subscription_modify_ack_deadline_error(self): - from google.cloud.exceptions import GrpcRendezvous + from grpc._channel import _Rendezvous ACK_ID1 = 'DEADBEEF' ACK_ID2 = 'BEADCAFE' @@ -749,7 +749,7 @@ def test_subscription_modify_ack_deadline_error(self): gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(GrpcRendezvous): + with self.assertRaises(_Rendezvous): api.subscription_modify_ack_deadline( self.SUB_PATH, [ACK_ID1, ACK_ID2], NEW_DEADLINE) From f2ba165921a4c57961ab124417dba14e295b805a Mon Sep 17 00:00:00 2001 From: Danny Hermes Date: Thu, 22 Sep 2016 10:15:13 -0700 Subject: [PATCH 2/3] Revert "Fixing broken exception handling after GAX 0.13.0 upgrade." This reverts commit f916093e4fc6f052c8512fc9abe5ebe55ac7bf5c. --- google/cloud/_helpers.py | 19 +++++ google/cloud/logging/_gax.py | 39 +++++----- google/cloud/pubsub/_gax.py | 57 +++++++------- system_tests/pubsub.py | 6 +- unit_tests/_testing.py | 6 +- unit_tests/logging/test__gax.py | 90 +++++++++++----------- unit_tests/pubsub/test__gax.py | 132 ++++++++++++++++---------------- unit_tests/test__helpers.py | 31 ++++++++ 8 files changed, 215 insertions(+), 165 deletions(-) diff --git a/google/cloud/_helpers.py b/google/cloud/_helpers.py index 3b214db77b82..a7da90880bbc 100644 --- a/google/cloud/_helpers.py +++ b/google/cloud/_helpers.py @@ -31,9 +31,13 @@ except ImportError: app_identity = None try: + from google.gax.grpc import exc_to_code as beta_exc_to_code import grpc + from grpc._channel import _Rendezvous except ImportError: # pragma: NO COVER + beta_exc_to_code = None grpc = None + _Rendezvous = Exception import six from six.moves import http_client from six.moves import configparser @@ -681,6 +685,21 @@ def make_insecure_stub(stub_class, host, port=None): return stub_class(channel) +def exc_to_code(exc): + """Retrieves the status code from a gRPC exception. + + :type exc: :class:`Exception` + :param exc: An exception from gRPC beta or stable. + + :rtype: :class:`grpc.StatusCode` + :returns: The status code attached to the exception. + """ + if isinstance(exc, _Rendezvous): + return exc.code() + else: + return beta_exc_to_code(exc) + + try: from pytz import UTC # pylint: disable=unused-import,wrong-import-order except ImportError: diff --git a/google/cloud/logging/_gax.py b/google/cloud/logging/_gax.py index ec4f70d42cc8..8f6c76572cf1 100644 --- a/google/cloud/logging/_gax.py +++ b/google/cloud/logging/_gax.py @@ -18,17 +18,18 @@ from google.gax import CallOptions from google.gax import INITIAL_PAGE +from google.gax.errors import GaxError from google.logging.type.log_severity_pb2 import LogSeverity from google.logging.v2.logging_config_pb2 import LogSink from google.logging.v2.logging_metrics_pb2 import LogMetric from google.logging.v2.log_entry_pb2 import LogEntry from google.protobuf.json_format import Parse from grpc import StatusCode -from grpc._channel import _Rendezvous # pylint: disable=ungrouped-imports from google.cloud._helpers import _datetime_to_pb_timestamp from google.cloud._helpers import _pb_timestamp_to_rfc3339 +from google.cloud._helpers import exc_to_code from google.cloud.exceptions import Conflict from google.cloud.exceptions import NotFound # pylint: enable=ungrouped-imports @@ -122,8 +123,8 @@ def logger_delete(self, project, logger_name): path = 'projects/%s/logs/%s' % (project, logger_name) try: self._gax_api.delete_log(path, options) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(path) raise @@ -194,8 +195,8 @@ def sink_create(self, project, sink_name, filter_, destination): destination=destination) try: self._gax_api.create_sink(parent, sink_pb, options) - except _Rendezvous as exc: - if exc.code() == StatusCode.FAILED_PRECONDITION: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.FAILED_PRECONDITION: path = 'projects/%s/sinks/%s' % (project, sink_name) raise Conflict(path) raise @@ -217,8 +218,8 @@ def sink_get(self, project, sink_name): path = 'projects/%s/sinks/%s' % (project, sink_name) try: sink_pb = self._gax_api.get_sink(path, options) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(path) raise return _log_sink_pb_to_mapping(sink_pb) @@ -249,8 +250,8 @@ def sink_update(self, project, sink_name, filter_, destination): sink_pb = LogSink(name=path, filter=filter_, destination=destination) try: self._gax_api.update_sink(path, sink_pb, options) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(path) raise return _log_sink_pb_to_mapping(sink_pb) @@ -268,8 +269,8 @@ def sink_delete(self, project, sink_name): path = 'projects/%s/sinks/%s' % (project, sink_name) try: self._gax_api.delete_sink(path, options) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(path) raise @@ -339,8 +340,8 @@ def metric_create(self, project, metric_name, filter_, description): description=description) try: self._gax_api.create_log_metric(parent, metric_pb, options) - except _Rendezvous as exc: - if exc.code() == StatusCode.FAILED_PRECONDITION: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.FAILED_PRECONDITION: path = 'projects/%s/metrics/%s' % (project, metric_name) raise Conflict(path) raise @@ -362,8 +363,8 @@ def metric_get(self, project, metric_name): path = 'projects/%s/metrics/%s' % (project, metric_name) try: metric_pb = self._gax_api.get_log_metric(path, options) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(path) raise return _log_metric_pb_to_mapping(metric_pb) @@ -394,8 +395,8 @@ def metric_update(self, project, metric_name, filter_, description): description=description) try: self._gax_api.update_log_metric(path, metric_pb, options) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(path) raise return _log_metric_pb_to_mapping(metric_pb) @@ -413,8 +414,8 @@ def metric_delete(self, project, metric_name): path = 'projects/%s/metrics/%s' % (project, metric_name) try: self._gax_api.delete_log_metric(path, options) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(path) raise diff --git a/google/cloud/pubsub/_gax.py b/google/cloud/pubsub/_gax.py index 4e397bef2758..2a0b0393aa5b 100644 --- a/google/cloud/pubsub/_gax.py +++ b/google/cloud/pubsub/_gax.py @@ -18,14 +18,15 @@ from google.cloud.gapic.pubsub.v1.subscriber_api import SubscriberApi from google.gax import CallOptions from google.gax import INITIAL_PAGE +from google.gax.errors import GaxError from google.pubsub.v1.pubsub_pb2 import PubsubMessage from google.pubsub.v1.pubsub_pb2 import PushConfig -from grpc import insecure_channel +from grpc.beta.implementations import insecure_channel from grpc import StatusCode -from grpc._channel import _Rendezvous # pylint: disable=ungrouped-imports from google.cloud._helpers import _to_bytes +from google.cloud._helpers import exc_to_code from google.cloud._helpers import _pb_timestamp_to_rfc3339 from google.cloud.exceptions import Conflict from google.cloud.exceptions import NotFound @@ -92,8 +93,8 @@ def topic_create(self, topic_path): """ try: topic_pb = self._gax_api.create_topic(topic_path) - except _Rendezvous as exc: - if exc.code() == StatusCode.FAILED_PRECONDITION: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.FAILED_PRECONDITION: raise Conflict(topic_path) raise return {'name': topic_pb.name} @@ -115,8 +116,8 @@ def topic_get(self, topic_path): """ try: topic_pb = self._gax_api.get_topic(topic_path) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(topic_path) raise return {'name': topic_pb.name} @@ -133,8 +134,8 @@ def topic_delete(self, topic_path): """ try: self._gax_api.delete_topic(topic_path) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(topic_path) raise @@ -162,8 +163,8 @@ def topic_publish(self, topic_path, messages): try: result = self._gax_api.publish(topic_path, message_pbs, options=options) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(topic_path) raise return result.message_ids @@ -200,8 +201,8 @@ def topic_list_subscriptions(self, topic_path, page_size=0, try: page_iter = self._gax_api.list_topic_subscriptions( topic_path, page_size=page_size, options=options) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(topic_path) raise subs = page_iter.next() @@ -293,8 +294,8 @@ def subscription_create(self, subscription_path, topic_path, try: sub_pb = self._gax_api.create_subscription( subscription_path, topic_path, push_config, ack_deadline) - except _Rendezvous as exc: - if exc.code() == StatusCode.FAILED_PRECONDITION: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.FAILED_PRECONDITION: raise Conflict(topic_path) raise return _subscription_pb_to_mapping(sub_pb) @@ -315,8 +316,8 @@ def subscription_get(self, subscription_path): """ try: sub_pb = self._gax_api.get_subscription(subscription_path) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise return _subscription_pb_to_mapping(sub_pb) @@ -334,8 +335,8 @@ def subscription_delete(self, subscription_path): """ try: self._gax_api.delete_subscription(subscription_path) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise @@ -359,8 +360,8 @@ def subscription_modify_push_config(self, subscription_path, push_config = PushConfig(push_endpoint=push_endpoint) try: self._gax_api.modify_push_config(subscription_path, push_config) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise @@ -391,8 +392,8 @@ def subscription_pull(self, subscription_path, return_immediately=False, try: response_pb = self._gax_api.pull( subscription_path, max_messages, return_immediately) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise return [_received_message_pb_to_mapping(rmpb) @@ -414,8 +415,8 @@ def subscription_acknowledge(self, subscription_path, ack_ids): """ try: self._gax_api.acknowledge(subscription_path, ack_ids) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise @@ -441,8 +442,8 @@ def subscription_modify_ack_deadline(self, subscription_path, ack_ids, try: self._gax_api.modify_ack_deadline( subscription_path, ack_ids, ack_deadline) - except _Rendezvous as exc: - if exc.code() == StatusCode.NOT_FOUND: + except GaxError as exc: + if exc_to_code(exc.cause) == StatusCode.NOT_FOUND: raise NotFound(subscription_path) raise @@ -519,7 +520,7 @@ def make_gax_publisher_api(connection): """ channel = None if connection.in_emulator: - channel = insecure_channel(connection.host) + channel = insecure_channel(connection.host, None) return PublisherApi(channel=channel) @@ -539,5 +540,5 @@ def make_gax_subscriber_api(connection): """ channel = None if connection.in_emulator: - channel = insecure_channel(connection.host) + channel = insecure_channel(connection.host, None) return SubscriberApi(channel=channel) diff --git a/system_tests/pubsub.py b/system_tests/pubsub.py index cd5ffaacbe3f..4074d2724e61 100644 --- a/system_tests/pubsub.py +++ b/system_tests/pubsub.py @@ -15,11 +15,13 @@ import os import unittest +from google.gax.errors import GaxError from grpc import StatusCode from grpc._channel import _Rendezvous import httplib2 # pylint: disable=ungrouped-imports +from google.cloud import _helpers from google.cloud.environment_vars import PUBSUB_EMULATOR from google.cloud.pubsub import client # pylint: enable=ungrouped-imports @@ -32,10 +34,10 @@ def _unavailable(exc): - return exc.code() == StatusCode.UNAVAILABLE + return _helpers.exc_to_code(exc) == StatusCode.UNAVAILABLE -retry_unavailable = RetryErrors(_Rendezvous, _unavailable) +retry_unavailable = RetryErrors((GaxError, _Rendezvous), _unavailable) class Config(object): diff --git a/unit_tests/_testing.py b/unit_tests/_testing.py index 5436145073fd..d3a1a268b5b4 100644 --- a/unit_tests/_testing.py +++ b/unit_tests/_testing.py @@ -57,13 +57,9 @@ class _GAXBaseAPI(object): def __init__(self, **kw): self.__dict__.update(kw) - def _make_grpc_error(self, status_code=None): + def _make_grpc_error(self, status_code): from grpc._channel import _Rendezvous from grpc._channel import _RPCState - from grpc import StatusCode - - if status_code is None: - status_code = StatusCode.UNKNOWN details = 'Some error details.' exc_state = _RPCState((), None, None, status_code, details) diff --git a/unit_tests/logging/test__gax.py b/unit_tests/logging/test__gax.py index bed668cfa0c9..12baa74d5d4a 100644 --- a/unit_tests/logging/test__gax.py +++ b/unit_tests/logging/test__gax.py @@ -423,13 +423,12 @@ def test_logger_delete_not_found(self): self.assertIsNone(options) def test_logger_delete_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError LOG_PATH = 'projects/%s/logs/%s' % (self.PROJECT, self.LOG_NAME) gax_api = _GAXLoggingAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.logger_delete(self.PROJECT, self.LOG_NAME) log_name, options = gax_api._delete_log_called_with @@ -504,12 +503,11 @@ def test_list_sinks_w_paging(self): self.assertEqual(options.page_token, TOKEN) def test_sink_create_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXSinksAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.sink_create( self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI) @@ -550,12 +548,11 @@ def test_sink_get_error(self): api.sink_get(self.PROJECT, self.SINK_NAME) def test_sink_get_miss(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXSinksAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.sink_get(self.PROJECT, self.SINK_NAME) def test_sink_get_hit(self): @@ -578,12 +575,11 @@ def test_sink_get_hit(self): self.assertIsNone(options) def test_sink_update_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXSinksAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.sink_update( self.PROJECT, self.SINK_NAME, self.FILTER, self.DESTINATION_URI) @@ -618,12 +614,11 @@ def test_sink_update_hit(self): self.assertIsNone(options) def test_sink_delete_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXSinksAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.sink_delete(self.PROJECT, self.SINK_NAME) def test_sink_delete_miss(self): @@ -712,12 +707,11 @@ def test_list_metrics_w_paging(self): self.assertEqual(options.page_token, TOKEN) def test_metric_create_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXMetricsAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.metric_create( self.PROJECT, self.METRIC_NAME, self.FILTER, self.DESCRIPTION) @@ -758,12 +752,11 @@ def test_metric_get_error(self): api.metric_get(self.PROJECT, self.METRIC_NAME) def test_metric_get_miss(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXMetricsAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.metric_get(self.PROJECT, self.METRIC_NAME) def test_metric_get_hit(self): @@ -786,12 +779,11 @@ def test_metric_get_hit(self): self.assertIsNone(options) def test_metric_update_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXMetricsAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.metric_update( self.PROJECT, self.METRIC_NAME, self.FILTER, self.DESCRIPTION) @@ -826,12 +818,11 @@ def test_metric_update_hit(self): self.assertIsNone(options) def test_metric_delete_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXMetricsAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.metric_delete(self.PROJECT, self.METRIC_NAME) def test_metric_delete_miss(self): @@ -938,11 +929,12 @@ def write_log_entries(self, entries, log_name, resource, labels, entries, log_name, resource, labels, partial_success, options) def delete_log(self, log_name, options): + from google.gax.errors import GaxError self._delete_log_called_with = log_name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if self._delete_not_found: - raise self._make_grpc_not_found() + raise GaxError('notfound', self._make_grpc_not_found()) class _GAXSinksAPI(_GAXBaseAPI): @@ -955,36 +947,40 @@ def list_sinks(self, parent, page_size, options): return self._list_sinks_response def create_sink(self, parent, sink, options): + from google.gax.errors import GaxError self._create_sink_called_with = parent, sink, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if self._create_sink_conflict: - raise self._make_grpc_failed_precondition() + raise GaxError('conflict', self._make_grpc_failed_precondition()) def get_sink(self, sink_name, options): + from google.gax.errors import GaxError self._get_sink_called_with = sink_name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') try: return self._get_sink_response except AttributeError: - raise self._make_grpc_not_found() + raise GaxError('notfound', self._make_grpc_not_found()) def update_sink(self, sink_name, sink, options=None): + from google.gax.errors import GaxError self._update_sink_called_with = sink_name, sink, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') try: return self._update_sink_response except AttributeError: - raise self._make_grpc_not_found() + raise GaxError('notfound', self._make_grpc_not_found()) def delete_sink(self, sink_name, options=None): + from google.gax.errors import GaxError self._delete_sink_called_with = sink_name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if self._sink_not_found: - raise self._make_grpc_not_found() + raise GaxError('notfound', self._make_grpc_not_found()) class _GAXMetricsAPI(_GAXBaseAPI): @@ -997,36 +993,40 @@ def list_log_metrics(self, parent, page_size, options): return self._list_log_metrics_response def create_log_metric(self, parent, metric, options): + from google.gax.errors import GaxError self._create_log_metric_called_with = parent, metric, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if self._create_log_metric_conflict: - raise self._make_grpc_failed_precondition() + raise GaxError('conflict', self._make_grpc_failed_precondition()) def get_log_metric(self, metric_name, options): + from google.gax.errors import GaxError self._get_log_metric_called_with = metric_name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') try: return self._get_log_metric_response except AttributeError: - raise self._make_grpc_not_found() + raise GaxError('notfound', self._make_grpc_not_found()) def update_log_metric(self, metric_name, metric, options=None): + from google.gax.errors import GaxError self._update_log_metric_called_with = metric_name, metric, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') try: return self._update_log_metric_response except AttributeError: - raise self._make_grpc_not_found() + raise GaxError('notfound', self._make_grpc_not_found()) def delete_log_metric(self, metric_name, options=None): + from google.gax.errors import GaxError self._delete_log_metric_called_with = metric_name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if self._log_metric_not_found: - raise self._make_grpc_not_found() + raise GaxError('notfound', self._make_grpc_not_found()) class _HTTPRequestPB(object): diff --git a/unit_tests/pubsub/test__gax.py b/unit_tests/pubsub/test__gax.py index 2ea8beb84c4a..5a401494b924 100644 --- a/unit_tests/pubsub/test__gax.py +++ b/unit_tests/pubsub/test__gax.py @@ -123,12 +123,11 @@ def test_topic_create_already_exists(self): self.assertIsNone(options) def test_topic_create_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.topic_create(self.TOPIC_PATH) topic_path, options = gax_api._create_topic_called_with @@ -160,12 +159,11 @@ def test_topic_get_miss(self): self.assertIsNone(options) def test_topic_get_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.topic_get(self.TOPIC_PATH) topic_path, options = gax_api._get_topic_called_with @@ -195,12 +193,11 @@ def test_topic_delete_miss(self): self.assertIsNone(options) def test_topic_delete_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.topic_delete(self.TOPIC_PATH) topic_path, options = gax_api._delete_topic_called_with @@ -248,15 +245,14 @@ def test_topic_publish_miss_w_attrs_w_bytes_payload(self): def test_topic_publish_error(self): import base64 - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError PAYLOAD = b'This is the message text' B64 = base64.b64encode(PAYLOAD).decode('ascii') MESSAGE = {'data': B64, 'attributes': {}} gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.topic_publish(self.TOPIC_PATH, [MESSAGE]) topic_path, message_pbs, options = gax_api._publish_called_with @@ -333,12 +329,11 @@ def test_topic_list_subscriptions_miss(self): def test_topic_list_subscriptions_error(self): from google.gax import INITIAL_PAGE - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXPublisherAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.topic_list_subscriptions(self.TOPIC_PATH) topic_path, page_size, options = ( @@ -455,12 +450,11 @@ def test_subscription_create_already_exists(self): self.assertIsNone(options) def test_subscription_create_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.subscription_create(self.SUB_PATH, self.TOPIC_PATH) name, topic, push_config, ack_deadline, options = ( @@ -505,12 +499,11 @@ def test_subscription_get_miss(self): self.assertIsNone(options) def test_subscription_get_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.subscription_get(self.SUB_PATH) sub_path, options = gax_api._get_subscription_called_with @@ -540,12 +533,11 @@ def test_subscription_delete_miss(self): self.assertIsNone(options) def test_subscription_delete_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.subscription_delete(self.TOPIC_PATH) sub_path, options = gax_api._delete_subscription_called_with @@ -578,12 +570,11 @@ def test_subscription_modify_push_config_miss(self): self.assertIsNone(options) def test_subscription_modify_push_config_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.subscription_modify_push_config( self.SUB_PATH, self.PUSH_ENDPOINT) @@ -645,12 +636,11 @@ def test_subscription_pull_defaults_miss(self): self.assertIsNone(options) def test_subscription_pull_defaults_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.subscription_pull(self.SUB_PATH) sub_path, max_messages, return_immediately, options = ( @@ -689,14 +679,13 @@ def test_subscription_acknowledge_miss(self): self.assertIsNone(options) def test_subscription_acknowledge_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError ACK_ID1 = 'DEADBEEF' ACK_ID2 = 'BEADCAFE' gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.subscription_acknowledge(self.SUB_PATH, [ACK_ID1, ACK_ID2]) sub_path, ack_ids, options = gax_api._acknowledge_called_with @@ -741,15 +730,14 @@ def test_subscription_modify_ack_deadline_miss(self): self.assertIsNone(options) def test_subscription_modify_ack_deadline_error(self): - from grpc._channel import _Rendezvous - + from google.gax.errors import GaxError ACK_ID1 = 'DEADBEEF' ACK_ID2 = 'BEADCAFE' NEW_DEADLINE = 90 gax_api = _GAXSubscriberAPI(_random_gax_error=True) api = self._makeOne(gax_api) - with self.assertRaises(_Rendezvous): + with self.assertRaises(GaxError): api.subscription_modify_ack_deadline( self.SUB_PATH, [ACK_ID1, ACK_ID2], NEW_DEADLINE) @@ -799,8 +787,8 @@ def mock_publisher_api(channel): channels.append(channel) return mock_result - def mock_insecure_channel(host): - insecure_args.append(host) + def mock_insecure_channel(host, port): + insecure_args.append((host, port)) return mock_channel host = 'CURR_HOST:1234' @@ -811,7 +799,7 @@ def mock_insecure_channel(host): self.assertIs(result, mock_result) self.assertEqual(channels, [mock_channel]) - self.assertEqual(insecure_args, [host]) + self.assertEqual(insecure_args, [(host, None)]) @unittest.skipUnless(_HAVE_GAX, 'No gax-python') @@ -852,8 +840,8 @@ def mock_subscriber_api(channel): channels.append(channel) return mock_result - def mock_insecure_channel(host): - insecure_args.append(host) + def mock_insecure_channel(host, port): + insecure_args.append((host, port)) return mock_channel host = 'CURR_HOST:1234' @@ -864,7 +852,7 @@ def mock_insecure_channel(host): self.assertIs(result, mock_result) self.assertEqual(channels, [mock_channel]) - self.assertEqual(insecure_args, [host]) + self.assertEqual(insecure_args, [(host, None)]) class _GAXPublisherAPI(_GAXBaseAPI): @@ -876,46 +864,51 @@ def list_topics(self, name, page_size, options): return self._list_topics_response def create_topic(self, name, options=None): + from google.gax.errors import GaxError self._create_topic_called_with = name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if self._create_topic_conflict: - raise self._make_grpc_failed_precondition() + raise GaxError('conflict', self._make_grpc_failed_precondition()) return self._create_topic_response def get_topic(self, name, options=None): + from google.gax.errors import GaxError self._get_topic_called_with = name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') try: return self._get_topic_response except AttributeError: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) def delete_topic(self, name, options=None): + from google.gax.errors import GaxError self._delete_topic_called_with = name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if not self._delete_topic_ok: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) def publish(self, topic, messages, options=None): + from google.gax.errors import GaxError self._publish_called_with = topic, messages, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') try: return self._publish_response except AttributeError: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) def list_topic_subscriptions(self, topic, page_size, options=None): + from google.gax.errors import GaxError self._list_topic_subscriptions_called_with = topic, page_size, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') try: return self._list_topic_subscriptions_response except AttributeError: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) class _GAXSubscriberAPI(_GAXBaseAPI): @@ -932,61 +925,68 @@ def list_subscriptions(self, project, page_size, options=None): def create_subscription(self, name, topic, push_config, ack_deadline_seconds, options=None): + from google.gax.errors import GaxError self._create_subscription_called_with = ( name, topic, push_config, ack_deadline_seconds, options) if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if self._create_subscription_conflict: - raise self._make_grpc_failed_precondition() + raise GaxError('conflict', self._make_grpc_failed_precondition()) return self._create_subscription_response def get_subscription(self, name, options=None): + from google.gax.errors import GaxError self._get_subscription_called_with = name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') try: return self._get_subscription_response except AttributeError: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) def delete_subscription(self, name, options=None): + from google.gax.errors import GaxError self._delete_subscription_called_with = name, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if not self._delete_subscription_ok: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) def modify_push_config(self, name, push_config, options=None): + from google.gax.errors import GaxError self._modify_push_config_called_with = name, push_config, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if not self._modify_push_config_ok: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) def pull(self, name, max_messages, return_immediately, options=None): + from google.gax.errors import GaxError self._pull_called_with = ( name, max_messages, return_immediately, options) if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') try: return self._pull_response except AttributeError: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) def acknowledge(self, name, ack_ids, options=None): + from google.gax.errors import GaxError self._acknowledge_called_with = name, ack_ids, options if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if not self._acknowledge_ok: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) def modify_ack_deadline(self, name, ack_ids, deadline, options=None): + from google.gax.errors import GaxError self._modify_ack_deadline_called_with = ( name, ack_ids, deadline, options) if self._random_gax_error: - raise self._make_grpc_error() + raise GaxError('error') if not self._modify_ack_deadline_ok: - raise self._make_grpc_not_found() + raise GaxError('miss', self._make_grpc_not_found()) class _TopicPB(object): diff --git a/unit_tests/test__helpers.py b/unit_tests/test__helpers.py index 3a2950c42bbd..e315df6e6f2b 100644 --- a/unit_tests/test__helpers.py +++ b/unit_tests/test__helpers.py @@ -1025,6 +1025,37 @@ def test_without_port_argument(self): self._helper(host, host) +class Test_exc_to_code(unittest.TestCase): + + def _callFUT(self, exc): + from google.cloud._helpers import exc_to_code + return exc_to_code(exc) + + def test_with_stable(self): + from grpc._channel import _Rendezvous + from grpc._channel import _RPCState + from grpc import StatusCode + + status_code = StatusCode.FAILED_PRECONDITION + exc_state = _RPCState((), None, None, status_code, None) + exc = _Rendezvous(exc_state, None, None, None) + result = self._callFUT(exc) + self.assertEqual(result, status_code) + + def test_with_beta(self): + from grpc import StatusCode + from grpc.framework.interfaces.face.face import AbortionError + + status_code = StatusCode.UNIMPLEMENTED + exc = AbortionError(None, None, status_code, None) + result = self._callFUT(exc) + self.assertEqual(result, status_code) + + def test_with_none(self): + result = self._callFUT(None) + self.assertIsNone(result) + + class _AppIdentity(object): def __init__(self, app_id): From 523100f29fa2abf182c54f98fa633a65c9b2b8c4 Mon Sep 17 00:00:00 2001 From: Danny Hermes Date: Thu, 22 Sep 2016 11:13:21 -0700 Subject: [PATCH 3/3] Upgrading google-gax to 0.14.1. In the process, adding back code changes that were reverted in the process of reverting PR #2336. Some of the changes were necessary, but anything using GAX (logging and pubsub) could safely be restored. --- google/cloud/_helpers.py | 19 --------------- google/cloud/bigtable/row_data.py | 2 +- google/cloud/bigtable/table.py | 2 +- google/cloud/datastore/connection.py | 5 ++-- google/cloud/exceptions.py | 11 +++++++++ google/cloud/logging/_gax.py | 2 +- google/cloud/pubsub/_gax.py | 8 +++---- setup.py | 2 +- system_tests/bigtable.py | 4 ++-- system_tests/logging_.py | 5 ++-- system_tests/pubsub.py | 7 +++--- unit_tests/_testing.py | 4 ++-- unit_tests/datastore/test_connection.py | 10 ++++---- unit_tests/pubsub/test__gax.py | 12 +++++----- unit_tests/test__helpers.py | 31 ------------------------- 15 files changed, 42 insertions(+), 82 deletions(-) diff --git a/google/cloud/_helpers.py b/google/cloud/_helpers.py index a7da90880bbc..3b214db77b82 100644 --- a/google/cloud/_helpers.py +++ b/google/cloud/_helpers.py @@ -31,13 +31,9 @@ except ImportError: app_identity = None try: - from google.gax.grpc import exc_to_code as beta_exc_to_code import grpc - from grpc._channel import _Rendezvous except ImportError: # pragma: NO COVER - beta_exc_to_code = None grpc = None - _Rendezvous = Exception import six from six.moves import http_client from six.moves import configparser @@ -685,21 +681,6 @@ def make_insecure_stub(stub_class, host, port=None): return stub_class(channel) -def exc_to_code(exc): - """Retrieves the status code from a gRPC exception. - - :type exc: :class:`Exception` - :param exc: An exception from gRPC beta or stable. - - :rtype: :class:`grpc.StatusCode` - :returns: The status code attached to the exception. - """ - if isinstance(exc, _Rendezvous): - return exc.code() - else: - return beta_exc_to_code(exc) - - try: from pytz import UTC # pylint: disable=unused-import,wrong-import-order except ImportError: diff --git a/google/cloud/bigtable/row_data.py b/google/cloud/bigtable/row_data.py index add93113f6ad..91d457319973 100644 --- a/google/cloud/bigtable/row_data.py +++ b/google/cloud/bigtable/row_data.py @@ -183,7 +183,7 @@ class InvalidChunk(RuntimeError): class PartialRowsData(object): """Convenience wrapper for consuming a ``ReadRows`` streaming response. - :type response_iterator: :class:`grpc._channel._Rendezvous` + :type response_iterator: :class:`~google.cloud.exceptions.GrpcRendezvous` :param response_iterator: A streaming iterator returned from a ``ReadRows`` request. """ diff --git a/google/cloud/bigtable/table.py b/google/cloud/bigtable/table.py index 665491d7d0a7..6e3ac7eef370 100644 --- a/google/cloud/bigtable/table.py +++ b/google/cloud/bigtable/table.py @@ -302,7 +302,7 @@ def sample_row_keys(self): samples would require space roughly equal to the difference in their ``offset_bytes`` fields. - :rtype: :class:`grpc._channel._Rendezvous` + :rtype: :class:`~google.cloud.exceptions.GrpcRendezvous` :returns: A cancel-able iterator. Can be consumed by calling ``next()`` or by casting to a :class:`list` and can be cancelled by calling ``cancel()``. diff --git a/google/cloud/datastore/connection.py b/google/cloud/datastore/connection.py index 1f6416a1d39b..9382eb533176 100644 --- a/google/cloud/datastore/connection.py +++ b/google/cloud/datastore/connection.py @@ -24,18 +24,17 @@ from google.cloud.environment_vars import DISABLE_GRPC from google.cloud.environment_vars import GCD_HOST from google.cloud.exceptions import Conflict +from google.cloud.exceptions import GrpcRendezvous from google.cloud.exceptions import make_exception from google.cloud.datastore._generated import datastore_pb2 as _datastore_pb2 # pylint: disable=ungrouped-imports try: from grpc import StatusCode - from grpc._channel import _Rendezvous from google.cloud.datastore._generated import datastore_grpc_pb2 except ImportError: # pragma: NO COVER _HAVE_GRPC = False datastore_grpc_pb2 = None StatusCode = None - _Rendezvous = Exception else: _HAVE_GRPC = True # pylint: enable=ungrouped-imports @@ -313,7 +312,7 @@ def commit(self, project, request_pb): request_pb.project_id = project try: return self._stub.Commit(request_pb) - except _Rendezvous as exc: + except GrpcRendezvous as exc: if exc.code() == StatusCode.ABORTED: raise Conflict(exc.details()) raise diff --git a/google/cloud/exceptions.py b/google/cloud/exceptions.py index ae5e44dbe1fb..b21abad5cf84 100644 --- a/google/cloud/exceptions.py +++ b/google/cloud/exceptions.py @@ -25,6 +25,17 @@ _HTTP_CODE_TO_EXCEPTION = {} # populated at end of module +try: + from grpc._channel import _Rendezvous +except ImportError: # pragma: NO COVER + _Rendezvous = None + + +# pylint: disable=invalid-name +GrpcRendezvous = _Rendezvous +"""Exception class raised by gRPC stable.""" +# pylint: enable=invalid-name + class GoogleCloudError(Exception): """Base error class for Google Cloud errors (abstract). diff --git a/google/cloud/logging/_gax.py b/google/cloud/logging/_gax.py index 8f6c76572cf1..90230b743f51 100644 --- a/google/cloud/logging/_gax.py +++ b/google/cloud/logging/_gax.py @@ -19,6 +19,7 @@ from google.gax import CallOptions from google.gax import INITIAL_PAGE from google.gax.errors import GaxError +from google.gax.grpc import exc_to_code from google.logging.type.log_severity_pb2 import LogSeverity from google.logging.v2.logging_config_pb2 import LogSink from google.logging.v2.logging_metrics_pb2 import LogMetric @@ -29,7 +30,6 @@ # pylint: disable=ungrouped-imports from google.cloud._helpers import _datetime_to_pb_timestamp from google.cloud._helpers import _pb_timestamp_to_rfc3339 -from google.cloud._helpers import exc_to_code from google.cloud.exceptions import Conflict from google.cloud.exceptions import NotFound # pylint: enable=ungrouped-imports diff --git a/google/cloud/pubsub/_gax.py b/google/cloud/pubsub/_gax.py index 2a0b0393aa5b..d51342a7ce45 100644 --- a/google/cloud/pubsub/_gax.py +++ b/google/cloud/pubsub/_gax.py @@ -19,14 +19,14 @@ from google.gax import CallOptions from google.gax import INITIAL_PAGE from google.gax.errors import GaxError +from google.gax.grpc import exc_to_code from google.pubsub.v1.pubsub_pb2 import PubsubMessage from google.pubsub.v1.pubsub_pb2 import PushConfig -from grpc.beta.implementations import insecure_channel +from grpc import insecure_channel from grpc import StatusCode # pylint: disable=ungrouped-imports from google.cloud._helpers import _to_bytes -from google.cloud._helpers import exc_to_code from google.cloud._helpers import _pb_timestamp_to_rfc3339 from google.cloud.exceptions import Conflict from google.cloud.exceptions import NotFound @@ -520,7 +520,7 @@ def make_gax_publisher_api(connection): """ channel = None if connection.in_emulator: - channel = insecure_channel(connection.host, None) + channel = insecure_channel(connection.host) return PublisherApi(channel=channel) @@ -540,5 +540,5 @@ def make_gax_subscriber_api(connection): """ channel = None if connection.in_emulator: - channel = insecure_channel(connection.host, None) + channel = insecure_channel(connection.host) return SubscriberApi(channel=channel) diff --git a/setup.py b/setup.py index f1e9a2eac826..1ea55a2db99e 100644 --- a/setup.py +++ b/setup.py @@ -21,7 +21,7 @@ GRPC_PACKAGES = [ 'grpcio >= 1.0.0', - 'google-gax >= 0.13.0, < 0.14dev', + 'google-gax >= 0.14.1, < 0.15dev', 'gapic-google-pubsub-v1 >= 0.9.0, < 0.10dev', 'grpc-google-pubsub-v1 >= 0.9.0, < 0.10dev', 'gapic-google-logging-v2 >= 0.9.0, < 0.10dev', diff --git a/system_tests/bigtable.py b/system_tests/bigtable.py index af8668031a87..bcb4c41965c2 100644 --- a/system_tests/bigtable.py +++ b/system_tests/bigtable.py @@ -93,7 +93,7 @@ def _retry_on_unavailable(exc): def setUpModule(): - from grpc._channel import _Rendezvous + from google.cloud.exceptions import GrpcRendezvous Config.IN_EMULATOR = os.getenv(BIGTABLE_EMULATOR) is not None @@ -106,7 +106,7 @@ def setUpModule(): Config.INSTANCE = Config.CLIENT.instance(INSTANCE_ID, LOCATION_ID) if not Config.IN_EMULATOR: - retry = RetryErrors(_Rendezvous, + retry = RetryErrors(GrpcRendezvous, error_predicate=_retry_on_unavailable) instances, failed_locations = retry(Config.CLIENT.list_instances)() diff --git a/system_tests/logging_.py b/system_tests/logging_.py index 3ba39edc8912..dad765bfca80 100644 --- a/system_tests/logging_.py +++ b/system_tests/logging_.py @@ -73,9 +73,10 @@ def _logger_name(): return 'system-tests-logger' + unique_resource_id('-') def _list_entries(self, logger): - from grpc._channel import _Rendezvous + from google.gax.errors import GaxError + inner = RetryResult(_has_entries)(logger.list_entries) - outer = RetryErrors(_Rendezvous, _retry_on_unavailable)(inner) + outer = RetryErrors(GaxError, _retry_on_unavailable)(inner) return outer() def test_log_text(self): diff --git a/system_tests/pubsub.py b/system_tests/pubsub.py index 4074d2724e61..880810dad1d3 100644 --- a/system_tests/pubsub.py +++ b/system_tests/pubsub.py @@ -16,12 +16,11 @@ import unittest from google.gax.errors import GaxError +from google.gax.grpc import exc_to_code from grpc import StatusCode -from grpc._channel import _Rendezvous import httplib2 # pylint: disable=ungrouped-imports -from google.cloud import _helpers from google.cloud.environment_vars import PUBSUB_EMULATOR from google.cloud.pubsub import client # pylint: enable=ungrouped-imports @@ -34,10 +33,10 @@ def _unavailable(exc): - return _helpers.exc_to_code(exc) == StatusCode.UNAVAILABLE + return exc_to_code(exc) == StatusCode.UNAVAILABLE -retry_unavailable = RetryErrors((GaxError, _Rendezvous), _unavailable) +retry_unavailable = RetryErrors(GaxError, _unavailable) class Config(object): diff --git a/unit_tests/_testing.py b/unit_tests/_testing.py index d3a1a268b5b4..4ca985ca0086 100644 --- a/unit_tests/_testing.py +++ b/unit_tests/_testing.py @@ -58,12 +58,12 @@ def __init__(self, **kw): self.__dict__.update(kw) def _make_grpc_error(self, status_code): - from grpc._channel import _Rendezvous from grpc._channel import _RPCState + from google.cloud.exceptions import GrpcRendezvous details = 'Some error details.' exc_state = _RPCState((), None, None, status_code, details) - return _Rendezvous(exc_state, None, None, None) + return GrpcRendezvous(exc_state, None, None, None) def _make_grpc_not_found(self): from grpc import StatusCode diff --git a/unit_tests/datastore/test_connection.py b/unit_tests/datastore/test_connection.py index 44b829367b9f..5df1e08c762f 100644 --- a/unit_tests/datastore/test_connection.py +++ b/unit_tests/datastore/test_connection.py @@ -243,24 +243,24 @@ def _commit_failure_helper(self, exc, err_class): @unittest.skipUnless(_HAVE_GRPC, 'No gRPC') def test_commit_failure_aborted(self): from grpc import StatusCode - from grpc._channel import _Rendezvous from grpc._channel import _RPCState from google.cloud.exceptions import Conflict + from google.cloud.exceptions import GrpcRendezvous details = 'Bad things.' exc_state = _RPCState((), None, None, StatusCode.ABORTED, details) - exc = _Rendezvous(exc_state, None, None, None) + exc = GrpcRendezvous(exc_state, None, None, None) self._commit_failure_helper(exc, Conflict) @unittest.skipUnless(_HAVE_GRPC, 'No gRPC') def test_commit_failure_cancelled(self): from grpc import StatusCode - from grpc._channel import _Rendezvous from grpc._channel import _RPCState + from google.cloud.exceptions import GrpcRendezvous exc_state = _RPCState((), None, None, StatusCode.CANCELLED, None) - exc = _Rendezvous(exc_state, None, None, None) - self._commit_failure_helper(exc, _Rendezvous) + exc = GrpcRendezvous(exc_state, None, None, None) + self._commit_failure_helper(exc, GrpcRendezvous) @unittest.skipUnless(_HAVE_GRPC, 'No gRPC') def test_commit_failure_non_grpc_err(self): diff --git a/unit_tests/pubsub/test__gax.py b/unit_tests/pubsub/test__gax.py index 5a401494b924..4dd061d9cf8e 100644 --- a/unit_tests/pubsub/test__gax.py +++ b/unit_tests/pubsub/test__gax.py @@ -787,8 +787,8 @@ def mock_publisher_api(channel): channels.append(channel) return mock_result - def mock_insecure_channel(host, port): - insecure_args.append((host, port)) + def mock_insecure_channel(host): + insecure_args.append(host) return mock_channel host = 'CURR_HOST:1234' @@ -799,7 +799,7 @@ def mock_insecure_channel(host, port): self.assertIs(result, mock_result) self.assertEqual(channels, [mock_channel]) - self.assertEqual(insecure_args, [(host, None)]) + self.assertEqual(insecure_args, [host]) @unittest.skipUnless(_HAVE_GAX, 'No gax-python') @@ -840,8 +840,8 @@ def mock_subscriber_api(channel): channels.append(channel) return mock_result - def mock_insecure_channel(host, port): - insecure_args.append((host, port)) + def mock_insecure_channel(host): + insecure_args.append(host) return mock_channel host = 'CURR_HOST:1234' @@ -852,7 +852,7 @@ def mock_insecure_channel(host, port): self.assertIs(result, mock_result) self.assertEqual(channels, [mock_channel]) - self.assertEqual(insecure_args, [(host, None)]) + self.assertEqual(insecure_args, [host]) class _GAXPublisherAPI(_GAXBaseAPI): diff --git a/unit_tests/test__helpers.py b/unit_tests/test__helpers.py index e315df6e6f2b..3a2950c42bbd 100644 --- a/unit_tests/test__helpers.py +++ b/unit_tests/test__helpers.py @@ -1025,37 +1025,6 @@ def test_without_port_argument(self): self._helper(host, host) -class Test_exc_to_code(unittest.TestCase): - - def _callFUT(self, exc): - from google.cloud._helpers import exc_to_code - return exc_to_code(exc) - - def test_with_stable(self): - from grpc._channel import _Rendezvous - from grpc._channel import _RPCState - from grpc import StatusCode - - status_code = StatusCode.FAILED_PRECONDITION - exc_state = _RPCState((), None, None, status_code, None) - exc = _Rendezvous(exc_state, None, None, None) - result = self._callFUT(exc) - self.assertEqual(result, status_code) - - def test_with_beta(self): - from grpc import StatusCode - from grpc.framework.interfaces.face.face import AbortionError - - status_code = StatusCode.UNIMPLEMENTED - exc = AbortionError(None, None, status_code, None) - result = self._callFUT(exc) - self.assertEqual(result, status_code) - - def test_with_none(self): - result = self._callFUT(None) - self.assertIsNone(result) - - class _AppIdentity(object): def __init__(self, app_id):