blob: 0a329df78c0fcb59256182e66b242d265ed6a792 [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.
#
"""Test service for regression testing of Cloud Endpoints support."""
import logging
import endpoints
from protorpc import message_types
from protorpc import messages
from protorpc import remote
class TestRequest(messages.Message):
"""Simple ProtoRPC request, for testing."""
name = messages.StringField(1)
number = messages.IntegerField(2)
class TestResponse(messages.Message):
"""Simple ProtoRPC response with a text field."""
text = messages.StringField(1)
class TestDateTime(messages.Message):
"""Simple ProtoRPC request/response with a datetime."""
datetime_value = message_types.DateTimeField(1)
class TestIntegers(messages.Message):
"""Simple ProtoRPC request/response with a few integer types."""
var_int32 = messages.IntegerField(1, variant=messages.Variant.INT32)
var_int64 = messages.IntegerField(2, variant=messages.Variant.INT64)
var_repeated_int64 = messages.IntegerField(3, variant=messages.Variant.INT64,
repeated=True)
var_sint64 = messages.IntegerField(4, variant=messages.Variant.SINT64)
var_uint64 = messages.IntegerField(5, variant=messages.Variant.UINT64)
class TestBytes(messages.Message):
"""Simple ProtoRPC request/response with a bytes field."""
bytes_value = messages.BytesField(1)
my_api = endpoints.api(name='test_service', version='v1')
@my_api.api_class()
class TestService(remote.Service):
"""ProtoRPC test class for Cloud Endpoints."""
@endpoints.method(message_types.VoidMessage, TestResponse,
http_method='GET', scopes=[])
def test(self, unused_request):
return TestResponse(text='Test response')
@endpoints.method(message_types.VoidMessage, TestResponse,
http_method='GET', scopes=[])
def empty_test(self, unused_request):
return TestResponse()
@endpoints.method(TestRequest, TestResponse,
http_method='POST', name='t2name', path='t2path',
scopes=[])
def getenviron(self, request):
return TestResponse(text='%s %d' % (request.name, request.number))
@endpoints.method(message_types.DateTimeMessage,
message_types.DateTimeMessage,
http_method='POST', name='echodtmsg', scopes=[])
def echo_datetime_message(self, request):
return request
@endpoints.method(TestDateTime, TestDateTime,
http_method='POST', name='echodtfield',
path='echo_dt_field', scopes=[])
def echo_datetime_field(self, request):
# Make sure we can access the fields of the datetime object.
logging.info('Year %d, Month %d', request.datetime_value.year,
request.datetime_value.month)
return request
@endpoints.method(TestIntegers, TestIntegers, http_method='POST', scopes=[])
def increment_integers(self, request):
response = TestIntegers(
var_int32=request.var_int32 + 1,
var_int64=request.var_int64 + 1,
var_repeated_int64=[val + 1 for val in request.var_repeated_int64],
var_sint64=request.var_sint64 + 1,
var_uint64=request.var_uint64 + 1)
return response
@endpoints.method(TestBytes, TestBytes, scopes=[])
def echo_bytes(self, request):
logging.info('Found bytes: %s', request.bytes_value)
return request
@endpoints.method(message_types.VoidMessage, message_types.VoidMessage,
path='empty_response', http_method='GET', scopes=[])
def empty_response(self, unused_request):
return message_types.VoidMessage()
@my_api.api_class(resource_name='extraname', path='extrapath')
class ExtraMethods(remote.Service):
"""Some extra test methods in the test API."""
@endpoints.method(message_types.VoidMessage, TestResponse,
http_method='GET', name='test', path='test',
scopes=[])
def test(self, unused_request):
return TestResponse(text='Extra test response')
@endpoints.api(name='second_service', version='v1')
class SecondService(remote.Service):
"""Second test class for Cloud Endpoints."""
@endpoints.method(message_types.VoidMessage, TestResponse,
http_method='GET', name='test_name', path='test',
scopes=[])
def second_test(self, unused_request):
"""Test a second API, same version, same path. Shouldn't collide."""
return TestResponse(text='Second response')
application = endpoints.api_server([TestService, ExtraMethods, SecondService])