Сегодня многие 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