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..61f9d256a --- /dev/null +++ b/zulip/integrations/clickup/test_zulip_clickup.py @@ -0,0 +1,150 @@ +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, + "--zulip-webhook-url", + MOCK_WEBHOOK_URL, + ], + ), 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", MOCK_WEBHOOK_URL + ) + + 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, --zulip-webhook-url\n""", + ) + + @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 1[\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 2[\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 3[\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..42df1c5fb --- /dev/null +++ b/zulip/integrations/clickup/zulip_clickup.py @@ -0,0 +1,337 @@ +#!/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 process_url(input_url: str, base_url: str) -> str: + """ + Validates that the URL is the same as 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) + + is_same_domain: bool = parsed_input_url.netloc == parsed_base_url.netloc + auth_code = parse_qs(parsed_input_url.query).get("code") + + if is_same_domain and auth_code: + return auth_code[0] + else: + print("Unable to fetch the auth code.") + 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, endpoint: 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, endpoint) + + if endpoint == "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 initialize_access_token(self, auth_code: str) -> None: + """ + https://clickup.com/api/clickupreference/operation/GetAccessToken/ + """ + endpoint = "oauth/token" + query: Dict[str, str] = { + "client_id": self.client_id, + "client_secret": self.client_secret, + "code": auth_code, + } + data = self.make_clickup_request(endpoint, query, "POST") + if data is None or not data.get("access_token"): + print("Unable to fetch the API key.") + 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/ + """ + endpoint = f"team/{self.team_id}/webhook" + query: Dict[str, Union[str, List[str]]] = { + "endpoint": end_point, + "events": events, + } + data = self.make_clickup_request(endpoint, query, "POST") + if data is None: + print("We're unable to create webhook at the moment.") + sys.exit(1) + return data + + def get_webhooks(self) -> Dict[str, Any]: + """ + https://clickup.com/api/clickupreference/operation/GetWebhooks/ + """ + endpoint = f"team/{self.team_id}/webhook" + data = self.make_clickup_request(endpoint, {}, "GET") + if data is None: + print("We're unable to fetch webhooks at the moment.") + sys.exit(1) + return data + + def delete_webhook(self, webhook_id: str) -> None: + """ + https://clickup.com/api/clickupreference/operation/DeleteWebhook/ + """ + endpoint = f"webhook/{webhook_id}" + data = self.make_clickup_request(endpoint, {}, "DELETE") + if data is None: + print("Failed to delete webhook.") + sys.exit(1) + + +def redirect_to_clickup_auth(zulip_integration_url: str, client_id: str) -> None: + print( + """ +STEP 1 +---- +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) + + +def query_for_authorization_code(zulip_integration_url: str) -> str: + print( + """ +STEP 2 +---- +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 + + +def query_for_notification_events() -> List[str]: + print( + """ +STEP 3 +---- +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 + + Or, enter * to subscribe to all events. + +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_list: str = input("EVENT CODE(s): ") + user_input: List[str] = re.split(",", input_codes_list) + + input_is_valid: bool = len(user_input) > 0 + exhausted_options: List[str] = [] + if "*" in input_codes_list: + print("Subscribing to all events") + all_events = [event for events in EVENT_CHOICES.values() for event in events] + return all_events + 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() + zulip_url_domain = urlparse(zulip_integration_url).netloc + for webhook in data["webhooks"]: + 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, zulip_integration_url: str) -> None: + redirect_to_clickup_auth(zulip_integration_url, client_id) + auth_code = query_for_authorization_code(zulip_integration_url) + api_handler = ClickUpAPIHandler(client_id, client_secret, team_id) + api_handler.initialize_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." + ), + ) + parser.add_argument( + "--zulip-webhook-url", + required=True, + help=("This is the URL your incoming webhook bot has generated."), + ) + + options = parser.parse_args() + print("Running Zulip Clickup Integration...") + + run( + options.clickup_client_id, + options.clickup_client_secret, + options.clickup_team_id, + options.zulip_webhook_url, + ) + + +if __name__ == "__main__": + main()