如何单元测试Google Cloud Endpoints [英] How to unit test Google Cloud Endpoints

问题描述

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

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

  application = endpoints.api_server（[TestEndpoint]，restricted = False）
  

然后，我使用WebTest这种方式：

pre code客户端= webtest.TestApp（应用程序）
client.post（ '/ _ah / api / test / v1 / test'，params）

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

解决方案

经过大量实验并查看SDK代码后，我想出了两个在python中测试端点的方法：

1。使用webtest + testbed测试SPI端

您使用webtest正确运行，但只需确保正确转换您对SPI端点的请求。 / p>

Cloud Endpoints API前端和 dev_appserver EndpointsDispatcher c>将调用转换为 / _ ah / spi / * 对应的后端调用 / _ ah / api / * 。转换似乎是：
$ b

• 所有调用都是 application / json HTTP 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'）＃需要，因为端点需要a。在这个值中
 tb.activate（）
 tb.init_all_stubs（）
 self.testbed = tb 
 $ b $ def tearDown（self）：
 self。 testbed.deactivate（）
 
 def test_endpoint_insert（self）：
 app = endpoints.api_server（[TestEndpoint]，restricted = False）
 testapp = webtest.TestApp（app）
 msg = {...}＃代表插入期望的消息对象的字典
＃要通过webtest序列化为JSON 
 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从google.appengine.tools.devappserver2导入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_sd 
' - '，
] + APP_CONFIGS）
 server = devappserver2.DevelopmentServer（）
 server.start（选项）
 self.server =服务器
 
 def tearDown（self）：
 self.server.stop（）
  

现在，您需要向localhost：8123发出实际的HTTP请求，以针对API运行测试，但可以再次与GAE API交互以设置固件等。这显然很慢为每次测试运行创建并销毁新的开发服务器。

此时，我使用 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，{...期望响应为字典...}）
  

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

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?

解决方案

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.

这篇关于如何单元测试Google Cloud Endpoints的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！