So testen Sie Google Cloud Endpunkte

Ich brauche Hilfe bei der Einrichtung von Unentschieden für Google Cloud Endpunkte. Mit WebTest antworten alle Anfragen mit AppError: Schlechte Antwort: 404 Nicht gefunden. Ich bin nicht ganz sicher, ob Endpunkte mit WebTest kompatibel sind.

So wird die Anwendung generiert:

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

Dann benutze ich WebTest auf diese Weise:

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

Testen mit Curl funktioniert gut.

Soll ich Tests für Endpunkte schreiben? Was ist der Vorschlag von GAE Endpoints Team?

  • Wie fügt man Bulk-Daten in Google App Engine Datastore ein?
  • Name der Datei hochladen - Google App Engine Python
  • Wie kann ich auf den Produktionsdatenspeicher von meinem lokalen Entwicklungsserver zugreifen?
  • GAE "kein Attribut 'HTTPSHandler'" dev_appserver.py
  • Jinja2 gibt "Keine" Zeichenfolge für Google App Engine-Modelle zurück
  • Dateien in Google App Engine hochladen
  • 5 Solutions collect form web for “So testen Sie Google Cloud Endpunkte”

    Nach viel Experimentieren und Blick auf die SDK-Code habe ich kommen mit zwei Möglichkeiten, um Endpunkte in Python zu testen:

    1. Verwenden Sie webtest + testbed, um die SPI-Seite zu testen

    Sie sind auf dem richtigen Weg mit Webtest, aber müssen Sie nur sicherstellen, dass Sie Ihre Anfragen korrekt für den SPI-Endpunkt umwandeln.

    Die Cloud-Endpunkte API-Front-End und der EndpointsDispatcher in dev_appserver verwandeln Anrufe in /_ah/api/* in entsprechende "Backend" -Aufrufe an /_ah/spi/* . Die Umwandlung scheint zu sein:

    • Alle Anrufe sind application/json HTTP POSTs (auch wenn der REST Endpunkt etwas anderes ist).
    • Die Anforderungsparameter (Pfad, Abfrage und JSON-Körper) werden zusammen zu einer einzigen JSON-Körpernachricht zusammengeführt.
    • Der "Backend" Endpunkt verwendet die tatsächliche POST /_ah/spi/TestEndpoint.insert_message und Methodennamen in der URL, zB POST /_ah/spi/TestEndpoint.insert_message ruft TestEndpoint.insert_message() in deinem Code auf.
    • Die JSON-Antwort wird nur neu formatiert, bevor sie an den ursprünglichen Client zurückgegeben wird.

    So können Sie den Endpunkt mit folgendem Setup testen:

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

    Die Sache hier ist, können Sie ganz einfach passende Fixtures in den Datenspeicher oder andere GAE-Dienste vor dem Aufruf der Endpunkt, so können Sie mehr vollständig behaupten, die erwarteten Nebenwirkungen des Anrufs.

    2. Starten des Entwicklungs-Servers für den Vollintegrationstest

    Sie können den dev-Server in der gleichen Python-Umgebung mit so etwas wie dem folgenden starten:

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

    Jetzt müssen Sie aktuelle HTTP-Anfragen an localhost ausgeben: 8123, um Tests gegen die API auszuführen, aber wieder kann mit GAE-APIs interagieren, um Fixtures usw. einzurichten. Dies ist offensichtlich langsam, wie Sie erstellen und zerstören einen neuen dev-Server für jeden Testlauf.

    An diesem Punkt verwende ich den Google API-Python-Client , um die API zu konsumieren, anstatt die HTTP-Anfragen selbst zu erstellen:

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

    Dies ist eine Verbesserung gegenüber dem Testen mit CURL, da es Ihnen einen direkten Zugriff auf die GAE-APIs ermöglicht, um Fixtures einfach einzurichten und den internen Zustand zu überprüfen. Ich vermute, es gibt einen noch besseren Weg, um Integrationstests durchzuführen, die HTTP umgehen, indem sie die minimalen Komponenten im Dev-Server zusammenfügen, die den Endpunkt-Dispatch-Mechanismus implementieren, aber das erfordert mehr Forschungszeit als ich jetzt habe.

    Webtest kann vereinfacht werden, um Namensfehler zu reduzieren

    Für die folgenden 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") 

    Die tests.py sollte so aussehen

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

    Ich habe alles versucht, was ich mir vorstellen kann, um diese auf normale Weise zu testen. Ich habe versucht, die / _ah / spi-Methoden direkt zu schlagen und sogar zu versuchen, eine neue protorpc App mit service_mappings ohne Erfolg zu erstellen. Ich bin kein Googler auf dem Endpunkt-Team, vielleicht haben sie doch etwas Kluges, um das zu arbeiten, aber es scheint nicht, dass einfach mit Webtest arbeiten wird (es sei denn, ich habe etwas offensichtlich verpasst).

    In der Zwischenzeit können Sie ein Test-Skript schreiben, das den App-Engine-Test-Server mit einer isolierten Umgebung startet und einfach nur HTTP-Anfragen ausstellt.

    Beispiel, um den Server mit einer isolierten Umgebung auszuführen (bash, aber du kannst das einfach aus Python laufen lassen):

     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 . 

    Sie können dann nur Anfragen verwenden, um ala curl zu testen:

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

    Wenn du den kompletten HTTP-Stack nicht testen möchtest, wie von Ezequiel Muns beschrieben, kannst du auch endpoints.method ausschalten und deine API-Definition direkt testen:

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

    Meine Lösung verwendet eine dev_appserver-Instanz für das gesamte Testmodul, was schneller ist als das Wiederholen des dev_appserver für jede Testmethode.

    Durch die Verwendung der Google-Python-API-Client-Bibliothek bekomme ich auch die einfachste und zugleich leistungsstärkste Art, mit meiner API zu interagieren.

     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.. 
    Python ist die beste Programmiersprache der Welt.