如何对Google云端点进行单元测试

Question

我需要一些帮助为Google云端点设置unittests。使用WebTest all requests应答AppError:Bad response: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端点团队有什么建议？

Answer 1

可以简化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()

Answer 2

经过大量的实验和对SDK代码的研究，我想出了两种在python中测试端点的方法：

一。使用webtest+testbed测试SPI端

您使用webtest的方法是正确的，但是只需要确保您正确地转换了对SPI端点的请求。

云端点API前端和dev_appserver中的EndpointsDispatcher将对/_ah/api/*的调用转换为对/_ah/spi/*的相应“后端”调用。转变似乎是：

• 所有调用都是application/jsonHTTP post（即使REST端点是其他的）。
• 请求参数（path、query和JSON body）都合并到一个JSON body消息中。
• “backend”端点使用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服务中设置适当的fixture，这样您就可以更充分地断言调用的预期副作用。

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

您可以在同一python环境中使用如下方法启动dev服务器：

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:8123发出实际的HTTP请求，以便对API运行测试，但仍然可以与GAE API交互以设置fixture等。这显然很慢，因为您正在为每次测试运行创建和销毁一个新的dev服务器。

此时，我使用Google API Python client来使用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测试的改进，因为它让您可以直接访问gaeapi来轻松地设置fixture和检查内部状态。我怀疑有一种更好的方法可以绕过HTTP进行集成测试，方法是将实现端点分派机制的dev服务器中的最小组件缝合在一起，但这需要比现在更多的研究时间。

Answer 3

我想尽一切办法让它们以正常的方式测试。我尝试直接使用/uah/spi方法，甚至尝试使用服务映射创建一个新的protorpc应用程序，但都没有成功。我不是endpoints团队中的Googler，所以也许他们有一些聪明的东西可以让它工作，但似乎仅仅使用webtest是行不通的（除非我漏掉了一些明显的东西）。

同时，您可以编写一个测试脚本，用一个独立的环境启动app engine测试服务器，并向其发出http请求。

使用独立环境运行服务器的示例（bash，但您可以从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 .

然后您可以使用请求来测试ala curl：

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