Comment tester à l'unité les points de terminaison Google Cloud

Question

J'ai besoin d'aide pour configurer les tests unittests pour les points de terminaison Google Cloud. À l'aide de WebTest, toutes les demandes répondent avec AppError: Réponse incorrecte: 404 non trouvé. Je ne sais pas vraiment si les ordinateurs d'extrémité sont compatibles avec WebTest.

Voici comment l'application est générée:

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

Ensuite, j'utilise WebTest de cette façon:

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

Les tests avec curl fonctionnent bien.

Dois-je écrire des tests pour des points de terminaison différents? Quelle est la suggestion de l'équipe GAE Endpoints?

Answer 1

Après beaucoup d'expérimentation et de recherche dans le kit de développement de code, je suis venu avec deux méthodes pour tester les points de terminaison dans python:

1. À l'aide de webtest + banc d'essai pour tester le SPI côté

Vous êtes sur la bonne voie avec webtest, mais juste besoin de s'assurer que vous transformer vos demandes pour le SPI point de terminaison.

Le Nuage de points de terminaison de l'API front-end et l' EndpointsDispatcher en dev_appserver transforme les appels d' /_ah/api/* correspondants "backend" appels à l' /_ah/spi/*. La transformation semble être:

• Tous les appels sont application/json HTTP Postes (même si le point de terminaison REST est autre chose).
• Les paramètres de la requête (chemin d'accès, d'interrogation et de JSON corps) sont fusionnés en un seul JSON corps du message.
• Le "backend" point de terminaison utilise le réel python classe et les noms de méthode dans l'URL, par exemple, POST /_ah/spi/TestEndpoint.insert_message appellerons TestEndpoint.insert_message() dans votre code.
• La réponse JSON n'est reformaté avant d'être retournés au client d'origine.

Cela signifie que vous pouvez tester le point de terminaison avec la configuration suivante:

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

La chose ici est que vous pouvez facilement l'installation des appareils appropriés dans la banque de données ou d'autres GAE services avant d'appeler le point de terminaison, ainsi vous permettre de mieux faire valoir les attendus effets secondaires de l'appel.

2. Le démarrage du serveur de développement pour l'intégration complète de test

Vous pouvez démarrer le serveur de dev dans le même environnement python en utilisant quelque chose comme ce qui suit:

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

Maintenant, vous devez émettre réel les requêtes HTTP vers localhost:8123 pour exécuter des tests à l'encontre de l'API, mais encore une fois peut interagir avec GAE Api pour configurer le matériel, etc. C'est évidemment lent que vous êtes à la création et la destruction d'un nouveau serveur de dev pour chaque série de tests.

À ce stade, j'utilise l' API de Google Python client à consommer de l'API au lieu de construire les requêtes HTTP de moi:

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

C'est une amélioration par rapport à l'essai avec CURL, car elle vous donne un accès direct à la FGA Api pour créer facilement des montages et d'inspecter l'état interne. Je soupçonne qu'il ya une meilleure façon de faire des tests d'intégration qui contourne HTTP par assembler le minimum de composants dans le serveur de dev qui mettent en œuvre le point de terminaison d'expédition mécanisme, mais qui nécessite plus de temps de recherche que j'ai en ce moment.

Answer 2

webtest peut être simplifié pour réduire les erreurs de nommage

pour le suivant 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")
 

le tests.py devrait ressembler à ceci

 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

J'ai essayé tout ce que je pouvais penser pour permettre à ces à être testé de façon normale. J'ai essayé de frapper l' /_ah/spi méthodes directement ainsi que même en essayant de créer un nouveau protorpc application à l'aide de service_mappings en vain. Je ne suis pas un Googleurs sur les points de terminaison de l'équipe, donc peut-être qu'ils ont quelque chose d'intelligent pour permettre que cela fonctionne, mais il n'apparaît pas que la simple utilisation d'webtest fonctionne (sauf si j'ai raté quelque chose d'évident).

En attendant, vous pouvez écrire un script de test qui démarre le moteur d'applications serveur de test avec un environnement isolé et il vous suffit d'émettre des requêtes http à elle.

Exemple pour exécuter le serveur avec un environnement isolé (bash, mais vous pouvez facilement exécuter à partir de 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 .

Vous pouvez ensuite simplement utiliser les demandes de test ala curl:

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

Answer 4

Si vous ne souhaitez pas tester la pile HTTP complète comme décrit par Ezequiel Muns, vous pouvez également simuler endpoints.method et tester directement la définition de votre 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'))