blob: 6e65c402180779b175f3b3d47f8d06e0d66d4fed [file] [log] [blame]
#!/usr/bin/env python
#
# Copyright 2007 Google Inc.
#
# 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.
#
"""Unit tests for the endpoints_server module."""
import httplib
import json
import logging
import unittest
import google
import mox
from google.appengine.tools.devappserver2 import dispatcher
from google.appengine.tools.devappserver2.endpoints import api_config_manager
from google.appengine.tools.devappserver2.endpoints import api_request
from google.appengine.tools.devappserver2.endpoints import discovery_api_proxy
from google.appengine.tools.devappserver2.endpoints import endpoints_server
from google.appengine.tools.devappserver2.endpoints import errors
from google.appengine.tools.devappserver2.endpoints import test_utils
class JsonMatches(mox.Comparator):
"""A Mox comparator to compare a string of a JSON object to a JSON object."""
def __init__(self, json_object):
"""Constructor.
Args:
json_object: The JSON object to compare against.
"""
self._json_object = json_object
def equals(self, json_string):
"""Check if the given object matches our json object.
This converts json_string from a string to a JSON object, then compares it
against our json object.
Args:
json_string: A string containing a JSON object to be compared against.
Returns:
True if the object matches, False if not.
"""
other_json = json.loads(json_string)
return self._json_object == other_json
def __repr__(self):
return '<JsonMatches %r>' % self._json_object
class DevAppserverEndpointsServerTest(test_utils.TestsWithStartResponse):
def setUp(self):
"""Set up a dev Endpoints server."""
super(DevAppserverEndpointsServerTest, self).setUp()
self.mox = mox.Mox()
self.config_manager = api_config_manager.ApiConfigManager()
self.mock_dispatcher = self.mox.CreateMock(dispatcher.Dispatcher)
self.server = endpoints_server.EndpointsDispatcher(self.mock_dispatcher,
self.config_manager)
def tearDown(self):
self.mox.UnsetStubs()
def prepare_dispatch(self, config):
# The dispatch call will make a call to get_api_configs, making a
# dispatcher request. Set up that request.
request_method = 'POST'
request_path = '/_ah/spi/BackendService.getApiConfigs'
request_headers = [('Content-Type', 'application/json')]
request_body = '{}'
response_body = json.dumps({'items': [config]})
self.mock_dispatcher.add_request(
request_method, request_path, request_headers, request_body,
endpoints_server._SERVER_SOURCE_IP).AndReturn(
dispatcher.ResponseTuple('200 OK',
[('Content-Type', 'application/json'),
('Content-Length',
str(len(response_body)))],
response_body))
def assert_dispatch_to_spi(self, request, config, spi_path,
expected_spi_body_json=None):
"""Assert that dispatching a request to the SPI works.
Mock out the dispatcher.add_request and handle_spi_response, and use these
to ensure that the correct request is being sent to the back end when
Dispatch is called.
Args:
request: An ApiRequest, the request to dispatch.
config: A dict containing the API configuration.
spi_path: A string containing the relative path to the SPI.
expected_spi_body_json: If not None, this is a JSON object containing
the mock response sent by the back end. If None, this will create an
empty response.
"""
self.prepare_dispatch(config)
spi_headers = [('Content-Type', 'application/json')]
spi_body_json = expected_spi_body_json or {}
spi_response = dispatcher.ResponseTuple('200 OK', [], 'Test')
self.mock_dispatcher.add_request(
'POST', spi_path, spi_headers, JsonMatches(spi_body_json),
request.source_ip).AndReturn(spi_response)
self.mox.StubOutWithMock(self.server, 'handle_spi_response')
self.server.handle_spi_response(
mox.IsA(api_request.ApiRequest), mox.IsA(api_request.ApiRequest),
spi_response, mox.IsA(dict), self.start_response).AndReturn('Test')
# Run the test.
self.mox.ReplayAll()
response = self.server.dispatch(request, self.start_response)
self.mox.VerifyAll()
self.assertEqual('Test', response)
def test_dispatch_invalid_path(self):
config = json.dumps({
'name': 'guestbook_api',
'version': 'v1',
'methods': {
'guestbook.get': {
'httpMethod': 'GET',
'path': 'greetings/{gid}',
'rosyMethod': 'MyApi.greetings_get'
}
}
})
request = test_utils.build_request('/_ah/api/foo')
self.prepare_dispatch(config)
self.mox.ReplayAll()
response = self.server.dispatch(request, self.start_response)
self.mox.VerifyAll()
self.assert_http_match(response, 404,
[('Content-Type', 'text/plain'),
('Content-Length', '9')],
'Not Found')
def test_dispatch_invalid_enum(self):
config = json.dumps({
'name': 'guestbook_api',
'version': 'v1',
'methods': {
'guestbook.get': {
'httpMethod': 'GET',
'path': 'greetings/{gid}',
'rosyMethod': 'MyApi.greetings_get',
'request': {
'body': 'empty',
'parameters': {'gid': {'enum': {'X': {'backendValue': 'X'}},
'type': 'string'
}
}
}
}
}
})
request = test_utils.build_request(
'/_ah/api/guestbook_api/v1/greetings/invalid_enum')
self.prepare_dispatch(config)
self.mox.ReplayAll()
response = self.server.dispatch(request, self.start_response)
self.mox.VerifyAll()
logging.warning('Config %s', self.server.config_manager.configs)
self.assertEqual(self.response_status, '400 Bad Request')
body = ''.join(response)
body_json = json.loads(body)
self.assertEqual(1, len(body_json['error']['errors']))
self.assertEqual('gid', body_json['error']['errors'][0]['location'])
self.assertEqual('invalidParameter',
body_json['error']['errors'][0]['reason'])
def test_dispatch_spi_error(self):
"""Check the error response if the SPI returns an error."""
config = json.dumps({
'name': 'guestbook_api',
'version': 'v1',
'methods': {
'guestbook.get': {
'httpMethod': 'GET',
'path': 'greetings/{gid}',
'rosyMethod': 'MyApi.greetings_get'
}
}
})
request = test_utils.build_request('/_ah/api/foo')
self.prepare_dispatch(config)
self.mox.StubOutWithMock(self.server, 'call_spi')
# The application chose to throw a 404 error.
response = dispatcher.ResponseTuple('404 Not Found', [],
('{"state": "APPLICATION_ERROR",'
' "error_message": "Test error"}'))
self.server.call_spi(request, mox.IgnoreArg()).AndRaise(
errors.BackendError(response))
self.mox.ReplayAll()
response = self.server.dispatch(request, self.start_response)
self.mox.VerifyAll()
expected_response = (
'{\n'
' "error": {\n'
' "code": 404, \n'
' "errors": [\n'
' {\n'
' "domain": "global", \n'
' "message": "Test error", \n'
' "reason": "notFound"\n'
' }\n'
' ], \n'
' "message": "Test error"\n'
' }\n'
'}')
response = ''.join(response)
self.assert_http_match(response, '404 Not Found',
[('Content-Length', '%d' % len(expected_response)),
('Content-Type', 'application/json')],
expected_response)
def test_dispatch_rpc_error(self):
"""Test than an RPC call that returns an error is handled properly."""
config = json.dumps({
'name': 'guestbook_api',
'version': 'v1',
'methods': {
'guestbook.get': {
'httpMethod': 'GET',
'path': 'greetings/{gid}',
'rosyMethod': 'MyApi.greetings_get'
}
}
})
request = test_utils.build_request(
'/_ah/api/rpc',
'{"method": "foo.bar", "apiVersion": "X", "id": "gapiRpc"}')
self.prepare_dispatch(config)
self.mox.StubOutWithMock(self.server, 'call_spi')
# The application chose to throw a 404 error.
response = dispatcher.ResponseTuple('404 Not Found', [],
('{"state": "APPLICATION_ERROR",'
' "error_message": "Test error"}'))
self.server.call_spi(request, mox.IgnoreArg()).AndRaise(
errors.BackendError(response))
self.mox.ReplayAll()
response = self.server.dispatch(request, self.start_response)
self.mox.VerifyAll()
expected_response = {'error': {'code': 404,
'message': 'Test error',
'data': [{
'domain': 'global',
'reason': 'notFound',
'message': 'Test error',
}]
},
'id': 'gapiRpc'
}
response = ''.join(response)
self.assertEqual('200 OK', self.response_status)
self.assertEqual(expected_response, json.loads(response))
def test_dispatch_json_rpc(self):
config = json.dumps({
'name': 'guestbook_api',
'version': 'X',
'methods': {
'foo.bar': {
'httpMethod': 'GET',
'path': 'greetings/{gid}',
'rosyMethod': 'baz.bim'
}
}
})
request = test_utils.build_request(
'/_ah/api/rpc',
'{"method": "foo.bar", "apiVersion": "X"}')
self.assert_dispatch_to_spi(request, config,
'/_ah/spi/baz.bim')
def test_dispatch_rest(self):
config = json.dumps({
'name': 'myapi',
'version': 'v1',
'methods': {
'bar': {
'httpMethod': 'GET',
'path': 'foo/{id}',
'rosyMethod': 'baz.bim'
}
}
})
request = test_utils.build_request('/_ah/api/myapi/v1/foo/testId')
self.assert_dispatch_to_spi(request, config,
'/_ah/spi/baz.bim',
{'id': 'testId'})
def test_explorer_redirect(self):
request = test_utils.build_request('/_ah/api/explorer')
response = self.server.dispatch(request, self.start_response)
self.assert_http_match(response, 302,
[('Content-Length', '0'),
('Location', ('http://apis-explorer.appspot.com/'
'apis-explorer/?base='
'http://localhost:42/_ah/api'))],
'')
def test_static_existing_file(self):
relative_url = '/_ah/api/static/proxy.html'
# Set up mocks for the call to DiscoveryApiProxy.get_static_file.
discovery_api = self.mox.CreateMock(
discovery_api_proxy.DiscoveryApiProxy)
self.mox.StubOutWithMock(discovery_api_proxy, 'DiscoveryApiProxy')
discovery_api_proxy.DiscoveryApiProxy().AndReturn(discovery_api)
static_response = self.mox.CreateMock(httplib.HTTPResponse)
static_response.status = 200
static_response.reason = 'OK'
static_response.getheader('Content-Type').AndReturn('test/type')
test_body = 'test body'
discovery_api.get_static_file(relative_url).AndReturn(
(static_response, test_body))
# Make sure the dispatch works as expected.
request = test_utils.build_request(relative_url)
self.mox.ReplayAll()
response = self.server.dispatch(request, self.start_response)
self.mox.VerifyAll()
response = ''.join(response)
self.assert_http_match(response, '200 OK',
[('Content-Length', '%d' % len(test_body)),
('Content-Type', 'test/type')],
test_body)
def test_static_non_existing_file(self):
relative_url = '/_ah/api/static/blah.html'
# Set up mocks for the call to DiscoveryApiProxy.get_static_file.
discovery_api = self.mox.CreateMock(
discovery_api_proxy.DiscoveryApiProxy)
self.mox.StubOutWithMock(discovery_api_proxy, 'DiscoveryApiProxy')
discovery_api_proxy.DiscoveryApiProxy().AndReturn(discovery_api)
static_response = self.mox.CreateMock(httplib.HTTPResponse)
static_response.status = 404
static_response.reason = 'Not Found'
static_response.getheaders().AndReturn([('Content-Type', 'test/type')])
test_body = 'No Body'
discovery_api.get_static_file(relative_url).AndReturn(
(static_response, test_body))
# Make sure the dispatch works as expected.
request = test_utils.build_request(relative_url)
self.mox.ReplayAll()
response = self.server.dispatch(request, self.start_response)
self.mox.VerifyAll()
response = ''.join(response)
self.assert_http_match(response, '404 Not Found',
[('Content-Length', '%d' % len(test_body)),
('Content-Type', 'test/type')],
test_body)
def test_handle_non_json_spi_response(self):
orig_request = test_utils.build_request('/_ah/api/fake/path')
spi_request = orig_request.copy()
spi_response = dispatcher.ResponseTuple(
200, [('Content-type', 'text/plain')],
'This is an invalid response.')
response = self.server.handle_spi_response(orig_request, spi_request,
spi_response, {},
self.start_response)
error_json = {'error': {'message':
'Non-JSON reply: This is an invalid response.'}}
body = json.dumps(error_json)
self.assert_http_match(response, '500',
[('Content-Type', 'application/json'),
('Content-Length', '%d' % len(body))],
body)
def test_handle_non_json_spi_response_cors(self):
"""Test that an error response still handles CORS headers."""
server_response = dispatcher.ResponseTuple(
'200 OK', [('Content-type', 'text/plain')],
'This is an invalid response.')
response = self.check_cors([('origin', 'test.com')], True, 'test.com',
server_response=server_response)
self.assertEqual(
{'error': {'message': 'Non-JSON reply: This is an invalid response.'}},
json.loads(response))
def check_cors(self, request_headers, expect_response, expected_origin=None,
expected_allow_headers=None, server_response=None):
"""Check that CORS headers are handled correctly.
Args:
request_headers: A list of (header, value), to be used as headers in the
request.
expect_response: A boolean, whether or not CORS headers are expected in
the response.
expected_origin: A string or None. If this is a string, this is the value
that's expected in the response's allow origin header. This can be
None if expect_response is False.
expected_allow_headers: A string or None. If this is a string, this is
the value that's expected in the response's allow headers header. If
this is None, then the response shouldn't have any allow headers
headers.
server_response: A dispatcher.ResponseTuple or None. The backend's
response, to be wrapped and returned as the server's response. If
this is None, a generic response will be generated.
Returns:
A string containing the body of the response that would be sent.
"""
orig_request = test_utils.build_request('/_ah/api/fake/path',
http_headers=request_headers)
spi_request = orig_request.copy()
if server_response is None:
server_response = dispatcher.ResponseTuple(
'200 OK', [('Content-type', 'application/json')], '{}')
response = self.server.handle_spi_response(orig_request, spi_request,
server_response, {},
self.start_response)
headers = dict(self.response_headers)
if expect_response:
self.assertIn(endpoints_server._CORS_HEADER_ALLOW_ORIGIN, headers)
self.assertEqual(
headers[endpoints_server._CORS_HEADER_ALLOW_ORIGIN],
expected_origin)
self.assertIn(endpoints_server._CORS_HEADER_ALLOW_METHODS, headers)
self.assertEqual(set(headers[
endpoints_server._CORS_HEADER_ALLOW_METHODS].split(',')),
endpoints_server._CORS_ALLOWED_METHODS)
if expected_allow_headers is not None:
self.assertIn(endpoints_server._CORS_HEADER_ALLOW_HEADERS,
headers)
self.assertEqual(
headers[endpoints_server._CORS_HEADER_ALLOW_HEADERS],
expected_allow_headers)
else:
self.assertNotIn(endpoints_server._CORS_HEADER_ALLOW_HEADERS,
headers)
else:
self.assertNotIn(endpoints_server._CORS_HEADER_ALLOW_ORIGIN,
headers)
self.assertNotIn(endpoints_server._CORS_HEADER_ALLOW_METHODS,
headers)
self.assertNotIn(endpoints_server._CORS_HEADER_ALLOW_HEADERS,
headers)
return ''.join(response)
def test_handle_cors(self):
"""Test CORS support on a regular request."""
self.check_cors([('origin', 'test.com')], True, 'test.com')
def test_handle_cors_preflight(self):
"""Test a CORS preflight request."""
self.check_cors([('origin', 'http://example.com'),
('Access-control-request-method', 'GET')], True,
'http://example.com')
def test_handle_cors_preflight_invalid(self):
"""Test a CORS preflight request for an unaccepted OPTIONS request."""
self.check_cors([('origin', 'http://example.com'),
('Access-control-request-method', 'OPTIONS')], False)
def test_handle_cors_preflight_request_headers(self):
"""Test a CORS preflight request."""
self.check_cors([('origin', 'http://example.com'),
('Access-control-request-method', 'GET'),
('Access-Control-Request-Headers', 'Date,Expires')], True,
'http://example.com', 'Date,Expires')
def test_lily_uses_python_method_name(self):
"""Verify Lily protocol correctly uses python method name.
This test verifies the fix to http://b/7189819
"""
config = json.dumps({
'name': 'guestbook_api',
'version': 'X',
'methods': {
'author.greeting.info.get': {
'httpMethod': 'GET',
'path': 'authors/{aid}/greetings/{gid}/infos/{iid}',
'rosyMethod': 'InfoService.get'
}
}
})
request = test_utils.build_request(
'/_ah/api/rpc',
'{"method": "author.greeting.info.get", "apiVersion": "X"}')
self.assert_dispatch_to_spi(request, config,
'/_ah/spi/InfoService.get',
{})
def test_handle_spi_response_json_rpc(self):
"""Verify headers transformed, JsonRpc response transformed, written."""
orig_request = test_utils.build_request(
'/_ah/api/rpc', '{"method": "foo.bar", "apiVersion": "X"}')
self.assertTrue(orig_request.is_rpc())
orig_request.request_id = 'Z'
spi_request = orig_request.copy()
spi_response = dispatcher.ResponseTuple('200 OK', [('a', 'b')],
'{"some": "response"}')
response = self.server.handle_spi_response(orig_request, spi_request,
spi_response, {},
self.start_response)
response = ''.join(response) # Merge response iterator into single body.
self.assertEqual(self.response_status, '200 OK')
self.assertIn(('a', 'b'), self.response_headers)
self.assertEqual({'id': 'Z', 'result': {'some': 'response'}},
json.loads(response))
def test_handle_spi_response_batch_json_rpc(self):
"""Verify that batch requests have an appropriate batch response."""
orig_request = test_utils.build_request(
'/_ah/api/rpc', '[{"method": "foo.bar", "apiVersion": "X"}]')
self.assertTrue(orig_request.is_batch())
self.assertTrue(orig_request.is_rpc())
orig_request.request_id = 'Z'
spi_request = orig_request.copy()
spi_response = dispatcher.ResponseTuple('200 OK', [('a', 'b')],
'{"some": "response"}')
response = self.server.handle_spi_response(orig_request, spi_request,
spi_response, {},
self.start_response)
response = ''.join(response) # Merge response iterator into single body.
self.assertEqual(self.response_status, '200 OK')
self.assertIn(('a', 'b'), self.response_headers)
self.assertEqual([{'id': 'Z', 'result': {'some': 'response'}}],
json.loads(response))
def test_handle_spi_response_rest(self):
orig_request = test_utils.build_request('/_ah/api/test', '{}')
spi_request = orig_request.copy()
body = json.dumps({'some': 'response'}, indent=1)
spi_response = dispatcher.ResponseTuple('200 OK', [('a', 'b')], body)
response = self.server.handle_spi_response(orig_request, spi_request,
spi_response, {},
self.start_response)
self.assert_http_match(response, '200 OK',
[('a', 'b'),
('Content-Length', '%d' % len(body))],
body)
def test_transform_rest_response(self):
"""Verify the response is reformatted correctly."""
orig_response = '{"sample": "test", "value1": {"value2": 2}}'
expected_response = ('{\n'
' "sample": "test", \n'
' "value1": {\n'
' "value2": 2\n'
' }\n'
'}')
self.assertEqual(expected_response,
self.server.transform_rest_response(orig_response))
def test_transform_json_rpc_response_batch(self):
"""Verify request_id inserted into the body, and body into body.result."""
orig_request = test_utils.build_request(
'/_ah/api/rpc', '[{"params": {"sample": "body"}, "id": "42"}]')
request = orig_request.copy()
request.request_id = '42'
orig_response = '{"sample": "body"}'
response = self.server.transform_jsonrpc_response(request, orig_response)
self.assertEqual([{'result': {'sample': 'body'}, 'id': '42'}],
json.loads(response))
def test_lookup_rpc_method_no_body(self):
orig_request = test_utils.build_request('/_ah/api/rpc', '')
self.assertEqual(None, self.server.lookup_rpc_method(orig_request))
def test_lookup_rpc_method(self):
self.mox.StubOutWithMock(self.server.config_manager, 'lookup_rpc_method')
self.server.config_manager.lookup_rpc_method('foo', 'v1').AndReturn('bar')
self.mox.ReplayAll()
orig_request = test_utils.build_request(
'/_ah/api/rpc', '{"method": "foo", "apiVersion": "v1"}')
self.assertEqual('bar', self.server.lookup_rpc_method(orig_request))
self.mox.VerifyAll()
def test_verify_response(self):
response = dispatcher.ResponseTuple('200', [('Content-Type', 'a')], '')
# Expected response
self.assertEqual(True, self.server.verify_response(response, 200, 'a'))
# Any content type accepted
self.assertEqual(True, self.server.verify_response(response, 200, None))
# Status code mismatch
self.assertEqual(False, self.server.verify_response(response, 400, 'a'))
# Content type mismatch
self.assertEqual(False, self.server.verify_response(response, 200, 'b'))
response = dispatcher.ResponseTuple('200', [('Content-Length', '10')], '')
# Any content type accepted
self.assertEqual(True, self.server.verify_response(response, 200, None))
# Specified content type not matched
self.assertEqual(False, self.server.verify_response(response, 200, 'a'))
def test_check_empty_response(self):
"""Test that check_empty_response returns 204 for an empty response."""
orig_request = test_utils.build_request('/_ah/api/test', '{}')
method_config = {'response': {'body': 'empty'}}
empty_response = self.server.check_empty_response(orig_request,
method_config,
self.start_response)
self.assert_http_match(empty_response, 204, [('Content-Length', '0')], '')
def test_check_non_empty_response(self):
"""Test that check_empty_response returns None for a non-empty response."""
orig_request = test_utils.build_request('/_ah/api/test', '{}')
method_config = {'response': {'body': 'autoTemplate(backendResponse)'}}
empty_response = self.server.check_empty_response(orig_request,
method_config,
self.start_response)
self.assertIsNone(empty_response)
self.assertIsNone(self.response_status)
self.assertIsNone(self.response_headers)
self.assertIsNone(self.response_exc_info)
class TransformRequestTests(unittest.TestCase):
"""Tests that only hit the request transformation functions."""
def setUp(self):
"""Set up a dev Endpoints server."""
super(TransformRequestTests, self).setUp()
self.mox = mox.Mox()
self.config_manager = api_config_manager.ApiConfigManager()
self.mock_dispatcher = self.mox.CreateMock(dispatcher.Dispatcher)
self.server = endpoints_server.EndpointsDispatcher(self.mock_dispatcher,
self.config_manager)
def tearDown(self):
self.mox.UnsetStubs()
def test_transform_request(self):
"""Verify path is method name after a request is transformed."""
request = test_utils.build_request('/_ah/api/test/{gid}',
'{"sample": "body"}')
method_config = {'rosyMethod': 'GuestbookApi.greetings_get'}
new_request = self.server.transform_request(request, {'gid': 'X'},
method_config)
self.assertEqual({'sample': 'body', 'gid': 'X'},
json.loads(new_request.body))
self.assertEqual('GuestbookApi.greetings_get', new_request.path)
def test_transform_json_rpc_request(self):
"""Verify request_id is extracted and body is scoped to body.params."""
orig_request = test_utils.build_request(
'/_ah/api/rpc', '{"params": {"sample": "body"}, "id": "42"}')
new_request = self.server.transform_jsonrpc_request(orig_request)
self.assertEqual({'sample': 'body'},
json.loads(new_request.body))
self.assertEqual('42', new_request.request_id)
def _try_transform_rest_request(self, path_parameters, query_parameters,
body_json, expected, method_params=None):
"""Takes body, query and path values from a rest request for testing.
Args:
path_parameters: A dict containing the parameters parsed from the path.
For example if the request came through /a/b for the template /a/{x}
then we'd have {'x': 'b'}.
query_parameters: A dict containing the parameters parsed from the query
string.
body_json: A dict with the JSON object from the request body.
expected: A dict with the expected JSON body after being transformed.
method_params: Optional dictionary specifying the parameter configuration
associated with the method.
"""
method_params = method_params or {}
test_request = test_utils.build_request('/_ah/api/test')
test_request.body_json = body_json
test_request.body = json.dumps(body_json)
test_request.parameters = query_parameters
transformed_request = self.server.transform_rest_request(test_request,
path_parameters,
method_params)
self.assertEqual(expected, transformed_request.body_json)
self.assertEqual(transformed_request.body_json,
json.loads(transformed_request.body))
# Path only
def test_transform_rest_request_path_only(self):
path_parameters = {'gid': 'X'}
query_parameters = {}
body_object = {}
expected = {'gid': 'X'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_path_only_message_field(self):
path_parameters = {'gid.val': 'X'}
query_parameters = {}
body_object = {}
expected = {'gid': {'val': 'X'}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_path_only_enum(self):
query_parameters = {}
body_object = {}
enum_descriptor = {'X': {'backendValue': 'X'}}
method_params = {'gid': {'enum': enum_descriptor}}
# Good enum
path_parameters = {'gid': 'X'}
expected = {'gid': 'X'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
# Bad enum
path_parameters = {'gid': 'Y'}
expected = {'gid': 'Y'}
try:
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
self.fail('Bad enum should have caused failure.')
except errors.EnumRejectionError as error:
self.assertEqual(error.parameter_name, 'gid')
# Query only
def test_transform_rest_request_query_only(self):
path_parameters = {}
query_parameters = {'foo': ['bar']}
body_object = {}
expected = {'foo': 'bar'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_query_only_message_field(self):
path_parameters = {}
query_parameters = {'gid.val': ['X']}
body_object = {}
expected = {'gid': {'val': 'X'}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_query_only_multiple_values_not_repeated(self):
path_parameters = {}
query_parameters = {'foo': ['bar', 'baz']}
body_object = {}
expected = {'foo': 'bar'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_query_only_multiple_values_repeated(self):
path_parameters = {}
query_parameters = {'foo': ['bar', 'baz']}
body_object = {}
method_params = {'foo': {'repeated': True}}
expected = {'foo': ['bar', 'baz']}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
def test_transform_rest_request_query_only_enum(self):
path_parameters = {}
body_object = {}
enum_descriptor = {'X': {'backendValue': 'X'}}
method_params = {'gid': {'enum': enum_descriptor}}
# Good enum
query_parameters = {'gid': ['X']}
expected = {'gid': 'X'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
# Bad enum
query_parameters = {'gid': ['Y']}
expected = {'gid': 'Y'}
try:
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
self.fail('Bad enum should have caused failure.')
except errors.EnumRejectionError as error:
self.assertEqual(error.parameter_name, 'gid')
def test_transform_rest_request_query_only_repeated_enum(self):
path_parameters = {}
body_object = {}
enum_descriptor = {'X': {'backendValue': 'X'}, 'Y': {'backendValue': 'Y'}}
method_params = {'gid': {'enum': enum_descriptor, 'repeated': True}}
# Good enum
query_parameters = {'gid': ['X', 'Y']}
expected = {'gid': ['X', 'Y']}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
# Bad enum
query_parameters = {'gid': ['X', 'Y', 'Z']}
expected = {'gid': ['X', 'Y', 'Z']}
try:
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
self.fail('Bad enum should have caused failure.')
except errors.EnumRejectionError as error:
self.assertEqual(error.parameter_name, 'gid[2]')
# Body only
def test_transform_rest_request_body_only(self):
path_parameters = {}
query_parameters = {}
body_object = {'sample': 'body'}
expected = {'sample': 'body'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_body_only_any_old_value(self):
path_parameters = {}
query_parameters = {}
body_object = {'sample': {'body': ['can', 'be', 'anything']}}
expected = {'sample': {'body': ['can', 'be', 'anything']}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_body_only_message_field(self):
path_parameters = {}
query_parameters = {}
body_object = {'gid': {'val': 'X'}}
expected = {'gid': {'val': 'X'}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_body_only_enum(self):
path_parameters = {}
query_parameters = {}
enum_descriptor = {'X': {'backendValue': 'X'}}
method_params = {'gid': {'enum': enum_descriptor}}
# Good enum
body_object = {'gid': 'X'}
expected = {'gid': 'X'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
# Bad enum
body_object = {'gid': 'Y'}
expected = {'gid': 'Y'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
# Path and query only
def test_transform_rest_request_path_query_no_collision(self):
path_parameters = {'a': 'b'}
query_parameters = {'c': ['d']}
body_object = {}
expected = {'a': 'b', 'c': 'd'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_path_query_collision(self):
path_parameters = {'a': 'b'}
query_parameters = {'a': ['d']}
body_object = {}
expected = {'a': 'd'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_path_query_collision_in_repeated_param(self):
path_parameters = {'a': 'b'}
query_parameters = {'a': ['d', 'c']}
body_object = {}
expected = {'a': ['d', 'c', 'b']}
method_params = {'a': {'repeated': True}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
# Path and body only
def test_transform_rest_request_path_body_no_collision(self):
path_parameters = {'a': 'b'}
query_parameters = {}
body_object = {'c': 'd'}
expected = {'a': 'b', 'c': 'd'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_path_body_collision(self):
path_parameters = {'a': 'b'}
query_parameters = {}
body_object = {'a': 'd'}
expected = {'a': 'd'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_path_body_collision_in_repeated_param(self):
path_parameters = {'a': 'b'}
query_parameters = {}
body_object = {'a': ['d']}
expected = {'a': ['d']}
method_params = {'a': {'repeated': True}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
def test_transform_rest_request_path_body_message_field_cooperative(self):
path_parameters = {'gid.val1': 'X'}
query_parameters = {}
body_object = {'gid': {'val2': 'Y'}}
expected = {'gid': {'val1': 'X', 'val2': 'Y'}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_path_body_message_field_collision(self):
path_parameters = {'gid.val': 'X'}
query_parameters = {}
body_object = {'gid': {'val': 'Y'}}
expected = {'gid': {'val': 'Y'}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
# Query and body only
def test_transform_rest_request_query_body_no_collision(self):
path_parameters = {}
query_parameters = {'a': ['b']}
body_object = {'c': 'd'}
expected = {'a': 'b', 'c': 'd'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_query_body_collision(self):
path_parameters = {}
query_parameters = {'a': ['b']}
body_object = {'a': 'd'}
expected = {'a': 'd'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_query_body_collision_in_repeated_param(self):
path_parameters = {}
query_parameters = {'a': ['b']}
body_object = {'a': ['d']}
expected = {'a': ['d']}
method_params = {'a': {'repeated': True}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
def test_transform_rest_request_query_body_message_field_cooperative(self):
path_parameters = {}
query_parameters = {'gid.val1': ['X']}
body_object = {'gid': {'val2': 'Y'}}
expected = {'gid': {'val1': 'X', 'val2': 'Y'}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_query_body_message_field_collision(self):
path_parameters = {}
query_parameters = {'gid.val': ['X']}
body_object = {'gid': {'val': 'Y'}}
expected = {'gid': {'val': 'Y'}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
# Path, body and query
def test_transform_rest_request_path_query_body_no_collision(self):
path_parameters = {'a': 'b'}
query_parameters = {'c': ['d']}
body_object = {'e': 'f'}
expected = {'a': 'b', 'c': 'd', 'e': 'f'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_path_query_body_collision(self):
path_parameters = {'a': 'b'}
query_parameters = {'a': ['d']}
body_object = {'a': 'f'}
expected = {'a': 'f'}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected)
def test_transform_rest_request_unknown_parameters(self):
path_parameters = {'a': 'b'}
query_parameters = {'c': ['d']}
body_object = {'e': 'f'}
expected = {'a': 'b', 'c': 'd', 'e': 'f'}
method_params = {'X': {}, 'Y': {}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
# Other tests.
def test_type_conversions(self):
"""Verify that type conversion matches prod."""
path_parameters = {'int32_val': '1', 'uint32_val': '2',
'int64_val': '3', 'uint64_val': '4',
'true_bool_val': 'true', 'false_bool_val': 'FALSE'}
query_parameters = {'float_val': ['5.25'], 'double_val': ['6.5']}
body_object = {'int_body_val': '7'}
expected = {'int32_val': 1,
'uint32_val': 2,
'int64_val': '3',
'uint64_val': '4',
'true_bool_val': True,
'false_bool_val': False,
'float_val': 5.25,
'double_val': 6.5,
'int_body_val': '7'}
method_params = {'int32_val': {'type': 'int32'},
'uint32_val': {'type': 'uint32'},
'int64_val': {'type': 'int64'},
'uint64_val': {'type': 'uint64'},
'true_bool_val': {'type': 'boolean'},
'false_bool_val': {'type': 'boolean'},
'float_val': {'type': 'float'},
'double_val': {'type': 'double'},
'int_body_val': {'type': 'int32'}}
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected, method_params)
def test_invalid_conversions(self):
"""Verify that invalid parameter values for basic types raise errors."""
for type_name in ('int32', 'uint32', 'boolean', 'float', 'double'):
param_name = '%s_val' % type_name
path_parameters = {param_name: 'invalid'}
query_parameters = {}
body_object = {}
expected = {}
method_params = {param_name: {'type': type_name}}
try:
self._try_transform_rest_request(path_parameters, query_parameters,
body_object, expected,
method_params=method_params)
self.fail('Bad %s value should have caused failure.' % type_name)
except errors.BasicTypeParameterError as error:
self.assertEqual(error.parameter_name, param_name)
if __name__ == '__main__':
unittest.main()