Add more unit tests

Author: seratch

tests/slack_bolt/authorization/test_authorize.py

@@ -3,12 +3,13 @@
 from logging import Logger
 from typing import Optional
 
+import pytest
 from slack_sdk import WebClient
 from slack_sdk.oauth import InstallationStore
 from slack_sdk.oauth.installation_store import Bot, Installation
 
 from slack_bolt import BoltContext
-from slack_bolt.authorization.authorize import InstallationStoreAuthorize
+from slack_bolt.authorization.authorize import InstallationStoreAuthorize, Authorize
 from tests.mock_web_api_server import (
     cleanup_mock_web_api_server,
     setup_mock_web_api_server,
@@ -24,6 +25,16 @@ def setup_method(self):
     def teardown_method(self):
         cleanup_mock_web_api_server(self)
 
+    def test_root_class(self):
+        authorize = Authorize()
+        with pytest.raises(NotImplementedError):
+            authorize(
+                context=BoltContext(),
+                enterprise_id="E111",
+                team_id="T111",
+                user_id="U111",
+            )
+
     def test_installation_store_legacy(self):
         installation_store = LegacyMemoryInstallationStore()
         authorize = InstallationStoreAuthorize(

tests/slack_bolt/listener_matcher/test_builtins.py

@@ -7,6 +7,8 @@
     block_action,
     action,
     workflow_step_execute,
+    event,
+    shortcut,
 )
 
 
@@ -52,6 +54,7 @@ def test_block_action(self):
             action({"action_id": re.compile("invalid_.+")}).matches(req, resp) is False
         )
 
+        # block_id + action_id
         assert (
             action({"action_id": "valid_action_id", "block_id": "b"}).matches(req, resp)
             is True
@@ -100,6 +103,26 @@ def test_block_action(self):
             is False
         )
 
+        # with type
+        assert (
+            action({"action_id": "valid_action_id", "type": "block_actions"}).matches(
+                req, resp
+            )
+            is True
+        )
+        assert (
+            action(
+                {"callback_id": "valid_action_id", "type": "interactive_message"}
+            ).matches(req, resp)
+            is False
+        )
+        assert (
+            action(
+                {"callback_id": "valid_action_id", "type": "workflow_step_edit"}
+            ).matches(req, resp)
+            is False
+        )
+
     def test_workflow_step_execute(self):
         payload = {
             "team_id": "T111",
@@ -135,3 +158,137 @@ def test_workflow_step_execute(self):
 
         m = workflow_step_execute(re.compile("copy_.+"))
         assert m.matches(request, None) == True
+
+    def test_events(self):
+        request = BoltRequest(body=json.dumps(event_payload))
+
+        m = event("app_mention")
+        assert m.matches(request, None)
+        m = event({"type": "app_mention"})
+        assert m.matches(request, None)
+        m = event("message")
+        assert not m.matches(request, None)
+        m = event({"type": "message"})
+        assert not m.matches(request, None)
+
+        request = BoltRequest(body=f"payload={quote(json.dumps(shortcut_payload))}")
+
+        m = event("app_mention")
+        assert not m.matches(request, None)
+        m = event({"type": "app_mention"})
+        assert not m.matches(request, None)
+
+    def test_global_shortcuts(self):
+        request = BoltRequest(body=f"payload={quote(json.dumps(shortcut_payload))}")
+
+        m = shortcut("test-shortcut")
+        assert m.matches(request, None)
+        m = shortcut({"callback_id": "test-shortcut", "type": "shortcut"})
+        assert m.matches(request, None)
+
+        m = shortcut("test-shortcut!!!")
+        assert not m.matches(request, None)
+        m = shortcut({"callback_id": "test-shortcut", "type": "message_action"})
+        assert not m.matches(request, None)
+        m = shortcut({"callback_id": "test-shortcut!!!", "type": "shortcut"})
+        assert not m.matches(request, None)
+
+    def test_message_shortcuts(self):
+        request = BoltRequest(
+            body=f"payload={quote(json.dumps(message_shortcut_payload))}"
+        )
+
+        m = shortcut("test-shortcut")
+        assert m.matches(request, None)
+        m = shortcut({"callback_id": "test-shortcut", "type": "message_action"})
+        assert m.matches(request, None)
+
+        m = shortcut("test-shortcut!!!")
+        assert not m.matches(request, None)
+        m = shortcut({"callback_id": "test-shortcut", "type": "shortcut"})
+        assert not m.matches(request, None)
+        m = shortcut({"callback_id": "test-shortcut!!!", "type": "message_action"})
+        assert not m.matches(request, None)
+
+
+event_payload = {
+    "team_id": "T111",
+    "enterprise_id": "E111",
+    "api_app_id": "A111",
+    "event": {
+        "type": "app_mention",
+        "text": "<@W111> Hi there!",
+        "user": "W222",
+        "ts": "1595926230.009600",
+        "event_ts": "1595926230.009600",
+    },
+    "type": "event_callback",
+    "event_id": "Ev111",
+    "event_time": 1595926230,
+    "authorizations": [
+        {
+            "enterprise_id": "E111",
+            "team_id": "T111",
+            "user_id": "W111",
+            "is_bot": True,
+            "is_enterprise_install": True,
+        }
+    ],
+}
+
+shortcut_payload = {
+    "type": "shortcut",
+    "token": "verification_token",
+    "action_ts": "111.111",
+    "team": {
+        "id": "T111",
+        "domain": "workspace-domain",
+        "enterprise_id": "E111",
+        "enterprise_name": "Org Name",
+    },
+    "user": {"id": "W111", "username": "primary-owner", "team_id": "T111"},
+    "callback_id": "test-shortcut",
+    "trigger_id": "111.111.xxxxxx",
+}
+
+
+message_shortcut_payload = {
+    "type": "message_action",
+    "token": "verification_token",
+    "action_ts": "1583637157.207593",
+    "team": {
+        "id": "T111",
+        "domain": "test-test",
+        "enterprise_id": "E111",
+        "enterprise_name": "Org Name",
+    },
+    "user": {"id": "W111", "name": "test-test"},
+    "channel": {"id": "C111", "name": "dev"},
+    "callback_id": "test-shortcut",
+    "trigger_id": "111.222.xxx",
+    "message_ts": "1583636382.000300",
+    "message": {
+        "client_msg_id": "zzzz-111-222-xxx-yyy",
+        "type": "message",
+        "text": "<@W222> test",
+        "user": "W111",
+        "ts": "1583636382.000300",
+        "team": "T111",
+        "blocks": [
+            {
+                "type": "rich_text",
+                "block_id": "d7eJ",
+                "elements": [
+                    {
+                        "type": "rich_text_section",
+                        "elements": [
+                            {"type": "user", "user_id": "U222"},
+                            {"type": "text", "text": " test"},
+                        ],
+                    }
+                ],
+            }
+        ],
+    },
+    "response_url": "https://hooks.slack.com/app/T111/111/xxx",
+}

tests/slack_bolt/util/__init__.py

@@ -0,0 +1,60 @@
+import sys
+from typing import Set
+
+import pytest
+from slack_sdk.models import JsonObject
+
+from slack_bolt.error import BoltError
+from slack_bolt.util.utils import convert_to_dict, get_boot_message
+from tests.utils import remove_os_env_temporarily, restore_os_env
+
+
+class Data:
+    def __init__(self, name: str):
+        self.name = name
+
+
+class SerializableData(JsonObject):
+    @property
+    def attributes(self) -> Set[str]:
+        return {"name"}
+
+    def __init__(self, name: str):
+        self.name = name
+
+
+class TestUtil:
+    def setup_method(self):
+        self.old_os_env = remove_os_env_temporarily()
+
+    def teardown_method(self):
+        restore_os_env(self.old_os_env)
+
+    def test_convert_to_dict(self):
+        assert convert_to_dict({"foo": "bar"}) == {"foo": "bar"}
+        assert convert_to_dict(SerializableData("baz")) == {"name": "baz"}
+
+    def test_convert_to_dict_errors(self):
+        with pytest.raises(BoltError):
+            convert_to_dict(None)
+        with pytest.raises(BoltError):
+            convert_to_dict(123)
+        with pytest.raises(BoltError):
+            convert_to_dict("test")
+        with pytest.raises(BoltError):
+            convert_to_dict(Data("baz"))
+
+    def test_get_boot_message(self):
+        assert get_boot_message() == "⚡️ Bolt app is running!"
+        assert (
+            get_boot_message(development_server=True)
+            == "⚡️ Bolt app is running! (development server)"
+        )
+
+    def test_get_boot_message_win32(self):
+        sys_platform_backup = sys.platform
+        try:
+            sys.platform = "win32"
+            assert get_boot_message() == "Bolt app is running!"
+        finally:
+            sys.platform = sys_platform_backup

tests/slack_bolt/util/test_util.py

@@ -11,7 +11,10 @@
 )
 from slack_sdk.web.async_client import AsyncWebClient
 
-from slack_bolt.authorization.async_authorize import AsyncInstallationStoreAuthorize
+from slack_bolt.authorization.async_authorize import (
+    AsyncInstallationStoreAuthorize,
+    AsyncAuthorize,
+)
 from slack_bolt.context.async_context import AsyncBoltContext
 from tests.mock_web_api_server import (
     setup_mock_web_api_server,
@@ -38,6 +41,17 @@ def event_loop(self):
         finally:
             restore_os_env(old_os_env)
 
+    @pytest.mark.asyncio
+    async def test_root_class(self):
+        authorize = AsyncAuthorize()
+        with pytest.raises(NotImplementedError):
+            await authorize(
+                context=AsyncBoltContext(),
+                enterprise_id="T111",
+                team_id="T111",
+                user_id="U111",
+            )
+
     @pytest.mark.asyncio
     async def test_installation_store_legacy(self):
         installation_store = LegacyMemoryInstallationStore()