38

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?

6 Answers 6

30

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.

12
  • 1
    I tried the second option, and I am getting a AssertionError: 'app_identity_service'. do you know what could be the problem?
    – Arash
    Mar 13, 2014 at 21:31
  • Thanks Ezequiel, I used the first method and it works. Just wanna ask if you know how to handle an exception from the API. How to make testbed.post_json recognize sth like a "endpoints.UnauthorizedException('Access denied')" ?
    – Nathan Do
    Apr 30, 2014 at 9:42
  • I don't know off the top of my head. Supposedly that would result in a 403 being returned by the SPI, so you would be able to check that's what happened through self.assertEquals(resp.status_code, 403) or something like that. I haven't actually tried this but it may work. May 4, 2014 at 8:45
  • 1
    I tried to test endpoints exception too with your first solution, but the tests fail everytime for on error inside webtest (AssertionError: Content-Length is different from actual app_iter length (512!=67)). The test doesn't reach the assertEqual, it fails on the testapp.post_json. I also tried to use the solution given by @schibum, but for now I can't find a way to build a ResourceContainer parameter.
    – Malexandre
    May 5, 2014 at 14:41
  • i receive this error: StubNotSupportedError: Could not initialize images API; you are likely missing the Python "PIL" module. on this line tb.init_all_stubs() any idea? i tried to install PIL, but it does not install/exists anymore. and even Pillow does not install.
    – EsseTi
    May 13, 2014 at 19:54
6

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()
2
  • I tried the code above but kept getting this error: TypeError: services contains types that don't match _ServiceClass: [<class 'protorpc.remote._ServiceClass'>]
    – jwesonga
    Sep 26, 2014 at 13:19
  • I'm finding trouble with importing various appengine paths to python paths. I'm trying to do the tests under a virtualenv. Please tell me how you accomplished that.
    – Afzal S.H.
    Apr 6, 2017 at 9:53
2

I tried everything I could think of to allow these to be tested in the normal way. I tried hitting the /_ah/spi methods directly as well as even trying to create a new protorpc app using service_mappings to no avail. I'm not a Googler on the endpoints team so maybe they have something clever to allow this to work but it doesn't appear that simply using webtest will work (unless I missed something obvious).

In the meantime you can write a test script that starts the app engine test server with an isolated environment and just issue http requests to it.

Example to run the server with an isolated environment (bash but you can easily run this from python):

DATA_PATH=/tmp/appengine_data

if [ ! -d "$DATA_PATH" ]; then
    mkdir -p $DATA_PATH
fi

dev_appserver.py --storage_path=$DATA_PATH/storage --blobstore_path=$DATA_PATH/blobstore --datastore_path=$DATA_PATH/datastore --search_indexes_path=$DATA_PATH/searchindexes --show_mail_body=yes --clear_search_indexes --clear_datastore .

You can then just use requests to test ala curl:

requests.get('http://localhost:8080/_ah/...')
1

If you don't want to test the full HTTP stack as described by Ezequiel Muns, you can also just mock out endpoints.method and test your API definition directly:

def null_decorator(*args, **kwargs):
    def decorator(method):
        def wrapper(*args, **kwargs):
            return method(*args, **kwargs)
        return wrapper
    return decorator

from google.appengine.api.users import User
import endpoints
endpoints.method = null_decorator
# decorator needs to be mocked out before you load you endpoint api definitions
from mymodule import api


class FooTest(unittest.TestCase):
    def setUp(self):
        self.api = api.FooService()

    def test_bar(self):
        # pass protorpc messages directly
        self.api.foo_bar(api.MyRequestMessage(some='field'))
1
  • How do you tests the endpoints methods that required a ResourceContainer ? I can't find a way to instantiate an object. With some logging, I found out that the request received when the server is launched is a protorpc.messages.CombinedContainer object, but I can't instantiate it either. I'm quite new at Python for your information.
    – Malexandre
    May 5, 2014 at 14:53
1

My solution uses one dev_appserver instance for the entire test module, which is faster than restarting the dev_appserver for each test method.

By using Google's Python API client library, I also get the simplest and at the same time most powerful way of interacting with my API.

import unittest
import sys
import os

from apiclient.discovery import build
import dev_appserver


sys.path[1:1] = dev_appserver.EXTRA_PATHS

from google.appengine.tools.devappserver2 import devappserver2
from google.appengine.tools.devappserver2 import python_runtime

server = None


def setUpModule():
    # starting a dev_appserver instance for testing
    path_to_app_yaml = os.path.normpath('path_to_app_yaml')
    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(['--port', '8080',
                                           '--datastore_path', ':memory:',
                                           '--logs_path', ':memory:',
                                           '--skip_sdk_update_check',
                                           '--',
                                           ] + app_configs)
    global server
    server = devappserver2.DevelopmentServer()
    server.start(options)


def tearDownModule():
    # shutting down dev_appserver instance after testing
    server.stop()


class MyTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        # build a service object for interacting with the api
        # dev_appserver must be running and listening on port 8080
        api_root = 'http://127.0.0.1:8080/_ah/api'
        api = 'my_api'
        version = 'v0.1'
        discovery_url = '%s/discovery/v1/apis/%s/%s/rest' % (api_root, api,                     
                                                             version)
        cls.service = build(api, version, discoveryServiceUrl=discovery_url)

    def setUp(self):
        # create a parent entity and store its key for each test run
        body = {'name': 'test  parent'}
        response = self.service.parent().post(body=body).execute()   
        self.parent_key = response['parent_key']

    def test_post(self):
        # test my post method 
        # the tested method also requires a path argument "parent_key" 
        # .../_ah/api/my_api/sub_api/post/{parent_key}
        body = {'SomeProjectEntity': {'SomeId': 'abcdefgh'}}
        parent_key = self.parent_key
        req = self.service.sub_api().post(body=body,parent_key=parent_key)
        response = req.execute()
        etc..
1
  • I think dev_appserver2.py no longer contains a python_runtime. Dec 1, 2017 at 8:37
1

After digging through the sources, I believe things have changed in endpoints since Ezequiel Muns's (excellent) answer in 2014. For method 1 you now need to request from /_ah/api/* directly and use the correct HTTP method instead of using the /_ah/spi/* transformation. This makes the test file look like this:

from google.appengine.ext import testbed
import webtest
# ...
def setUp(self):
    tb = testbed.Testbed()
    # Setting current_version_id doesn't seem necessary anymore
    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 is no longer required
    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/api/test/v1/insert', msg)

    self.assertEqual(resp.json, {'expected': 'json response msg as dict'})

For searching's sake, the symptom of using the old method is endpoints raising a ValueError with Invalid request path: /_ah/spi/whatever. Hope that saves someone some time!

Your Answer

Reminder: Answers generated by Artificial Intelligence tools are not allowed on Stack Overflow. Learn more

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.