google/appengine/tools/devappserver2/endpoints/testdata/test_service.py

#!/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', name='test', path='test',
                    scopes=[])
  def test(self, unused_request):
    return TestResponse(text='Test response')

  @endpoints.method(message_types.VoidMessage, TestResponse,
                    http_method='GET', name='empty_test', path='empty_test',
                    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',
                    path='echo_datetime_message',
                    scopes=[])
  def echo_datetime_message(self, request):
    return request

  @endpoints.method(TestDateTime, TestDateTime,
                    http_method='POST', name='echodtfield',
                    path='echo_datetime_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',
                    path='increment_integers', 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,
                    path='echo_bytes', 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])