如何对Google Cloud Endpoints进行单元测试

Question

我需要一些帮助来设置Google Cloud Endpoints的单元测试.使用WebTest所有请求都回答AppError:错误响应:404 Not Found.我不确定端点是否与WebTest兼容.

这是应用程序的生成方式:

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

然后我用这种方式使用WebTest:

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

使用curl进行测试工作正常.

我应该为端点编写不同的测试吗？GAE Endpoints团队的建议是什么？

Answer 1

经过大量实验并查看SDK代码后,我提出了两种在python中测试端点的方法:

1.使用webtest + testbed测试SPI端

您使用webtest处于正确的轨道上,但只需要确保正确转换您对SPI端点的请求.

Cloud Endpoints API前端和EndpointsDispatcherin dev_appserver将调用转换/_ah/api/*为相应的"后端"调用/_ah/spi/*.转型似乎是:

• 所有调用都是application/jsonHTTP POST(即使REST端点是其他东西).
• 请求参数(路径,查询和JSON主体)全部合并为一个JSON正文消息.
• "后端"端点使用URL中的实际python类和方法名称,例如POST /_ah/spi/TestEndpoint.insert_message将调用TestEndpoint.insert_message()代码.
• JSON响应仅在返回到原始客户端之前重新格式化.

这意味着您可以使用以下设置测试端点:

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'})

这里的事情是您可以在调用端点之前在数据存储区或其他GAE服务中轻松设置适当的灯具,从而可以更充分地断言呼叫的预期副作用.

2.启动开发服务器以进行完全集成测试

您可以使用以下内容在同一个python环境中启动开发服务器:

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()

现在您需要向localhost 发出实际的 HTTP请求:8123以针对API运行测试,但是再次可以与GAE API交互以设置fixture等.当您为每个创建和销毁新的dev服务器时,这显然很慢测试运行.

此时,我使用Google API Python客户端来使用API​​,而不是自己构建HTTP请求:

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 ... })

这是对使用CURL进行测试的改进,因为它使您可以直接访问GAE API以轻松设置灯具并检查内部状态.我怀疑有一种更好的方法来进行集成测试,通过将实现端点调度机制的开发服务器中的最小组件拼接在一起来绕过HTTP,但这需要比我现在更多的研究时间.

Answer 2

webtest可以简化以减少命名错误

对于以下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")

tests.py应该是这样的

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()