1# Copyright 2014 Google LLC 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14 15import http.client 16import json 17 18import mock 19import pytest 20import requests 21 22try: 23 import grpc 24 from grpc_status import rpc_status 25except ImportError: 26 grpc = rpc_status = None 27 28from google.api_core import exceptions 29from google.protobuf import any_pb2, json_format 30from google.rpc import error_details_pb2, status_pb2 31 32 33def test_create_google_cloud_error(): 34 exception = exceptions.GoogleAPICallError("Testing") 35 exception.code = 600 36 assert str(exception) == "600 Testing" 37 assert exception.message == "Testing" 38 assert exception.errors == [] 39 assert exception.response is None 40 41 42def test_create_google_cloud_error_with_args(): 43 error = { 44 "code": 600, 45 "message": "Testing", 46 } 47 response = mock.sentinel.response 48 exception = exceptions.GoogleAPICallError("Testing", [error], response=response) 49 exception.code = 600 50 assert str(exception) == "600 Testing" 51 assert exception.message == "Testing" 52 assert exception.errors == [error] 53 assert exception.response == response 54 55 56def test_from_http_status(): 57 message = "message" 58 exception = exceptions.from_http_status(http.client.NOT_FOUND, message) 59 assert exception.code == http.client.NOT_FOUND 60 assert exception.message == message 61 assert exception.errors == [] 62 63 64def test_from_http_status_with_errors_and_response(): 65 message = "message" 66 errors = ["1", "2"] 67 response = mock.sentinel.response 68 exception = exceptions.from_http_status( 69 http.client.NOT_FOUND, message, errors=errors, response=response 70 ) 71 72 assert isinstance(exception, exceptions.NotFound) 73 assert exception.code == http.client.NOT_FOUND 74 assert exception.message == message 75 assert exception.errors == errors 76 assert exception.response == response 77 78 79def test_from_http_status_unknown_code(): 80 message = "message" 81 status_code = 156 82 exception = exceptions.from_http_status(status_code, message) 83 assert exception.code == status_code 84 assert exception.message == message 85 86 87def make_response(content): 88 response = requests.Response() 89 response._content = content 90 response.status_code = http.client.NOT_FOUND 91 response.request = requests.Request( 92 method="POST", url="https://example.com" 93 ).prepare() 94 return response 95 96 97def test_from_http_response_no_content(): 98 response = make_response(None) 99 100 exception = exceptions.from_http_response(response) 101 102 assert isinstance(exception, exceptions.NotFound) 103 assert exception.code == http.client.NOT_FOUND 104 assert exception.message == "POST https://example.com/: unknown error" 105 assert exception.response == response 106 107 108def test_from_http_response_text_content(): 109 response = make_response(b"message") 110 response.encoding = "UTF8" # suppress charset_normalizer warning 111 112 exception = exceptions.from_http_response(response) 113 114 assert isinstance(exception, exceptions.NotFound) 115 assert exception.code == http.client.NOT_FOUND 116 assert exception.message == "POST https://example.com/: message" 117 118 119def test_from_http_response_json_content(): 120 response = make_response( 121 json.dumps({"error": {"message": "json message", "errors": ["1", "2"]}}).encode( 122 "utf-8" 123 ) 124 ) 125 126 exception = exceptions.from_http_response(response) 127 128 assert isinstance(exception, exceptions.NotFound) 129 assert exception.code == http.client.NOT_FOUND 130 assert exception.message == "POST https://example.com/: json message" 131 assert exception.errors == ["1", "2"] 132 133 134def test_from_http_response_bad_json_content(): 135 response = make_response(json.dumps({"meep": "moop"}).encode("utf-8")) 136 137 exception = exceptions.from_http_response(response) 138 139 assert isinstance(exception, exceptions.NotFound) 140 assert exception.code == http.client.NOT_FOUND 141 assert exception.message == "POST https://example.com/: unknown error" 142 143 144def test_from_http_response_json_unicode_content(): 145 response = make_response( 146 json.dumps( 147 {"error": {"message": "\u2019 message", "errors": ["1", "2"]}} 148 ).encode("utf-8") 149 ) 150 151 exception = exceptions.from_http_response(response) 152 153 assert isinstance(exception, exceptions.NotFound) 154 assert exception.code == http.client.NOT_FOUND 155 assert exception.message == "POST https://example.com/: \u2019 message" 156 assert exception.errors == ["1", "2"] 157 158 159@pytest.mark.skipif(grpc is None, reason="No grpc") 160def test_from_grpc_status(): 161 message = "message" 162 exception = exceptions.from_grpc_status(grpc.StatusCode.OUT_OF_RANGE, message) 163 assert isinstance(exception, exceptions.BadRequest) 164 assert isinstance(exception, exceptions.OutOfRange) 165 assert exception.code == http.client.BAD_REQUEST 166 assert exception.grpc_status_code == grpc.StatusCode.OUT_OF_RANGE 167 assert exception.message == message 168 assert exception.errors == [] 169 170 171@pytest.mark.skipif(grpc is None, reason="No grpc") 172def test_from_grpc_status_as_int(): 173 message = "message" 174 exception = exceptions.from_grpc_status(11, message) 175 assert isinstance(exception, exceptions.BadRequest) 176 assert isinstance(exception, exceptions.OutOfRange) 177 assert exception.code == http.client.BAD_REQUEST 178 assert exception.grpc_status_code == grpc.StatusCode.OUT_OF_RANGE 179 assert exception.message == message 180 assert exception.errors == [] 181 182 183@pytest.mark.skipif(grpc is None, reason="No grpc") 184def test_from_grpc_status_with_errors_and_response(): 185 message = "message" 186 response = mock.sentinel.response 187 errors = ["1", "2"] 188 exception = exceptions.from_grpc_status( 189 grpc.StatusCode.OUT_OF_RANGE, message, errors=errors, response=response 190 ) 191 192 assert isinstance(exception, exceptions.OutOfRange) 193 assert exception.message == message 194 assert exception.errors == errors 195 assert exception.response == response 196 197 198@pytest.mark.skipif(grpc is None, reason="No grpc") 199def test_from_grpc_status_unknown_code(): 200 message = "message" 201 exception = exceptions.from_grpc_status(grpc.StatusCode.OK, message) 202 assert exception.grpc_status_code == grpc.StatusCode.OK 203 assert exception.message == message 204 205 206@pytest.mark.skipif(grpc is None, reason="No grpc") 207def test_from_grpc_error(): 208 message = "message" 209 error = mock.create_autospec(grpc.Call, instance=True) 210 error.code.return_value = grpc.StatusCode.INVALID_ARGUMENT 211 error.details.return_value = message 212 213 exception = exceptions.from_grpc_error(error) 214 215 assert isinstance(exception, exceptions.BadRequest) 216 assert isinstance(exception, exceptions.InvalidArgument) 217 assert exception.code == http.client.BAD_REQUEST 218 assert exception.grpc_status_code == grpc.StatusCode.INVALID_ARGUMENT 219 assert exception.message == message 220 assert exception.errors == [error] 221 assert exception.response == error 222 223 224@pytest.mark.skipif(grpc is None, reason="No grpc") 225def test_from_grpc_error_non_call(): 226 message = "message" 227 error = mock.create_autospec(grpc.RpcError, instance=True) 228 error.__str__.return_value = message 229 230 exception = exceptions.from_grpc_error(error) 231 232 assert isinstance(exception, exceptions.GoogleAPICallError) 233 assert exception.code is None 234 assert exception.grpc_status_code is None 235 assert exception.message == message 236 assert exception.errors == [error] 237 assert exception.response == error 238 239 240def create_bad_request_details(): 241 bad_request_details = error_details_pb2.BadRequest() 242 field_violation = bad_request_details.field_violations.add() 243 field_violation.field = "document.content" 244 field_violation.description = "Must have some text content to annotate." 245 status_detail = any_pb2.Any() 246 status_detail.Pack(bad_request_details) 247 return status_detail 248 249 250def test_error_details_from_rest_response(): 251 bad_request_detail = create_bad_request_details() 252 status = status_pb2.Status() 253 status.code = 3 254 status.message = ( 255 "3 INVALID_ARGUMENT: One of content, or gcs_content_uri must be set." 256 ) 257 status.details.append(bad_request_detail) 258 259 # See JSON schema in https://cloud.google.com/apis/design/errors#http_mapping 260 http_response = make_response( 261 json.dumps({"error": json.loads(json_format.MessageToJson(status))}).encode( 262 "utf-8" 263 ) 264 ) 265 exception = exceptions.from_http_response(http_response) 266 want_error_details = [json.loads(json_format.MessageToJson(bad_request_detail))] 267 assert want_error_details == exception.details 268 # 404 POST comes from make_response. 269 assert str(exception) == ( 270 "404 POST https://example.com/: 3 INVALID_ARGUMENT:" 271 " One of content, or gcs_content_uri must be set." 272 " [{'@type': 'type.googleapis.com/google.rpc.BadRequest'," 273 " 'fieldViolations': [{'field': 'document.content'," 274 " 'description': 'Must have some text content to annotate.'}]}]" 275 ) 276 277 278def test_error_details_from_v1_rest_response(): 279 response = make_response( 280 json.dumps( 281 {"error": {"message": "\u2019 message", "errors": ["1", "2"]}} 282 ).encode("utf-8") 283 ) 284 exception = exceptions.from_http_response(response) 285 assert exception.details == [] 286 287 288@pytest.mark.skipif(grpc is None, reason="gRPC not importable") 289def test_error_details_from_grpc_response(): 290 status = rpc_status.status_pb2.Status() 291 status.code = 3 292 status.message = ( 293 "3 INVALID_ARGUMENT: One of content, or gcs_content_uri must be set." 294 ) 295 status_detail = create_bad_request_details() 296 status.details.append(status_detail) 297 298 # Actualy error doesn't matter as long as its grpc.Call, 299 # because from_call is mocked. 300 error = mock.create_autospec(grpc.Call, instance=True) 301 with mock.patch("grpc_status.rpc_status.from_call") as m: 302 m.return_value = status 303 exception = exceptions.from_grpc_error(error) 304 305 bad_request_detail = error_details_pb2.BadRequest() 306 status_detail.Unpack(bad_request_detail) 307 assert exception.details == [bad_request_detail] 308 309 310@pytest.mark.skipif(grpc is None, reason="gRPC not importable") 311def test_error_details_from_grpc_response_unknown_error(): 312 status_detail = any_pb2.Any() 313 314 status = rpc_status.status_pb2.Status() 315 status.code = 3 316 status.message = ( 317 "3 INVALID_ARGUMENT: One of content, or gcs_content_uri must be set." 318 ) 319 status.details.append(status_detail) 320 321 error = mock.create_autospec(grpc.Call, instance=True) 322 with mock.patch("grpc_status.rpc_status.from_call") as m: 323 m.return_value = status 324 exception = exceptions.from_grpc_error(error) 325 assert exception.details == [status_detail] 326