diff --git a/zulip/integrations/clickup/README.md b/zulip/integrations/clickup/README.md new file mode 100644 index 000000000..3807285cd --- /dev/null +++ b/zulip/integrations/clickup/README.md @@ -0,0 +1,39 @@ +# A script that automates setting up a webhook with ClickUp + +This script registers a ClickUp webhook that points at your Zulip +incoming webhook bot, so that ClickUp events show up as messages in +Zulip. + +Usage: + +1. Gather the required credentials before running the script: + - **ClickUp Team ID** — the number immediately following the base + ClickUp URL. For `https://app.clickup.com/25567147/home` the team + ID is `25567147`. + - **ClickUp personal API token** — generate one in ClickUp under + *Settings > ClickUp API > API Token* (it looks like `pk_...`). The script + uses it to register the webhook, and you enter the same token when + creating the Zulip bot so the server can resolve ClickUp entity + IDs to human-readable names. + - **Zulip webhook URL** — the URL generated by your Zulip incoming + webhook bot. + +2. Run the script: + + ``` + $ python zulip_clickup.py \ + --clickup-team-id \ + --clickup-api-key \ + --zulip-webhook-url "" + ``` + + The script asks which ClickUp events you'd like to receive, then + registers the webhook. + +3. (Optional) Pass `--replace-existing` to delete any webhook already + pointing at the same Zulip URL before creating a new one. Use this + when re-running the script so you don't end up with duplicate + webhooks. + +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..cdf3b3a5e --- /dev/null +++ b/zulip/integrations/clickup/test_zulip_clickup.py @@ -0,0 +1,291 @@ +import io +from typing import Any, List +from unittest import TestCase +from unittest.mock import MagicMock, patch +from urllib.error import HTTPError, URLError + +from typing_extensions import override + +from integrations.clickup import zulip_clickup +from integrations.clickup.zulip_clickup import ClickUpAPIHandler + +SCRIPT_PATH = "integrations.clickup.zulip_clickup" + +MOCK_WEBHOOK_URL = ( + "https://YourZulipApp.com/api/v1/external/clickup?api_key=TJ9DnIiNqt51bpfyPll5n2uT4iYxMBW9" +) + +MOCK_API_KEY = "pk_" + "1" * 32 +CLICKUP_TEAM_ID = "teamid123" + +MOCK_CREATED_WEBHOOK_ID = "13-13-13-13-1313-13" +MOCK_DELETE_WEBHOOK_ID = "12-12-12-12-12" + +REQUIRED_ARGS = [ + "zulip_clickup.py", + "--clickup-team-id", + CLICKUP_TEAM_ID, + "--clickup-api-key", + MOCK_API_KEY, + "--zulip-webhook-url", + MOCK_WEBHOOK_URL, +] + + +def make_urlopen_response(status: int, body: bytes) -> MagicMock: + """Build a mock that behaves like the context manager urlopen() returns.""" + response = MagicMock() + response.status = status + response.read.return_value = body + response.__enter__.return_value = response + response.__exit__.return_value = False + return response + + +class ArgumentParsingTest(TestCase): + @patch("sys.stdout", new_callable=io.StringIO) + @patch(SCRIPT_PATH + ".run") + @patch("sys.argv", REQUIRED_ARGS) + def test_valid_arguments(self, mock_run: MagicMock, mock_stdout: io.StringIO) -> None: + zulip_clickup.main() + self.assertRegex(mock_stdout.getvalue(), r"Running Zulip Clickup Integration...") + mock_run.assert_called_once_with(CLICKUP_TEAM_ID, MOCK_API_KEY, MOCK_WEBHOOK_URL, False) + + @patch("sys.stdout", new_callable=io.StringIO) + @patch(SCRIPT_PATH + ".run") + @patch("sys.argv", [*REQUIRED_ARGS, "--replace-existing"]) + def test_replace_existing_flag(self, mock_run: MagicMock, mock_stdout: io.StringIO) -> None: + zulip_clickup.main() + self.assertRegex(mock_stdout.getvalue(), r"Running Zulip Clickup Integration...") + mock_run.assert_called_once_with(CLICKUP_TEAM_ID, MOCK_API_KEY, MOCK_WEBHOOK_URL, True) + + @patch("sys.stderr", new_callable=io.StringIO) + @patch("sys.argv", ["zulip_clickup.py"]) + def test_missing_arguments(self, mock_stderr: io.StringIO) -> None: + with self.assertRaises(SystemExit) as cm: + zulip_clickup.main() + self.assertEqual(cm.exception.code, 2) + self.assertRegex( + mock_stderr.getvalue(), + r"the following arguments are required: " + r"--clickup-team-id, --clickup-api-key, --zulip-webhook-url", + ) + + +class EventSelectionTest(TestCase): + @patch("sys.stdout", new_callable=io.StringIO) + @patch("builtins.input", side_effect=["1"]) + def test_select_event_group(self, mock_input: MagicMock, mock_stdout: io.StringIO) -> None: + events = zulip_clickup.query_for_notification_events() + menu = mock_stdout.getvalue() + for expected in ( + "1 = task", + "2 = list", + "3 = folder", + "4 = space", + "5 = goal", + "6 = key result", + ): + self.assertIn(expected, menu) + self.assertEqual(events, ["taskCreated", "taskUpdated", "taskDeleted"]) + + @patch("sys.stdout", new_callable=io.StringIO) + @patch("builtins.input", side_effect=["1,2,3"]) + def test_select_multiple_event_groups( + self, mock_input: MagicMock, mock_stdout: io.StringIO + ) -> None: + events = zulip_clickup.query_for_notification_events() + self.assertEqual( + events, + [ + "taskCreated", + "taskUpdated", + "taskDeleted", + "listCreated", + "listUpdated", + "listDeleted", + "folderCreated", + "folderUpdated", + "folderDeleted", + ], + ) + + @patch("sys.stdout", new_callable=io.StringIO) + @patch("builtins.input", side_effect=["*"]) + def test_select_all_events(self, mock_input: MagicMock, mock_stdout: io.StringIO) -> None: + events = zulip_clickup.query_for_notification_events() + expected = [event for group in zulip_clickup.EVENT_CHOICES.values() for event in group] + self.assertEqual(events, expected) + + @patch("sys.stdout", new_callable=io.StringIO) + @patch("builtins.input", side_effect=["9", "1"]) + def test_invalid_then_valid_input( + self, mock_input: MagicMock, mock_stdout: io.StringIO + ) -> None: + events = zulip_clickup.query_for_notification_events() + self.assertEqual(events, ["taskCreated", "taskUpdated", "taskDeleted"]) + self.assertRegex( + mock_stdout.getvalue(), + r"Please enter a valid set of options and only select each option once", + ) + + +class ClickUpAPIHandlerTest(TestCase): + @override + def setUp(self) -> None: + self.handler = ClickUpAPIHandler(MOCK_API_KEY, CLICKUP_TEAM_ID) + + def test_constructor_and_endpoints(self) -> None: + self.assertEqual(self.handler.api_key, MOCK_API_KEY) + self.assertEqual(self.handler.team_id, CLICKUP_TEAM_ID) + self.assertEqual(set(self.handler.ENDPOINTS), {"team", "webhook"}) + + @patch(SCRIPT_PATH + ".urlopen", return_value=make_urlopen_response(200, b'{"id": "abc"}')) + def test_valid_request(self, mock_urlopen: MagicMock) -> None: + data = self.handler.make_clickup_request( + self.handler.ENDPOINTS["team"], + {"endpoint": "u", "events": ["taskCreated"]}, + "POST", + ) + self.assertEqual(data, {"id": "abc"}) + + def test_get_and_delete_send_no_body(self) -> None: + captured: List[Any] = [] + + def fake_urlopen(req: Any) -> MagicMock: + captured.append(req) + return make_urlopen_response(200, b"{}") + + with patch(SCRIPT_PATH + ".urlopen", side_effect=fake_urlopen): + self.handler.make_clickup_request(self.handler.ENDPOINTS["team"], {}, "GET") + self.handler.make_clickup_request( + self.handler.ENDPOINTS["team"], {"endpoint": "u", "events": ["x"]}, "POST" + ) + self.handler.make_clickup_request("webhook/1", {}, "DELETE") + + get_req, post_req, delete_req = captured + self.assertIsNone(get_req.data) + self.assertIsNotNone(post_req.data) + self.assertIsNone(delete_req.data) + + def test_authorization_header_is_raw_token(self) -> None: + captured: List[Any] = [] + + def fake_urlopen(req: Any) -> MagicMock: + captured.append(req) + return make_urlopen_response(200, b"{}") + + with patch(SCRIPT_PATH + ".urlopen", side_effect=fake_urlopen): + self.handler.make_clickup_request(self.handler.ENDPOINTS["team"], {}, "GET") + self.assertEqual(captured[0].headers["Authorization"], MOCK_API_KEY) + + def test_response_with_httperror(self) -> None: + err = HTTPError(MOCK_WEBHOOK_URL, 403, "Forbidden", {}, None) # type: ignore[arg-type] + with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout: + with patch(SCRIPT_PATH + ".urlopen", side_effect=err): + data = self.handler.make_clickup_request(self.handler.ENDPOINTS["team"], {}, "GET") + self.assertIsNone(data) + self.assertRegex(mock_stdout.getvalue(), r"HTTPError occurred: 403") + + def test_response_with_urlerror(self) -> None: + with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout: + with patch(SCRIPT_PATH + ".urlopen", side_effect=URLError("network down")): + data = self.handler.make_clickup_request(self.handler.ENDPOINTS["team"], {}, "GET") + self.assertIsNone(data) + self.assertRegex(mock_stdout.getvalue(), r"Network error occurred") + + @patch( + SCRIPT_PATH + ".urlopen", + return_value=make_urlopen_response(200, b"Service Unavailable"), + ) + def test_response_with_non_json_body(self, mock_urlopen: MagicMock) -> None: + with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout: + data = self.handler.make_clickup_request(self.handler.ENDPOINTS["team"], {}, "GET") + self.assertIsNone(data) + self.assertRegex(mock_stdout.getvalue(), r"Could not parse ClickUp's response") + + @patch.object( + ClickUpAPIHandler, "make_clickup_request", return_value={"id": MOCK_CREATED_WEBHOOK_ID} + ) + def test_create_webhook(self, mock_request: MagicMock) -> None: + data = self.handler.create_webhook(MOCK_WEBHOOK_URL, ["taskCreated"]) + self.assertEqual(data["id"], MOCK_CREATED_WEBHOOK_ID) + + @patch("sys.stdout", new_callable=io.StringIO) + @patch.object(ClickUpAPIHandler, "make_clickup_request", return_value=None) + def test_create_webhook_with_failure( + self, mock_request: MagicMock, mock_stdout: io.StringIO + ) -> None: + with self.assertRaises(SystemExit) as cm: + self.handler.create_webhook(MOCK_WEBHOOK_URL, ["taskCreated"]) + self.assertEqual(cm.exception.code, 1) + + @patch("sys.stdout", new_callable=io.StringIO) + @patch.object(ClickUpAPIHandler, "make_clickup_request", return_value=None) + def test_get_webhooks_with_failure( + self, mock_request: MagicMock, mock_stdout: io.StringIO + ) -> None: + with self.assertRaises(SystemExit) as cm: + self.handler.get_webhooks() + self.assertEqual(cm.exception.code, 1) + + +class DeleteOldWebhooksTest(TestCase): + @patch.object(ClickUpAPIHandler, "delete_webhook") + @patch.object(ClickUpAPIHandler, "get_webhooks") + def test_matches_exact_url_only( + self, mock_get_webhooks: MagicMock, mock_delete_webhook: MagicMock + ) -> None: + target = MOCK_WEBHOOK_URL + same_host_other_bot = "https://YourZulipApp.com/api/v1/external/clickup?api_key=SOMEOTHERBOTKEY0000000000000000" + mock_get_webhooks.return_value = { + "webhooks": [ + {"id": "match", "endpoint": target}, + {"id": "other-bot", "endpoint": same_host_other_bot}, + ] + } + deleted: List[str] = [] + mock_delete_webhook.side_effect = deleted.append + + handler = ClickUpAPIHandler(MOCK_API_KEY, CLICKUP_TEAM_ID) + zulip_clickup.delete_old_webhooks(target, handler) + + self.assertEqual(deleted, ["match"]) + + +class RunTest(TestCase): + @patch("sys.stdout", new_callable=io.StringIO) + @patch(SCRIPT_PATH + ".query_for_notification_events", return_value=["taskCreated"]) + @patch(SCRIPT_PATH + ".delete_old_webhooks") + @patch.object(ClickUpAPIHandler, "create_webhook", return_value={"id": MOCK_CREATED_WEBHOOK_ID}) + def test_run_without_replace_skips_delete( + self, + mock_create: MagicMock, + mock_delete_old: MagicMock, + mock_events: MagicMock, + mock_stdout: io.StringIO, + ) -> None: + with self.assertRaises(SystemExit) as cm: + zulip_clickup.run(CLICKUP_TEAM_ID, MOCK_API_KEY, MOCK_WEBHOOK_URL, False) + self.assertEqual(cm.exception.code, 0) + mock_delete_old.assert_not_called() + mock_create.assert_called_once() + self.assertRegex(mock_stdout.getvalue(), r"SUCCESS: Completed integrating") + + @patch("sys.stdout", new_callable=io.StringIO) + @patch(SCRIPT_PATH + ".query_for_notification_events", return_value=["taskCreated"]) + @patch(SCRIPT_PATH + ".delete_old_webhooks") + @patch.object(ClickUpAPIHandler, "create_webhook", return_value={"id": MOCK_CREATED_WEBHOOK_ID}) + def test_run_with_replace_calls_delete( + self, + mock_create: MagicMock, + mock_delete_old: MagicMock, + mock_events: MagicMock, + mock_stdout: io.StringIO, + ) -> None: + with self.assertRaises(SystemExit) as cm: + zulip_clickup.run(CLICKUP_TEAM_ID, MOCK_API_KEY, MOCK_WEBHOOK_URL, True) + self.assertEqual(cm.exception.code, 0) + mock_delete_old.assert_called_once() + self.assertEqual(mock_delete_old.call_args[0][0], MOCK_WEBHOOK_URL) + mock_create.assert_called_once() diff --git a/zulip/integrations/clickup/zulip_clickup.py b/zulip/integrations/clickup/zulip_clickup.py new file mode 100755 index 000000000..ac52c8c2c --- /dev/null +++ b/zulip/integrations/clickup/zulip_clickup.py @@ -0,0 +1,243 @@ +#!/usr/bin/env python3 +# +# A ClickUp integration script for Zulip. + +import argparse +import json +import re +import sys +from typing import Any, Optional, Union +from urllib.error import HTTPError, URLError +from urllib.parse import urljoin +from urllib.request import Request, urlopen + +EVENT_CHOICES: dict[str, tuple[str, ...]] = { + "task": ("taskCreated", "taskUpdated", "taskDeleted"), + "list": ("listCreated", "listUpdated", "listDeleted"), + "folder": ("folderCreated", "folderUpdated", "folderDeleted"), + "space": ("spaceCreated", "spaceUpdated", "spaceDeleted"), + "goal": ("goalCreated", "goalUpdated", "goalDeleted"), + "key result": ("keyResultCreated", "keyResultUpdated", "keyResultDeleted"), +} + + +def get_event_choices_string() -> str: + choices_string = "" + for index, key in enumerate(EVENT_CHOICES): + choices_string += f" {index + 1} = {key}\n" + return choices_string + + +class ClickUpAPIHandler: + def __init__( + self, + api_key: str, + team_id: str, + ) -> None: + self.api_key: str = api_key + self.team_id: str = team_id + self.ENDPOINTS: dict[str, str] = { + "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) + + headers: dict[str, str] = { + "Content-Type": "application/json", + "Authorization": self.api_key, + } + body = json.dumps(query).encode("utf-8") if query else None + req = Request(api_endpoint, data=body, headers=headers, method=method) # noqa: S310 + + try: + with urlopen(req) as response: # noqa: S310 + if response.status != 200: + print(f"Error : {response.status}") + sys.exit(1) + raw = response.read().decode("utf-8") + data: dict[str, Any] = json.loads(raw) + return data + except HTTPError as err: + print(f"HTTPError occurred: {err.code} {err.reason}") + return None + except URLError as err: + print(f"Network error occurred: {err.reason}") + return None + except ValueError as err: + print(f"Could not parse ClickUp's response: {err}") + return None + + 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, + } + + if (data := self.make_clickup_request(self.ENDPOINTS["team"], query, "POST")) 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/ + if (data := self.make_clickup_request(self.ENDPOINTS["team"], {}, "GET")) 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) + if self.make_clickup_request(endpoint, {}, "DELETE") is None: + print("Failed to delete webhook.") + sys.exit(1) + + +def query_for_notification_events() -> list[str]: + event_choices: str = get_event_choices_string() + print( + f""" +Please select which ClickUp event notification(s) you'd +like to receive in your Zulip app. +EVENT CODES: +{event_choices} + 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 +""" + ) + + code_to_event_dict = {i + 1: key for i, key in enumerate(EVENT_CHOICES)} + + def process_input() -> list[str]: + selected_events: list[str] = [] + input_codes_list: str = input("EVENT CODE(s): ") + + if "*" in input_codes_list: + all_events = [event for events in EVENT_CHOICES.values() for event in events] + return all_events + + user_input: list[str] = re.split(",", input_codes_list) + input_is_valid: bool = len(user_input) > 0 + + exhausted_options: list[str] = [] + for raw_code in user_input: + event_code = raw_code.strip() + if ( + event_code.strip().isdigit() + and (event := code_to_event_dict.get(int(event_code))) in EVENT_CHOICES + ): + if event_code not in exhausted_options: + events = EVENT_CHOICES[event] + selected_events += events + 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") + return process_input() + else: + return selected_events + + return process_input() + + +def delete_old_webhooks(zulip_integration_url: str, api_handler: ClickUpAPIHandler) -> None: + data: dict[str, Any] = api_handler.get_webhooks() + for webhook in data["webhooks"]: + if webhook["endpoint"] == zulip_integration_url: + 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 run(team_id: str, api_key: str, zulip_integration_url: str, replace_existing: bool) -> None: + api_handler = ClickUpAPIHandler(api_key, team_id) + events_payload: list[str] = query_for_notification_events() + if replace_existing: + delete_old_webhooks(zulip_integration_url, api_handler) + + response: dict[str, Any] = api_handler.create_webhook( + end_point=zulip_integration_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 number immediately following the base ClickUp URL. " + "For instance, the team_id for https://app.clickup.com/25567147/home " + "is 25567147." + ), + ) + parser.add_argument( + "--clickup-api-key", + required=True, + help=( + "Your personal ClickUp API token, used to register the webhook and, on " + "the Zulip side, to fetch task and entity names. Generate one in ClickUp " + "under Settings > ClickUp API > API Token." + ), + ) + parser.add_argument( + "--zulip-webhook-url", + required=True, + help="This is the URL your incoming webhook bot has generated.", + ) + parser.add_argument( + "--replace-existing", + action="store_true", + help=( + "Before registering, delete any existing ClickUp webhook that points " + "to this exact Zulip URL. Use when re-running the script to avoid " + "creating duplicate webhooks." + ), + ) + + options = parser.parse_args() + print("Running Zulip Clickup Integration...") + + run( + options.clickup_team_id, + options.clickup_api_key, + options.zulip_webhook_url, + options.replace_existing, + ) + + +if __name__ == "__main__": + main()