Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Bigtable: add explicit coverage for 'row_data._retry_read_rows_exception'. #6364

Merged
merged 1 commit into from
Nov 1, 2018
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 65 additions & 0 deletions bigtable/tests/unit/test_row_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -320,6 +320,71 @@ class _Client(object):
data_stub = None


class Test_retry_read_rows_exception(unittest.TestCase):

@staticmethod
def _call_fut(exc):
from google.cloud.bigtable.row_data import _retry_read_rows_exception

return _retry_read_rows_exception(exc)

@staticmethod
def _make_grpc_call_error(exception):
from grpc import Call
from grpc import RpcError

class TestingException(Call, RpcError):
def __init__(self, exception):
self.exception = exception

def code(self):
return self.exception.grpc_status_code

def details(self):
return 'Testing'

return TestingException(exception)

def test_w_miss(self):
from google.api_core.exceptions import Conflict

exception = Conflict('testing')
self.assertFalse(self._call_fut(exception))

def test_w_service_unavailable(self):
from google.api_core.exceptions import ServiceUnavailable

exception = ServiceUnavailable('testing')
self.assertTrue(self._call_fut(exception))

def test_w_deadline_exceeded(self):
from google.api_core.exceptions import DeadlineExceeded

exception = DeadlineExceeded('testing')
self.assertTrue(self._call_fut(exception))

def test_w_miss_wrapped_in_grpc(self):
from google.api_core.exceptions import Conflict

wrapped = Conflict('testing')
exception = self._make_grpc_call_error(wrapped)
self.assertFalse(self._call_fut(exception))

def test_w_service_unavailable_wrapped_in_grpc(self):
from google.api_core.exceptions import ServiceUnavailable

wrapped = ServiceUnavailable('testing')
exception = self._make_grpc_call_error(wrapped)
self.assertTrue(self._call_fut(exception))

def test_w_deadline_exceeded_wrapped_in_grpc(self):
from google.api_core.exceptions import DeadlineExceeded

wrapped = DeadlineExceeded('testing')
exception = self._make_grpc_call_error(wrapped)
self.assertTrue(self._call_fut(exception))


class TestPartialRowsData(unittest.TestCase):
ROW_KEY = b'row-key'
FAMILY_NAME = u'family'
Expand Down