diff --git a/zulip/integrations/clickup/README.md b/zulip/integrations/clickup/README.md new file mode 100644 index 000000000..1ed8b2db6 --- /dev/null +++ b/zulip/integrations/clickup/README.md @@ -0,0 +1,19 @@ +# 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 \ + --zulip-webhook-url "GENERATED_WEBHOOK_URL" + +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..15d3e9c37 --- /dev/null +++ b/zulip/integrations/clickup/test_zulip_clickup.py @@ -0,0 +1,175 @@ +import io +import re +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 +from integrations.clickup.zulip_clickup import ClickUpAPIHandler + +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]]: + clickup_api = ClickUpAPIHandler(CLICKUP_CLIENT_ID, CLICKUP_CLIENT_SECRET, CLICKUP_TEAM_ID) + api_data_mapper: Dict[str, Dict[str, Dict[str, Any]]] = { # path -> method -> response + clickup_api.ENDPOINTS["oauth"]: { + "POST": {"access_token": MOCK_API_KEY}, + }, + clickup_api.ENDPOINTS["team"]: { + "POST": {"id": MOCK_CREATED_WEBHOOK_ID}, + "GET": {"webhooks": [MOCK_GET_WEBHOOK_IDS]}, + }, + clickup_api.ENDPOINTS["webhook"].format(webhook_id=MOCK_DELETE_WEBHOOK_ID): {"DELETE": {}}, + } + 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( + CLICKUP_CLIENT_ID, CLICKUP_CLIENT_SECRET, CLICKUP_TEAM_ID, 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_redirect_to_auth_page(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={CLICKUP_CLIENT_ID}&redirect_uri={redirect_uri}" + ) + expected_output = ( + r"STEP 1[\s\S]*" + r"ClickUp authorization page will open in your browser\.[\s\S]*" + r"Please authorize your workspace\(s\)\.[\s\S]*" + r"Click 'Connect Workspace' on the page to proceed\.\.\." + ) + + self.assertRegex( + mock_stdout.getvalue(), + expected_output, + ) + + @mock_script_args() + def test_query_for_auth_code(self) -> None: + with patch("webbrowser.open"), patch("sys.stdout", new=io.StringIO()) as mock_stdout: + zulip_clickup.main() + expected_output = ( + "STEP 2\n----\nAfter you've authorized your workspace,\n" + "you should be redirected to your home URL.\n" + "Please copy your home URL and paste it below.\n" + "It should contain a code, and look similar to this:\n\n" + "e.g. " + re.escape(MOCK_AUTH_CODE_URL) + ) + self.assertRegex( + mock_stdout.getvalue(), + expected_output, + ) + + @mock_script_args() + def test_select_clickup_events(self) -> None: + with patch("webbrowser.open"), patch("sys.stdout", new=io.StringIO()) as mock_stdout: + zulip_clickup.main() + expected_output = r""" +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 +""" + self.assertRegex( + mock_stdout.getvalue(), + expected_output, + ) + + @mock_script_args() + def test_success_message(self) -> None: + with patch("webbrowser.open"), patch("sys.stdout", new=io.StringIO()) as mock_stdout: + zulip_clickup.main() + expected_output = 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\." + self.assertRegex(mock_stdout.getvalue(), expected_output) diff --git a/zulip/integrations/clickup/zulip_clickup.py b/zulip/integrations/clickup/zulip_clickup.py new file mode 100644 index 000000000..8c6ecbd82 --- /dev/null +++ b/zulip/integrations/clickup/zulip_clickup.py @@ -0,0 +1,346 @@ +#!/usr/bin/env python3 # noqa: EXE001 +# +# A ClickUp integration script for Zulip. + +import argparse +import json +import re +import sys +import time +import webbrowser +from typing import Any, Dict, List, Optional, Tuple, Union +from urllib.error import HTTPError +from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlunparse +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) + + +def get_example_url_with_auth_code(zulip_integration_url: str) -> str: + parsed_base_url = urlparse(zulip_integration_url) + base_url = urlunparse(parsed_base_url._replace(path="/", query="")) + new_query = urlencode({"code": "332KKA3321NNAK3MADS"}) + print("base_url: ", base_url) + return f"{base_url}?{new_query}" + + +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 + self.ENDPOINTS: Dict[str, str] = { + "oauth": "oath/token", + "team": f"team/{self.team_id}/webhook", + "webhook": "webhook/{webhook_id}", + } + + 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 == self.ENDPOINTS["oauth"]: + 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/ + """ + 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(self.ENDPOINTS["oauth"], 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/ + """ + query: Dict[str, Union[str, List[str]]] = { + "endpoint": end_point, + "events": events, + } + data = self.make_clickup_request(self.ENDPOINTS["team"], 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/ + """ + data = self.make_clickup_request(self.ENDPOINTS["team"], {}, "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 = self.ENDPOINTS["webhook"].format(webhook_id=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: + example_url = get_example_url_with_auth_code(zulip_integration_url) + print( + f""" +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. {example_url} +""" + ) + 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: + 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()