mirror of
https://github.com/zulip/zulip.git
synced 2025-11-04 05:53:43 +00:00
119 lines
4.1 KiB
Python
Executable File
119 lines
4.1 KiB
Python
Executable File
#!/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
|
|
import subprocess
|
|
from typing import Any, Dict
|
|
|
|
import requests
|
|
import ujson
|
|
|
|
from zerver.models import UserProfile, Message, 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)
|
|
# Delete all messages, so new message is the only one it's message group
|
|
Message.objects.filter(sender=bot).delete()
|
|
|
|
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(response.json())
|
|
print('Failed to trigger webhook')
|
|
sys.exit(1)
|
|
|
|
print('Triggered {} webhook'.format(integration.name))
|
|
message_id = str(Message.objects.filter(sender=bot).last().id)
|
|
screenshot_script = os.path.join(TOOLS_DIR, 'message-screenshot.js')
|
|
subprocess.check_call(['node', screenshot_script, integration.name, message_id])
|