Статьи

Тестирование обратных вызовов HTTP API веб-хука с помощью ngrok в Python

Сегодня многие API-сервисы предоставляют веб-пользователям обратный вызов вашего веб-сайта или системы через HTTP. Это позволяет простым сторонним межпроцессным коммуникациям и уведомлениям для веб-сайтов. Однако, если вы не работаете в производственной среде, вы часто попадаете в ситуацию, когда невозможно получить доступную через Интернет конечную точку HTTP через общедоступный IP-адрес. Эти ситуации могут включать ваш домашний рабочий стол, общедоступную точку доступа WI-FI или службы непрерывной интеграции. Таким образом, разработка или тестирование на основе API webhook становятся болезненными для современных разработчиков кочевников.

ngrok  ( источник ) — это сервис «плати сколько хочешь» для создания HTTP-туннелей через сторонние реле. Что делает ngrok привлекательным, так это то, что регистрация очень проста с учетными данными Github и авансовые платежи не требуются. ngrok также имеет открытый исходный код, поэтому вы можете запустить собственный ретранслятор для конфиденциального трафика.

В этом посте я представляю решение Python, как программно создавать туннели ngrok по требованию. Это особенно полезно для модульных тестов webhook, поскольку у вас есть нулевые туннели конфигурации, доступные везде, где вы запускаете свой код. ngrok создается как контролируемый подпроцесс для данного URL. Затем вы можете указать своему поставщику услуг webhook использовать этот URL-адрес для повторного вызова ваших модульных тестов.

Можно использовать ngrok полностью логин бесплатно. В этом случае вы теряете возможность называть свои конечные точки HTTP. Я считаю практичным иметь контроль над URL-адресами конечных точек, поскольку это значительно упрощает отладку.

Для реального использования вы можете проверить  проект cryptoassets.core, в  котором я использовал метод ngrok. ngrok успешно вытащил меня из  службы drone.io CI  и моего ноутбука.

Монтаж

Установка ngrok на OSX из  Homebrew :

brew install ngrok

Установка ngrok для Ubuntu:

apt-get install -y unzip
cd /tmp
wget -O ngrok.zip "https://api.equinox.io/1/Applications/ap_pJSFC5wQYkAyI0FIVwKYs9h1hW/Updates/Asset/ngrok.zip?os=linux&arch=386&channel=stable"
unzip ngrok
mv ngrok /usr/local/bin

Официальная загрузка ngrok, автономные молнии .

Зарегистрируйтесь в службе ngrok и получите свой токен авторизации.

Экспортируйте токен аутентификации как переменную среды в вашей оболочке, не храните его в системе контроля версий:

export NGROK_AUTH_TOKEN=xxx

Код туннеля Нгрок

Ниже приведен код Python 3 для класса NgrokTunnel. Смотрите  полный исходный код здесь .

import os
import time
import uuid
import logging
import subprocess
from distutils.spawn import find_executable


logger = logging.getLogger(__name__)


class NgrokTunnel:

    def __init__(self, port, auth_token, subdomain_base="zoq-fot-pik"):
        """Initalize Ngrok tunnel.

        :param auth_token: Your auth token string you get after logging into ngrok.com

        :param port: int, localhost port forwarded through tunnel

        :parma subdomain_base: Each new tunnel gets a generated subdomain. This is the prefix used for a random string.
        """
        assert find_executable("ngrok"), "ngrok command must be installed, see https://ngrok.com/"
        self.port = port
        self.auth_token = auth_token
        self.subdomain = "{}-{}".format(subdomain_base, str(uuid.uuid4()))

    def start(self, ngrok_die_check_delay=0.5):
        """Starts the thread on the background and blocks until we get a tunnel URL.

        :return: the tunnel URL which is now publicly open for your localhost port
        """

        logger.debug("Starting ngrok tunnel %s for port %d", self.subdomain, self.port)

        self.ngrok = subprocess.Popen(["ngrok", "-authtoken={}".format(self.auth_token), "-log=stdout", "-subdomain={}".format(self.subdomain), str(self.port)], stdout=subprocess.DEVNULL)

        # See that we don't instantly die
        time.sleep(ngrok_die_check_delay)
        assert self.ngrok.poll() is None, "ngrok terminated abrutly"
        url = "https://{}.ngrok.com".format(self.subdomain)
        return url

    def stop(self):
        """Tell ngrok to tear down the tunnel.

        Stop the background tunneling process.
        """
        self.ngrok.terminate()

Пример использования в тестах

Вот небольшой псевдо-пример из модульных тестов обработчика webhook cryptoassets.core block.io. Смотрите полный код модульного теста здесь .

class BlockWebhookTestCase(CoinTestRoot, unittest.TestCase):

    def setUp(self):

        self.ngrok = None

        self.backend.walletnotify_config["class"] = "cryptoassets.core.backend.blockiowebhook.BlockIoWebhookNotifyHandler"

        # We need ngrok tunnel for webhook notifications
        auth_token = os.environ["NGROK_AUTH_TOKEN"]
        self.ngrok = NgrokTunnel(21211, auth_token)

        # Pass dynamically generated tunnel URL to backend config
        tunnel_url = self.ngrok.start()
        self.backend.walletnotify_config["url"] = tunnel_url
        self.backend.walletnotify_config["port"] = 21211

        # Start the web server
        self.incoming_transactions_runnable = self.backend.setup_incoming_transactions(self.app.conflict_resolver, self.app.event_handler_registry)

        self.incoming_transactions_runnable.start()

    def teardown(self):

        # Stop webserver
        incoming_transactions_runnable = getattr(self, "incoming_transactions_runnable", None)
        if incoming_transactions_runnable:
            incoming_transactions_runnable.stop()

        # Stop tunnelling
        if self.ngrok:
            self.ngrok.stop()
            self.ngrok = None