From 070853e2338cd5a7aca7522e8867bf47293fae9b Mon Sep 17 00:00:00 2001 From: pieterck Date: Mon, 1 Apr 2024 14:10:28 +0700 Subject: [PATCH] integrations: Add ClickUp integration script. Add a python script to help integrate Zulip with Clickup. Urlopen is used instead of the usual requests library inorder to make the script standalone. Fixes zulip#26529 --- zulip/integrations/clickup/README.md | 18 + zulip/integrations/clickup/__init__.py | 0 .../clickup/test_zulip_clickup.py | 157 ++++++++ zulip/integrations/clickup/zulip_clickup.py | 372 ++++++++++++++++++ 4 files changed, 547 insertions(+) create mode 100644 zulip/integrations/clickup/README.md create mode 100644 zulip/integrations/clickup/__init__.py create mode 100644 zulip/integrations/clickup/test_zulip_clickup.py create mode 100644 zulip/integrations/clickup/zulip_clickup.py diff --git a/zulip/integrations/clickup/README.md b/zulip/integrations/clickup/README.md new file mode 100644 index 000000000..0cff0a065 --- /dev/null +++ b/zulip/integrations/clickup/README.md @@ -0,0 +1,18 @@ +# A script that automates setting up a webhook with ClickUp + +Usage : + +1. Make sure you have all of the relevant ClickUp credentials before + executing the script: + - The ClickUp Team ID + - The ClickUp Client ID + - The ClickUp Client Secret + +2. Execute the script : + + $ python zulip_clickup.py --clickup-team-id \ + --clickup-client-id \ + --clickup-client-secret \ + +For more information, please see Zulip's documentation on how to set up +a ClickUp integration [here](https://zulip.com/integrations/doc/clickup). diff --git a/zulip/integrations/clickup/__init__.py b/zulip/integrations/clickup/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/zulip/integrations/clickup/test_zulip_clickup.py b/zulip/integrations/clickup/test_zulip_clickup.py new file mode 100644 index 000000000..e70c86ed2 --- /dev/null +++ b/zulip/integrations/clickup/test_zulip_clickup.py @@ -0,0 +1,157 @@ +import io +from functools import wraps +from typing import Any, Callable, Dict, List, Optional, Union +from unittest import TestCase +from unittest.mock import DEFAULT, patch + +from integrations.clickup import zulip_clickup + +MOCK_WEBHOOK_URL = ( + "https://YourZulipApp.com/api/v1/external/clickup?api_key=TJ9DnIiNqt51bpfyPll5n2uT4iYxMBW9" +) + +MOCK_AUTH_CODE = "332KKA3321NNAK3MADS" +MOCK_AUTH_CODE_URL = f"https://YourZulipApp.com/?code={MOCK_AUTH_CODE}" +MOCK_API_KEY = "X" * 32 + +SCRIPT_PATH = "integrations.clickup.zulip_clickup" + +MOCK_CREATED_WEBHOOK_ID = "13-13-13-13-1313-13" +MOCK_DELETE_WEBHOOK_ID = "12-12-12-12-12" +MOCK_GET_WEBHOOK_IDS = {"endpoint": MOCK_WEBHOOK_URL, "id": MOCK_DELETE_WEBHOOK_ID} + +CLICKUP_TEAM_ID = "teamid123" +CLICKUP_CLIENT_ID = "clientid321" +CLICKUP_CLIENT_SECRET = "clientsecret322" # noqa: S105 + + +def make_clickup_request_side_effect( + path: str, query: Dict[str, Union[str, List[str]]], method: str +) -> Optional[Dict[str, Any]]: + api_data_mapper: Dict[str, Dict[str, Dict[str, Any]]] = { # path -> method -> response + "oauth/token": { + "POST": {"access_token": MOCK_API_KEY}, + }, # used for get_access_token() + f"team/{CLICKUP_TEAM_ID}/webhook": { + "POST": {"id": MOCK_CREATED_WEBHOOK_ID}, + "GET": {"webhooks": [MOCK_GET_WEBHOOK_IDS]}, + }, # used for create_webhook(), get_webhooks() + f"webhook/{MOCK_DELETE_WEBHOOK_ID}": {"DELETE": {}}, # used for delete_webhook() + } + return api_data_mapper.get(path, {}).get(method, DEFAULT) + + +def mock_script_args() -> Callable[[Any], Callable[..., Any]]: + def decorator(func: Callable[..., Any]) -> Callable[..., Any]: + @wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + mock_user_inputs = [ + MOCK_WEBHOOK_URL, # input for 1st step + MOCK_AUTH_CODE_URL, # input for 3rd step + "1,2,3,4,5", # third input for 4th step + ] + with patch( + "sys.argv", + [ + "zulip_clickup.py", + "--clickup-team-id", + CLICKUP_TEAM_ID, + "--clickup-client-id", + CLICKUP_CLIENT_ID, + "--clickup-client-secret", + CLICKUP_CLIENT_SECRET, + ], + ), patch("os.system"), patch("time.sleep"), patch("sys.exit"), patch( + "builtins.input", side_effect=mock_user_inputs + ), patch( + SCRIPT_PATH + ".ClickUpAPIHandler.make_clickup_request", + side_effect=make_clickup_request_side_effect, + ): + result = func(*args, **kwargs) + + return result + + return wrapper + + return decorator + + +class ZulipClickUpScriptTest(TestCase): + @mock_script_args() + def test_valid_arguments(self) -> None: + with patch(SCRIPT_PATH + ".run") as mock_run, patch( + "sys.stdout", new=io.StringIO() + ) as mock_stdout: + zulip_clickup.main() + self.assertRegex(mock_stdout.getvalue(), r"Running Zulip Clickup Integration...") + mock_run.assert_called_once_with("clientid321", "clientsecret322", "teamid123") + + def test_missing_arguments(self) -> None: + with self.assertRaises(SystemExit) as cm: + with patch("sys.stderr", new=io.StringIO()) as mock_stderr: + zulip_clickup.main() + self.assertEqual(cm.exception.code, 2) + self.assertRegex( + mock_stderr.getvalue(), + r"""the following arguments are required: --clickup-team-id, --clickup-client-id, --clickup-client-secret\n""", + ) + + @mock_script_args() + def test_step_one(self) -> None: + with patch("webbrowser.open"), patch("sys.stdout", new=io.StringIO()) as mock_stdout: + zulip_clickup.main() + self.assertRegex( + mock_stdout.getvalue(), + ( + r"STEP 1[\s\S]*Please enter the integration URL you've just generated[\s\S]*It should look similar to this:[\s\S]*e.g. http://YourZulipApp\.com/api/v1/external/clickup\?api_key=TJ9DnIiNqt51bpfyPll5n2uT4iYxMBW9" + ), + ) + + @mock_script_args() + def test_step_two(self) -> None: + with patch("webbrowser.open") as mock_open, patch( + "sys.stdout", new=io.StringIO() + ) as mock_stdout: + zulip_clickup.main() + redirect_uri = "https://YourZulipApp.com" + mock_open.assert_called_once_with( + f"https://app.clickup.com/api?client_id=clientid321&redirect_uri={redirect_uri}" + ) + expected_output = r"STEP 2[\s\S]*ClickUp authorization page will open in your browser\.[\s\S]*Please authorize your workspace\(s\)\.[\s\S]*Click 'Connect Workspace' on the page to proceed\.\.\." + self.assertRegex( + mock_stdout.getvalue(), + expected_output, + ) + + @mock_script_args() + def test_step_three(self) -> None: + with patch("webbrowser.open"), patch("sys.stdout", new=io.StringIO()) as mock_stdout: + zulip_clickup.main() + self.assertRegex( + mock_stdout.getvalue(), + ( + r"STEP 3[\s\S]*After you've authorized your workspace,\s*you should be redirected to your home URL.\s*Please copy your home URL and paste it below.\s*It should contain a code, and look similar to this:\s*e.g. https://YourZulipDomain\.com/\?code=332KKA3321NNAK3MADS" + ), + ) + + @mock_script_args() + def test_step_four(self) -> None: + with patch("webbrowser.open"), patch("sys.stdout", new=io.StringIO()) as mock_stdout: + zulip_clickup.main() + self.assertRegex( + mock_stdout.getvalue(), + ( + r"STEP 4[\s\S]*Please select which ClickUp event notification\(s\) you'd[\s\S]*like to receive in your Zulip app\.[\s\S]*EVENT CODES:[\s\S]*1 = task[\s\S]*2 = list[\s\S]*3 = folder[\s\S]*4 = space[\s\S]*5 = goals[\s\S]*Here's an example input if you intend to only receive notifications[\s\S]*related to task, list and folder: 1,2,3" + ), + ) + + @mock_script_args() + def test_final_step(self) -> None: + with patch("webbrowser.open"), patch("sys.stdout", new=io.StringIO()) as mock_stdout: + zulip_clickup.main() + self.assertRegex( + mock_stdout.getvalue(), + ( + r"SUCCESS: Completed integrating your Zulip app with ClickUp!\s*webhook_id: \d+-\d+-\d+-\d+-\d+-\d+\s*You may delete this script or run it again to reconfigure\s*your integration\." + ), + ) diff --git a/zulip/integrations/clickup/zulip_clickup.py b/zulip/integrations/clickup/zulip_clickup.py new file mode 100644 index 000000000..b2ecdcd35 --- /dev/null +++ b/zulip/integrations/clickup/zulip_clickup.py @@ -0,0 +1,372 @@ +#!/usr/bin/env python3 # noqa: EXE001 +# +# A ClickUp integration script for Zulip. + +import argparse +import json +import os +import re +import sys +import time +import webbrowser +from typing import Any, Callable, Dict, List, Optional, Tuple, Union +from urllib.error import HTTPError +from urllib.parse import parse_qs, urlencode, urljoin, urlparse +from urllib.request import Request, urlopen + +EVENT_CHOICES: Dict[str, Tuple[str, ...]] = { + "1": ("taskCreated", "taskUpdated", "taskDeleted"), + "2": ("listCreated", "listUpdated", "listDeleted"), + "3": ("folderCreated", "folderUpdated", "folderDeleted"), + "4": ("spaceCreated", "spaceUpdated", "spaceDeleted"), + "5": ("goalCreated", "goalUpdated", "goalDeleted"), +} + + +def clear_terminal_and_sleep(sleep_duration: int = 3) -> Callable[[Any], Callable[..., Any]]: + """ + Decorator to clear the terminal and sleep for a specified duration + before and after the execution of the decorated function. + """ + cmd = "cls" if os.name == "nt" else "clear" + + def decorator(func: Any) -> Any: + def wrapper(*args: Any, **kwargs: Any) -> Any: + os.system(cmd) # noqa: S605 + result = func(*args, **kwargs) + time.sleep(sleep_duration) + os.system(cmd) # noqa: S605 + return result + + return wrapper + + return decorator + + +def process_url(input_url: str, base_url: str) -> str: + """ + Validates that the URL is the same the users zulip app URL. + Returns the authorization code from the URL query + """ + parsed_input_url = urlparse(input_url) + parsed_base_url = urlparse(base_url) + + same_domain: bool = parsed_input_url.netloc == parsed_base_url.netloc + auth_code = parse_qs(parsed_input_url.query).get("code") + + if same_domain and auth_code: + return auth_code[0] + else: + print("Unable to fetch the auth code. exiting") + sys.exit(1) + + +class ClickUpAPIHandler: + def __init__( + self, + client_id: str, + client_secret: str, + team_id: str, + ) -> None: + self.client_id: str = client_id + self.client_secret: str = client_secret + self.team_id: str = team_id + self.API_KEY: Optional[str] = None + + def make_clickup_request( + self, path: str, query: Dict[str, Union[str, List[str]]], method: str + ) -> Optional[Dict[str, Any]]: + base_url = "https://api.clickup.com/api/v2/" + api_endpoint = urljoin(base_url, path) + + if path == "oath/token": + encoded_query = urlencode(query).encode("utf-8") + req = Request(api_endpoint, data=encoded_query, method=method) # noqa: S310 + else: + headers: Dict[str, str] = { + "Content-Type": "application/json", + "Authorization": self.API_KEY if self.API_KEY else "", + } + encoded_query = json.dumps(query).encode("utf-8") + req = Request( # noqa: S310 + api_endpoint, data=encoded_query, headers=headers, method=method + ) + + try: + with urlopen(req) as response: # noqa: S310 + if response.status != 200: + print(f"Error : {response.status}") + sys.exit(1) + data: Dict[str, str] = json.loads(response.read().decode("utf-8")) + return data + except HTTPError as err: + print(f"HTTPError occurred: {err.code} {err.reason}") + return None + + def get_access_token(self, auth_code: str) -> None: + """ + https://clickup.com/api/clickupreference/operation/GetAccessToken/ + """ + path = "oauth/token" + query: Dict[str, Union[str, List[str]]] = { + "client_id": self.client_id, + "client_secret": self.client_secret, + "code": auth_code, + } + data = self.make_clickup_request(path, query, "POST") + if data is None or not data.get("access_token"): + print("Unable to fetch the API key. exiting") + sys.exit(1) + self.API_KEY = data.get("access_token") + + def create_webhook(self, end_point: str, events: List[str]) -> Dict[str, Any]: + """ + https://clickup.com/api/clickupreference/operation/CreateWebhook/ + """ + path = f"team/{self.team_id}/webhook" + query: Dict[str, Union[str, List[str]]] = { + "endpoint": end_point, + "events": events, + } + data = self.make_clickup_request(path, query, "POST") + if data is None: + print("We're unable to create webhook at the moment. exiting") + sys.exit(1) + return data + + def get_webhooks(self) -> Dict[str, Any]: + """ + https://clickup.com/api/clickupreference/operation/GetWebhooks/ + """ + path = f"team/{self.team_id}/webhook" + data = self.make_clickup_request(path, {}, "GET") + if data is None: + print("We're unable to fetch webhooks at the moment. exiting") + sys.exit(1) + return data + + def delete_webhook(self, webhook_id: str) -> None: + """ + https://clickup.com/api/clickupreference/operation/DeleteWebhook/ + """ + path = f"webhook/{webhook_id}" + data = self.make_clickup_request(path, {}, "DELETE") + if data is None: + print("Failed to delete webhook. exiting") + sys.exit(1) + + +@clear_terminal_and_sleep(1) +def query_for_integration_url() -> str: + print( + """ + STEP 1 + ---- + Please enter the integration URL you've just generated + from your Zulip app settings. + + It should look similar to this: + e.g. http://YourZulipApp.com/api/v1/external/clickup?api_key=TJ9DnIiNqt51bpfyPll5n2uT4iYxMBW9 + """ + ) + while True: + input_url: str = input("INTEGRATION URL: ") + if input_url: + break + return input_url + + +@clear_terminal_and_sleep(3) +def redirect_to_clickup_auth(zulip_integration_url: str, client_id: str) -> None: + print( + """ + STEP 2 + ---- + ClickUp authorization page will open in your browser. + Please authorize your workspace(s). + + Click 'Connect Workspace' on the page to proceed... + """ + ) + parsed_url = urlparse(zulip_integration_url) + base_url: str = f"{parsed_url.scheme}://{parsed_url.netloc}" + url: str = f"https://app.clickup.com/api?client_id={client_id}&redirect_uri={base_url}" + time.sleep(1) + webbrowser.open(url) + + +@clear_terminal_and_sleep(2) +def query_for_authorization_code(zulip_integration_url: str) -> str: + print( + """ + STEP 3 + ---- + After you've authorized your workspace, + you should be redirected to your home URL. + Please copy your home URL and paste it below. + It should contain a code, and look similar to this: + + e.g. https://YourZulipDomain.com/?code=332KKA3321NNAK3MADS + """ + ) + input_url: str = input("YOUR HOME URL: ") + auth_code: str = process_url(input_url=input_url, base_url=zulip_integration_url) + return auth_code + + +@clear_terminal_and_sleep(1) +def query_for_notification_events() -> List[str]: + print( + """ + STEP 4 + ---- + Please select which ClickUp event notification(s) you'd + like to receive in your Zulip app. + EVENT CODES: + 1 = task + 2 = list + 3 = folder + 4 = space + 5 = goals + + Here's an example input if you intend to only receive notifications + related to task, list and folder: 1,2,3 + """ + ) + querying_user_input: bool = True + selected_events: List[str] = [] + + while querying_user_input: + input_codes: str = input("EVENT CODE(s): ") + user_input: List[str] = re.split(",", input_codes) + + input_is_valid: bool = len(user_input) > 0 + exhausted_options: List[str] = [] + + for event_code in user_input: + if event_code in EVENT_CHOICES and event_code not in exhausted_options: + selected_events += EVENT_CHOICES[event_code] + exhausted_options.append(event_code) + else: + input_is_valid = False + + if not input_is_valid: + print("Please enter a valid set of options and only select each option once") + + querying_user_input = not input_is_valid + + return selected_events + + +def delete_old_webhooks(zulip_integration_url: str, api_handler: ClickUpAPIHandler) -> None: + """ + Checks for existing webhooks with the same endpoint and delete them if found. + """ + data: Dict[str, Any] = api_handler.get_webhooks() + for webhook in data["webhooks"]: + print("webhook: ", webhook) + zulip_url_domain = urlparse(zulip_integration_url).netloc + registered_webhook_domain = urlparse(webhook["endpoint"]).netloc + + if zulip_url_domain in registered_webhook_domain: + api_handler.delete_webhook(webhook["id"]) + + +def display_success_msg(webhook_id: str) -> None: + print( + f""" + SUCCESS: Completed integrating your Zulip app with ClickUp! + webhook_id: {webhook_id} + + You may delete this script or run it again to reconfigure + your integration. + """ + ) + + +def add_query_params(url: str, params: Dict[str, List[str]]) -> str: + parsed_url = urlparse(url) + query_dict = parse_qs(parsed_url.query) + query_dict.update(params) + return parsed_url._replace(query=urlencode(query_dict)).geturl() + + +def run(client_id: str, client_secret: str, team_id: str) -> None: + zulip_integration_url = query_for_integration_url() + + redirect_to_clickup_auth(zulip_integration_url, client_id) + auth_code: str = query_for_authorization_code(zulip_integration_url) + api_handler = ClickUpAPIHandler(client_id, client_secret, team_id) + api_handler.get_access_token(auth_code) + events_payload: List[str] = query_for_notification_events() + delete_old_webhooks( + zulip_integration_url, api_handler + ) # to avoid setting up multiple identical webhooks + + zulip_webhook_url = add_query_params( + zulip_integration_url, + { + "clickup_api_key": [api_handler.API_KEY if api_handler.API_KEY else ""], + "team_id": [team_id], + }, + ) + + response: Dict[str, Any] = api_handler.create_webhook( + end_point=zulip_webhook_url, events=events_payload + ) + + display_success_msg(response["id"]) + sys.exit(0) + + +def main() -> None: + description = """ + zulip_clickup.py is a handy little script that allows Zulip users to + quickly set up a ClickUp webhook. + + Note: The ClickUp webhook instructions available on your Zulip server + may be outdated. Please make sure you follow the updated instructions + at . + """ + + parser = argparse.ArgumentParser(description=description) + + parser.add_argument( + "--clickup-team-id", + required=True, + help=( + "Your team_id is the numbers immediately following the base ClickUp URL" + "https://app.clickup.com/25567147/home" + "For instance, the team_id for the URL above would be 25567147" + ), + ) + + parser.add_argument( + "--clickup-client-id", + required=True, + help=( + "Visit https://clickup.com/api/developer-portal/authentication/#step-1-create-an-oauth-app" + "and follow 'Step 1: Create an OAuth app' to generate client_id & client_secret." + ), + ) + parser.add_argument( + "--clickup-client-secret", + required=True, + help=( + "Visit https://clickup.com/api/developer-portal/authentication/#step-1-create-an-oauth-app" + "and follow 'Step 1: Create an OAuth app' to generate client_id & client_secret." + ), + ) + + options = parser.parse_args() + print("Running Zulip Clickup Integration...") + + run( + options.clickup_client_id, + options.clickup_client_secret, + options.clickup_team_id, + ) + + +if __name__ == "__main__": + main()