Подтвердить что ты не робот

Как unit test Конечные точки Google Cloud

Мне нужна помощь в настройке unittests для Google Cloud Endpoints. Используя WebTest, все запросы отвечают с AppError: Плохой ответ: 404 Не найдено. Я не уверен, совместимы ли конечные точки с WebTest.

Вот как создается приложение:

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

Затем я использую WebTest следующим образом:

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

Тестирование с помощью curl отлично работает.

Должен ли я писать тесты для конечных точек разными? Каково предложение команды GAE Endpoints?

4b9b3361

ОТВЕТЫ

Ответ 1

После долгих экспериментов и поиска кода SDK я нашел два способа тестирования конечных точек в python:

1. Использование webtest + testbed для тестирования стороны SPI

Вы находитесь на правильном пути с помощью веб-сайта, но просто убедитесь, что вы правильно преобразовали свои запросы для конечной точки SPI.

Интерфейс API конечных точек Cloud и EndpointsDispatcher в dev_appserver преобразует вызовы на /_ah/api/* в соответствующие "backend" вызовы на /_ah/spi/*. Кажется, что преобразование:

  • Все вызовы application/json HTTP POST (даже если конечная точка REST - это что-то еще).
  • Параметры запроса (путь, запрос и тело JSON) объединены вместе в одно сообщение тела JSON.
  • Конечная точка "backend" использует фактические имена класса и метода python в URL-адресе, например. 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') #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. Запуск сервера разработки для полного теста интеграции

Вы можете запустить сервер dev в той же среде python, используя что-то вроде следующего:

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 для запуска тестов против API, но снова может взаимодействовать с API-интерфейсами GAE для настройки приборов и т.д. Это явно медленно, поскольку вы создаете и уничтожаете нового разработчика сервер для каждого тестового прогона.

В этот момент я использую клиент 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, { ... expected reponse as dict ... })

Это улучшение по сравнению с CURL, поскольку оно дает вам прямой доступ к API GAE, чтобы легко настроить приборы и проверить внутреннее состояние. Я подозреваю, что есть еще лучший способ провести интеграционное тестирование, которое обходит HTTP, сшивая минимальные компоненты на сервере-разработчике, которые реализуют механизм отправки конечных точек, но для этого требуется больше времени на исследования, чем сейчас.

Ответ 2

веб-сайт может быть упрощен, чтобы уменьшить количество именований ошибок

для следующих 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")

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

Ответ 3

Я попробовал все, что мог придумать, чтобы они были протестированы обычным способом. Я попытался напрямую использовать методы /_ah/spi, а также попытался создать новое приложение protorpc с помощью service_mappings безрезультатно. Я не Googler в команде конечных точек, поэтому, возможно, у них есть что-то умное, чтобы это работало, но не кажется, что просто использование веб-теста будет работать (если только я не пропустил что-то очевидное).

Тем временем вы можете написать тест script, который запускает тестовый сервер движка приложений с изолированной средой и просто выдает ему 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 .

Затем вы можете просто использовать запросы для проверки ala curl:

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

Ответ 4

Если вы не хотите протестировать полный стек HTTP, как описано Ezequiel Muns, вы также можете просто издеваться над endpoints.method и непосредственно проверить свое определение 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'))

Ответ 5

В моем решении используется один экземпляр dev_appserver для всего тестового модуля, который быстрее, чем перезапуск 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..