python - for - Cómo probar Google Cloud Endpoints
python flask google cloud (6)
Necesito ayuda para configurar unitrests para Google Cloud Endpoints. Al usar WebTest todas las solicitudes responden con AppError: Mala respuesta: 404 no encontrado. No estoy seguro de si los puntos finales son compatibles con WebTest.
Así es como se genera la aplicación:
application = endpoints.api_server([TestEndpoint], restricted=False)
Luego uso WebTest de esta manera:
client = webtest.TestApp(application)
client.post(''/_ah/api/test/v1/test'', params)
La prueba con curl funciona bien.
¿Debo escribir pruebas para los puntos finales diferentes? ¿Cuál es la sugerencia del equipo GAE Endpoints?
Después de investigar las fuentes, creo que las cosas han cambiado en los puntos finales desde la excelente respuesta de Ezequiel Muns en 2014. Para el método 1, ahora debe solicitar directamente a / _ah / api / * y usar el método HTTP correcto en lugar de usar el / _ah / spi / * transformación. Esto hace que el archivo de prueba se vea así:
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''})
En aras de la búsqueda, el síntoma de utilizar el método anterior es que los puntos finales generan un ValueError con una Invalid request path: /_ah/spi/whatever . Espero que le ahorre a alguien algo de tiempo!
Después de mucho experimentar y mirar el código SDK, he encontrado dos formas de probar los puntos finales dentro de Python:
1. Usando webtest + testbed para probar el lado de SPI
Está en el camino correcto con webtest, pero solo necesita asegurarse de que transforme correctamente sus solicitudes para el punto final SPI.
El front-end de API de Cloud Endpoints y EndpointsDispatcher en dev_appserver transforma las llamadas a /_ah/api/* en las llamadas de "back-end" correspondientes a /_ah/spi/* . La transformación parece ser:
- Todas las llamadas son HTTP POST de
application/json(incluso si el punto final REST es algo más). - Los parámetros de solicitud (ruta, consulta y cuerpo JSON) se fusionan en un único mensaje cuerpo JSON.
- El punto final "back-end" usa la clase real de python y los nombres de los métodos en la URL, por ejemplo,
POST /_ah/spi/TestEndpoint.insert_messagellamará aTestEndpoint.insert_message()en su código. - La respuesta JSON solo se reformatea antes de ser devuelta al cliente original.
Esto significa que puede probar el punto final con la siguiente configuración:
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''})
Aquí se puede configurar fácilmente los dispositivos apropiados en el almacén de datos u otros servicios GAE antes de llamar al punto final, de esta manera puede afirmar más plenamente los efectos secundarios esperados de la llamada.
2. Iniciando el servidor de desarrollo para la prueba de integración completa
Puede iniciar el servidor de desarrollo dentro del mismo entorno de python usando algo como lo siguiente:
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()
Ahora necesita emitir solicitudes HTTP reales a localhost: 8123 para ejecutar pruebas contra la API, pero de nuevo puede interactuar con las API de GAE para configurar dispositivos, etc. Esto es obviamente lento ya que está creando y destruyendo un nuevo servidor de desarrollo para cada prueba de funcionamiento.
En este punto, uso el API Python de API de Google para consumir la API en lugar de crear las solicitudes HTTP por mi cuenta:
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 ... })
Esta es una mejora sobre las pruebas con CURL, ya que le da acceso directo a las API de GAE para configurar fácilmente los dispositivos e inspeccionar el estado interno. Sospecho que hay una forma aún mejor de hacer pruebas de integración que elude HTTP al unir los componentes mínimos en el servidor de desarrollo que implementan el mecanismo de despacho del punto final, pero eso requiere más tiempo de investigación que el que tengo ahora.
Mi solución utiliza una instancia de dev_appserver para todo el módulo de prueba, que es más rápido que reiniciar dev_appserver para cada método de prueba.
Al utilizar la biblioteca de cliente API de Python de Google, también obtengo la forma más simple y al mismo tiempo más poderosa de interactuar con mi 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..
Probé todo lo que pude pensar para permitir que estos sean probados de la manera normal. Intenté pulsar directamente los métodos / _ah / spi e incluso intentar crear una nueva aplicación protorpc utilizando service_mappings inútilmente. No soy un miembro de Google en el equipo de puntos finales, así que tal vez tengan algo inteligente para permitir que esto funcione, pero no parece que el simple uso de webtest funcione (a menos que me pierda algo obvio).
Mientras tanto, puede escribir un script de prueba que inicie el servidor de prueba del motor de la aplicación con un entorno aislado y simplemente emita solicitudes http.
Ejemplo para ejecutar el servidor con un entorno aislado (bash pero puedes ejecutarlo fácilmente desde 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 .
A continuación, puede usar solicitudes para probar ala rizo:
requests.get(''http://localhost:8080/_ah/...'')
Si no quiere probar la pila HTTP completa como lo describe Ezequiel Muns, también puede simular endpoints.method y probar su definición de API directamente:
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''))
webtest se puede simplificar para reducir los errores de nomenclatura
para el siguiente 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")
el tests.py debería verse así
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()