zulip/tools/generate-integration-docs-s...

111 lines
3.7 KiB
Plaintext
Raw Normal View History

#!/usr/bin/env python3
"""Create or update a webhook integration screenshot using a test fixture."""
# check for the venv
from lib import sanity_check
sanity_check.check_venv(__file__)
import os
import sys
TOOLS_DIR = os.path.abspath(os.path.dirname(__file__))
ROOT_DIR = os.path.dirname(TOOLS_DIR)
sys.path.insert(0, ROOT_DIR)
from scripts.lib.setup_path import setup_path
setup_path()
os.environ["DJANGO_SETTINGS_MODULE"] = "zproject.settings"
import django
django.setup()
import argparse
from typing import Any, Dict
import requests
import ujson
from zerver.models import UserProfile, get_user_by_delivery_email, get_realm
from zerver.lib.actions import do_create_user, notify_created_bot
from zerver.lib.upload import upload_avatar_image
from zerver.lib.actions import do_change_avatar_fields
from zerver.lib.integrations import WebhookIntegration, INTEGRATIONS, split_fixture_path
from zerver.lib.webhooks.common import get_fixture_http_headers
from setup.generate_zulip_bots_static_files import create_png_from_svg
def create_integration_bot(integration_name: str) -> UserProfile:
realm = get_realm('zulip')
owner = get_user_by_delivery_email("iago@zulip.com", realm)
bot_email = "{}-bot@example.com".format(integration_name)
bot_name = "{} Bot".format(integration_name.capitalize())
try:
bot = UserProfile.objects.get(email=bot_email)
except UserProfile.DoesNotExist:
bot = do_create_user(
email=bot_email,
password="123",
realm=owner.realm,
full_name=bot_name,
short_name=bot_name,
bot_type=UserProfile.INCOMING_WEBHOOK_BOT,
bot_owner=owner,
)
notify_created_bot(bot)
if integration.logo_url is None:
return bot
logo_relative_path = integration.logo_url[len(realm.uri) + 1:]
logo_path = os.path.join(ROOT_DIR, logo_relative_path)
if logo_path.endswith(".svg"):
logo_path = create_png_from_svg(logo_path)
with open(logo_path, "rb") as f:
upload_avatar_image(f, owner, bot)
do_change_avatar_fields(bot, UserProfile.AVATAR_FROM_USER)
return bot
def get_integration(integration_name: str) -> WebhookIntegration:
integration = INTEGRATIONS[integration_name]
assert isinstance(integration, WebhookIntegration), "Not a WebhookIntegration"
return integration
def get_requests_headers(integration_name: str, fixture_name: str) -> Dict[str, Any]:
headers = get_fixture_http_headers(integration_name, fixture_name)
def fix_name(header: str) -> str:
header = header if not header.startswith('HTTP_') else header[len('HTTP_'):]
return header.replace('_', '-')
return {fix_name(k): v for k, v in headers.items()}
def webhook_json_fixture(path: str) -> str:
path = os.path.abspath(path)
if not (os.path.exists(path) and path.endswith('.json') and 'webhooks' in path):
raise ValueError('Not a valid webhook JSON fixture')
return path
parser = argparse.ArgumentParser()
parser.add_argument('fixture', type=webhook_json_fixture, help='Path to the fixture to use')
options = parser.parse_args()
integration_name, fixture_name = split_fixture_path(options.fixture)
integration = get_integration(integration_name)
bot = create_integration_bot(integration.name)
assert isinstance(bot.bot_owner, UserProfile)
url = "{}/{}?api_key={}&stream=devel".format(
bot.bot_owner.realm.uri, integration.url, bot.api_key
)
with open(options.fixture) as f:
data = ujson.load(f)
headers = get_requests_headers(integration_name, fixture_name)
response = requests.post(url, json=data, headers=headers)
if response.status_code == 200:
print('Triggered {} webhook'.format(integration.name))
else:
print(response.json())