如何unit testingGoogle Cloud Endpoints

我需要一些帮助为Google Cloud Endpoint设置unit testing。 使用WebTest的所有请求与AppError答案：错误的响应：404未find。 我不确定terminal是否与WebTest兼容。

这是如何生成应用程序：

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

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

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

curltesting工作正常。

我应该为不同的端点编写testing吗？ GAE端点小组的build议是什么？

Answer 1

经过大量的实验并查看SDK代码，我想出了两种方法来testingpython中的端点：

1.使用webtest + testbedtestingSPI端

您正在使用webtest进入正确的轨道，但只需确保您正确转换您对SPI端点的请求。

Cloud Endpoints API前端和dev_appserver的EndpointsDispatcher /_ah/api/*调用转换为对/_ah/spi/*相应“后端”调用。 转型似乎是：

• 所有调用都是application/json HTTP POST（即使REST端点是别的）。
• 请求参数（path，查询和JSON主体）全部合并成单个JSON主体消息。
• “后端”端点使用URL中实际的python类和方法名称，例如POST /_ah/spi/TestEndpoint.insert_message将在您的代码中调用TestEndpoint.insert_message() 。
• JSON响应仅在被返回到原始客户端之前被重新格式化。

这意味着您可以使用以下设置testing端点：

 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.启动开发服务器进行完整集成testing

您可以使用类似下面的内容在相同的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() 

现在，您需要发出实际的 HTTP请求到localhost：8123来运行APItesting，但是可以再次与GAE API交互来设置灯具等等。这显然很慢，因为您正在为每个创build和销毁新的开发服务器testing运行。

此时，我使用Google API Python客户端来使用API​​，而不是自己构buildHTTP请求：

 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进行testing相比，这是一个改进，因为它使您可以直接访问GAE API，从而轻松设置灯具并检查内部状态。 我怀疑还有一种更好的方法来做绕过HTTP的集成testing，把实现端点调度机制的dev服务器中的最小组件拼接在一起，但是这需要比我现在更多的研究时间。

Answer 2

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

为以下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 3

我尝试了所有我能想到的方法，以便以正常的方式进行testing。 我尝试直接打击/ _ah / spi方法，甚至尝试使用service_mappings创build一个新的protorpc应用程序无济于事。 我不是端点团队的Google员工，所以也许他们有一些聪明的工作，但是似乎并没有简单地使用webtest（除非我错过了一些明显的东西）。

与此同时，您可以编写一个testing脚本来启动具有隔离环境的应用程序引擎testing服务器，并向其发出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 . 

然后你可以使用请求来testingala curl：

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

Answer 4

如果你不想像Ezequiel Muns所描述的那样testing完整的HTTP栈，你也可以模拟endpoints.method并直接testing你的API定义：

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

Answer 5

我的解决scheme为整个testing模块使用一个dev_appserver实例，比为每个testing方法重新启动dev_appserver要快。

通过使用Google的Python API客户端库，我也可以获得最简单，最有效的与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..