How to unit test Google Cloud Endpoints

Question

I'm needing some help setting up unittests for Google Cloud Endpoints. Using WebTest all requests answer with AppError: Bad response: 404 Not Found. I'm not really sure if endpoints is compatible with WebTest.

This is how the application is generated:

application = endpoints.api_server([TestEndpoint], restricted=False) 

Then I use WebTest this way:

client = webtest.TestApp(application) client.post('/_ah/api/test/v1/test', params) 

Testing with curl works fine.

Should I write tests for endpoints different? What is the suggestion from GAE Endpoints team?

Answer 1

After much experimenting and looking at the SDK code I've come up with two ways to test endpoints within python:

1. Using webtest + testbed to test the SPI side

You are on the right track with webtest, but just need to make sure you correctly transform your requests for the SPI endpoint.

The Cloud Endpoints API front-end and the EndpointsDispatcher in dev_appserver transforms calls to /_ah/api/* into corresponding "backend" calls to /_ah/spi/*. The transformation seems to be:

• All calls are application/json HTTP POSTs (even if the REST endpoint is something else).
• The request parameters (path, query and JSON body) are all merged together into a single JSON body message.
• The "backend" endpoint uses the actual python class and method names in the URL, e.g. POST /_ah/spi/TestEndpoint.insert_message will call TestEndpoint.insert_message() in your code.
• The JSON response is only reformatted before being returned to the original client.

This means you can test the endpoint with the following setup:

from google.appengine.ext import testbed import webtest # ... def setUp(self):     tb = testbed.Testbed()     tb.setup_env(current_version_id='testbed.version') #needed because endpoints expects a . in this value     tb.activate()     tb.init_all_stubs()     self.testbed = tb  def tearDown(self):     self.testbed.deactivate()  def test_endpoint_insert(self):     app = endpoints.api_server([TestEndpoint], restricted=False)     testapp = webtest.TestApp(app)     msg = {...} # a dict representing the message object expected by insert                 # To be serialised to JSON by webtest     resp = testapp.post_json('/_ah/spi/TestEndpoint.insert', msg)      self.assertEqual(resp.json, {'expected': 'json response msg as dict'}) 

The thing here is you can easily setup appropriate fixtures in the datastore or other GAE services prior to calling the endpoint, thus you can more fully assert the expected side effects of the call.

2. Starting the development server for full integration test

You can start the dev server within the same python environment using something like the following:

import sys import os import dev_appserver sys.path[1:1] = dev_appserver._DEVAPPSERVER2_PATHS  from google.appengine.tools.devappserver2 import devappserver2 from google.appengine.tools.devappserver2 import python_runtime # ... def setUp(self):     APP_CONFIGS = ['/path/to/app.yaml']      python_runtime._RUNTIME_ARGS = [         sys.executable,         os.path.join(os.path.dirname(dev_appserver.__file__),                      '_python_runtime.py')     ]     options = devappserver2.PARSER.parse_args([         '--admin_port', '0',         '--port', '8123',          '--datastore_path', ':memory:',         '--logs_path', ':memory:',         '--skip_sdk_update_check',         '--',     ] + APP_CONFIGS)     server = devappserver2.DevelopmentServer()     server.start(options)     self.server = server  def tearDown(self):     self.server.stop() 

Now you need to issue actual HTTP requests to localhost:8123 to run tests against the API, but again can interact with GAE APIs to set up fixtures, etc. This is obviously slow as you're creating and destroying a new dev server for every test run.

At this point I use the Google API Python client to consume the API instead of building the HTTP requests myself:

import apiclient.discovery # ... def test_something(self):     apiurl = 'http://%s/_ah/api/discovery/v1/apis/{api}/{apiVersion}/rest' \                     % self.server.module_to_address('default')     service = apiclient.discovery.build('testendpoint', 'v1', apiurl)      res = service.testresource().insert({... message ... }).execute()     self.assertEquals(res, { ... expected reponse as dict ... }) 

This is an improvement over testing with CURL as it gives you direct access to the GAE APIs to easily set up fixtures and inspect internal state. I suspect there is an even better way to do integration testing that bypasses HTTP by stitching together the minimal components in the dev server that implement the endpoint dispatch mechanism, but that requires more research time than I have right now.

Answer 2

webtest can be simplified to reduce naming bugs

for the following TestApi

import endpoints import protorpc import logging  class ResponseMessageClass(protorpc.messages.Message):     message = protorpc.messages.StringField(1) class RequestMessageClass(protorpc.messages.Message):     message = protorpc.messages.StringField(1)   @endpoints.api(name='testApi',version='v1',                description='Test API',                allowed_client_ids=[endpoints.API_EXPLORER_CLIENT_ID]) class TestApi(protorpc.remote.Service):      @endpoints.method(RequestMessageClass,                       ResponseMessageClass,                       name='test',                       path='test',                       http_method='POST')     def test(self, request):         logging.info(request.message)         return ResponseMessageClass(message="response message") 

the tests.py should look like this

import webtest import logging import unittest from google.appengine.ext import testbed from protorpc.remote import protojson import endpoints  from api.test_api import TestApi, RequestMessageClass, ResponseMessageClass   class AppTest(unittest.TestCase):     def setUp(self):         logging.getLogger().setLevel(logging.DEBUG)          tb = testbed.Testbed()         tb.setup_env(current_version_id='testbed.version')          tb.activate()         tb.init_all_stubs()         self.testbed = tb       def tearDown(self):         self.testbed.deactivate()       def test_endpoint_testApi(self):         application = endpoints.api_server([TestApi], restricted=False)          testapp = webtest.TestApp(application)          req = RequestMessageClass(message="request message")          response = testapp.post('/_ah/spi/' + TestApi.__name__ + '.' + TestApi.test.__name__, protojson.encode_message(req),content_type='application/json')          res = protojson.decode_message(ResponseMessageClass,response.body)          self.assertEqual(res.message, 'response message')   if __name__ == '__main__':     unittest.main()