1 files changed, 353 insertions, 0 deletions
diff --git a/tests/unit/test_exceptions.py b/tests/unit/test_exceptions.py
new file mode 100644
index 0000000..622f58a
--- /dev/null
+++ b/tests/unit/test_exceptions.py
@@ -0,0 +1,353 @@
+# Copyright 2014 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import http.client
+import json
+
+import mock
+import pytest
+import requests
+
+try:
+    import grpc
+    from grpc_status import rpc_status
+except ImportError:
+    grpc = rpc_status = None
+
+from google.api_core import exceptions
+from google.protobuf import any_pb2, json_format
+from google.rpc import error_details_pb2, status_pb2
+
+
+def test_create_google_cloud_error():
+    exception = exceptions.GoogleAPICallError("Testing")
+    exception.code = 600
+    assert str(exception) == "600 Testing"
+    assert exception.message == "Testing"
+    assert exception.errors == []
+    assert exception.response is None
+
+
+def test_create_google_cloud_error_with_args():
+    error = {
+        "code": 600,
+        "message": "Testing",
+    }
+    response = mock.sentinel.response
+    exception = exceptions.GoogleAPICallError("Testing", [error], response=response)
+    exception.code = 600
+    assert str(exception) == "600 Testing"
+    assert exception.message == "Testing"
+    assert exception.errors == [error]
+    assert exception.response == response
+
+
+def test_from_http_status():
+    message = "message"
+    exception = exceptions.from_http_status(http.client.NOT_FOUND, message)
+    assert exception.code == http.client.NOT_FOUND
+    assert exception.message == message
+    assert exception.errors == []
+
+
+def test_from_http_status_with_errors_and_response():
+    message = "message"
+    errors = ["1", "2"]
+    response = mock.sentinel.response
+    exception = exceptions.from_http_status(
+        http.client.NOT_FOUND, message, errors=errors, response=response
+    )
+
+    assert isinstance(exception, exceptions.NotFound)
+    assert exception.code == http.client.NOT_FOUND
+    assert exception.message == message
+    assert exception.errors == errors
+    assert exception.response == response
+
+
+def test_from_http_status_unknown_code():
+    message = "message"
+    status_code = 156
+    exception = exceptions.from_http_status(status_code, message)
+    assert exception.code == status_code
+    assert exception.message == message
+
+
+def make_response(content):
+    response = requests.Response()
+    response._content = content
+    response.status_code = http.client.NOT_FOUND
+    response.request = requests.Request(
+        method="POST", url="https://example.com"
+    ).prepare()
+    return response
+
+
+def test_from_http_response_no_content():
+    response = make_response(None)
+
+    exception = exceptions.from_http_response(response)
+
+    assert isinstance(exception, exceptions.NotFound)
+    assert exception.code == http.client.NOT_FOUND
+    assert exception.message == "POST https://example.com/: unknown error"
+    assert exception.response == response
+
+
+def test_from_http_response_text_content():
+    response = make_response(b"message")
+    response.encoding = "UTF8"  # suppress charset_normalizer warning
+
+    exception = exceptions.from_http_response(response)
+
+    assert isinstance(exception, exceptions.NotFound)
+    assert exception.code == http.client.NOT_FOUND
+    assert exception.message == "POST https://example.com/: message"
+
+
+def test_from_http_response_json_content():
+    response = make_response(
+        json.dumps({"error": {"message": "json message", "errors": ["1", "2"]}}).encode(
+            "utf-8"
+        )
+    )
+
+    exception = exceptions.from_http_response(response)
+
+    assert isinstance(exception, exceptions.NotFound)
+    assert exception.code == http.client.NOT_FOUND
+    assert exception.message == "POST https://example.com/: json message"
+    assert exception.errors == ["1", "2"]
+
+
+def test_from_http_response_bad_json_content():
+    response = make_response(json.dumps({"meep": "moop"}).encode("utf-8"))
+
+    exception = exceptions.from_http_response(response)
+
+    assert isinstance(exception, exceptions.NotFound)
+    assert exception.code == http.client.NOT_FOUND
+    assert exception.message == "POST https://example.com/: unknown error"
+
+
+def test_from_http_response_json_unicode_content():
+    response = make_response(
+        json.dumps(
+            {"error": {"message": "\u2019 message", "errors": ["1", "2"]}}
+        ).encode("utf-8")
+    )
+
+    exception = exceptions.from_http_response(response)
+
+    assert isinstance(exception, exceptions.NotFound)
+    assert exception.code == http.client.NOT_FOUND
+    assert exception.message == "POST https://example.com/: \u2019 message"
+    assert exception.errors == ["1", "2"]
+
+
+@pytest.mark.skipif(grpc is None, reason="No grpc")
+def test_from_grpc_status():
+    message = "message"
+    exception = exceptions.from_grpc_status(grpc.StatusCode.OUT_OF_RANGE, message)
+    assert isinstance(exception, exceptions.BadRequest)
+    assert isinstance(exception, exceptions.OutOfRange)
+    assert exception.code == http.client.BAD_REQUEST
+    assert exception.grpc_status_code == grpc.StatusCode.OUT_OF_RANGE
+    assert exception.message == message
+    assert exception.errors == []
+
+
+@pytest.mark.skipif(grpc is None, reason="No grpc")
+def test_from_grpc_status_as_int():
+    message = "message"
+    exception = exceptions.from_grpc_status(11, message)
+    assert isinstance(exception, exceptions.BadRequest)
+    assert isinstance(exception, exceptions.OutOfRange)
+    assert exception.code == http.client.BAD_REQUEST
+    assert exception.grpc_status_code == grpc.StatusCode.OUT_OF_RANGE
+    assert exception.message == message
+    assert exception.errors == []
+
+
+@pytest.mark.skipif(grpc is None, reason="No grpc")
+def test_from_grpc_status_with_errors_and_response():
+    message = "message"
+    response = mock.sentinel.response
+    errors = ["1", "2"]
+    exception = exceptions.from_grpc_status(
+        grpc.StatusCode.OUT_OF_RANGE, message, errors=errors, response=response
+    )
+
+    assert isinstance(exception, exceptions.OutOfRange)
+    assert exception.message == message
+    assert exception.errors == errors
+    assert exception.response == response
+
+
+@pytest.mark.skipif(grpc is None, reason="No grpc")
+def test_from_grpc_status_unknown_code():
+    message = "message"
+    exception = exceptions.from_grpc_status(grpc.StatusCode.OK, message)
+    assert exception.grpc_status_code == grpc.StatusCode.OK
+    assert exception.message == message
+
+
+@pytest.mark.skipif(grpc is None, reason="No grpc")
+def test_from_grpc_error():
+    message = "message"
+    error = mock.create_autospec(grpc.Call, instance=True)
+    error.code.return_value = grpc.StatusCode.INVALID_ARGUMENT
+    error.details.return_value = message
+
+    exception = exceptions.from_grpc_error(error)
+
+    assert isinstance(exception, exceptions.BadRequest)
+    assert isinstance(exception, exceptions.InvalidArgument)
+    assert exception.code == http.client.BAD_REQUEST
+    assert exception.grpc_status_code == grpc.StatusCode.INVALID_ARGUMENT
+    assert exception.message == message
+    assert exception.errors == [error]
+    assert exception.response == error
+
+
+@pytest.mark.skipif(grpc is None, reason="No grpc")
+def test_from_grpc_error_non_call():
+    message = "message"
+    error = mock.create_autospec(grpc.RpcError, instance=True)
+    error.__str__.return_value = message
+
+    exception = exceptions.from_grpc_error(error)
+
+    assert isinstance(exception, exceptions.GoogleAPICallError)
+    assert exception.code is None
+    assert exception.grpc_status_code is None
+    assert exception.message == message
+    assert exception.errors == [error]
+    assert exception.response == error
+
+
+@pytest.mark.skipif(grpc is None, reason="No grpc")
+def test_from_grpc_error_bare_call():
+    message = "Testing"
+
+    class TestingError(grpc.Call, grpc.RpcError):
+        def __init__(self, exception):
+            self.exception = exception
+
+        def code(self):
+            return self.exception.grpc_status_code
+
+        def details(self):
+            return message
+
+    nested_message = "message"
+    error = TestingError(exceptions.GoogleAPICallError(nested_message))
+
+    exception = exceptions.from_grpc_error(error)
+
+    assert isinstance(exception, exceptions.GoogleAPICallError)
+    assert exception.code is None
+    assert exception.grpc_status_code is None
+    assert exception.message == message
+    assert exception.errors == [error]
+    assert exception.response == error
+    assert exception.details == []
+
+
+def create_bad_request_details():
+    bad_request_details = error_details_pb2.BadRequest()
+    field_violation = bad_request_details.field_violations.add()
+    field_violation.field = "document.content"
+    field_violation.description = "Must have some text content to annotate."
+    status_detail = any_pb2.Any()
+    status_detail.Pack(bad_request_details)
+    return status_detail
+
+
+def test_error_details_from_rest_response():
+    bad_request_detail = create_bad_request_details()
+    status = status_pb2.Status()
+    status.code = 3
+    status.message = (
+        "3 INVALID_ARGUMENT: One of content, or gcs_content_uri must be set."
+    )
+    status.details.append(bad_request_detail)
+
+    # See JSON schema in https://cloud.google.com/apis/design/errors#http_mapping
+    http_response = make_response(
+        json.dumps({"error": json.loads(json_format.MessageToJson(status))}).encode(
+            "utf-8"
+        )
+    )
+    exception = exceptions.from_http_response(http_response)
+    want_error_details = [json.loads(json_format.MessageToJson(bad_request_detail))]
+    assert want_error_details == exception.details
+    # 404 POST comes from make_response.
+    assert str(exception) == (
+        "404 POST https://example.com/: 3 INVALID_ARGUMENT:"
+        " One of content, or gcs_content_uri must be set."
+        " [{'@type': 'type.googleapis.com/google.rpc.BadRequest',"
+        " 'fieldViolations': [{'field': 'document.content',"
+        " 'description': 'Must have some text content to annotate.'}]}]"
+    )
+
+
+def test_error_details_from_v1_rest_response():
+    response = make_response(
+        json.dumps(
+            {"error": {"message": "\u2019 message", "errors": ["1", "2"]}}
+        ).encode("utf-8")
+    )
+    exception = exceptions.from_http_response(response)
+    assert exception.details == []
+
+
+@pytest.mark.skipif(grpc is None, reason="gRPC not importable")
+def test_error_details_from_grpc_response():
+    status = rpc_status.status_pb2.Status()
+    status.code = 3
+    status.message = (
+        "3 INVALID_ARGUMENT: One of content, or gcs_content_uri must be set."
+    )
+    status_detail = create_bad_request_details()
+    status.details.append(status_detail)
+
+    # Actualy error doesn't matter as long as its grpc.Call,
+    # because from_call is mocked.
+    error = mock.create_autospec(grpc.Call, instance=True)
+    with mock.patch("grpc_status.rpc_status.from_call") as m:
+        m.return_value = status
+        exception = exceptions.from_grpc_error(error)
+
+    bad_request_detail = error_details_pb2.BadRequest()
+    status_detail.Unpack(bad_request_detail)
+    assert exception.details == [bad_request_detail]
+
+
+@pytest.mark.skipif(grpc is None, reason="gRPC not importable")
+def test_error_details_from_grpc_response_unknown_error():
+    status_detail = any_pb2.Any()
+
+    status = rpc_status.status_pb2.Status()
+    status.code = 3
+    status.message = (
+        "3 INVALID_ARGUMENT: One of content, or gcs_content_uri must be set."
+    )
+    status.details.append(status_detail)
+
+    error = mock.create_autospec(grpc.Call, instance=True)
+    with mock.patch("grpc_status.rpc_status.from_call") as m:
+        m.return_value = status
+        exception = exceptions.from_grpc_error(error)
+    assert exception.details == [status_detail]