Add unit tests

Author: WilliamBergamin

tests/scenario_tests/test_events_assistant.py

@@ -192,11 +192,17 @@ def start_thread(say: Say, set_suggested_prompts: SetSuggestedPrompts, set_statu
             say("Hi, how can I help you today?")
             state["called"] = True
 
-        @app.event("message")
-        def handle_message(say: Say, context: BoltContext, body: dict):
-            if context.get("set_status") is not None:
-                assert say.thread_ts == context.thread_ts
+        @app.message()
+        def handle_user_message(say: Say, set_status: SetStatus, context: BoltContext):
+            assert context.channel_id == "D111"
+            assert context.thread_ts == "1726133698.626339"
+            assert say.thread_ts == context.thread_ts
+            try:
+                set_status("is typing...")
+                say("Here you are!")
                 state["called"] = True
+            except Exception as e:
+                say(f"Oops, something went wrong (error: {e}")
 
         request = BoltRequest(body=thread_started_event_body, mode="socket_mode")
         response = app.dispatch(request)

tests/scenario_tests_async/test_events_assistant.py

@@ -153,7 +153,7 @@ async def start_thread(
             await say("Hi, how can I help you today?")
             state["called"] = True
 
-        @app.event("message")
+        @app.message()
         async def handle_message(say: AsyncSay, context: AsyncBoltContext):
             if context.get("set_status") is not None:
                 assert say.thread_ts == context.thread_ts

tests/slack_bolt/middleware/attaching_agent_kwargs/__init__.py

@@ -0,0 +1,88 @@
+from slack_sdk import WebClient
+
+from slack_bolt.middleware.attaching_agent_kwargs import AttachingAgentKwargs
+from slack_bolt.request import BoltRequest
+from slack_bolt.response import BoltResponse
+from tests.scenario_tests.test_events_assistant import (
+    thread_started_event_body,
+    user_message_event_body,
+    channel_user_message_event_body,
+)
+
+
+def next():
+    return BoltResponse(status=200)
+
+
+AGENT_KWARGS = ("say", "set_status", "set_title", "set_suggested_prompts", "get_thread_context", "save_thread_context")
+
+
+class TestAttachingAgentKwargs:
+    def test_assistant_event_attaches_kwargs(self):
+        middleware = AttachingAgentKwargs()
+        req = BoltRequest(body=thread_started_event_body, mode="socket_mode")
+        req.context["client"] = WebClient(token="xoxb-test")
+        resp = BoltResponse(status=404)
+
+        resp = middleware.process(req=req, resp=resp, next=next)
+
+        assert resp.status == 200
+        for key in AGENT_KWARGS:
+            assert key in req.context, f"{key} should be set on context"
+        assert req.context["say"].thread_ts == "1726133698.626339"
+
+    def test_user_message_assistant_event_attaches_kwargs(self):
+        middleware = AttachingAgentKwargs()
+        req = BoltRequest(body=user_message_event_body, mode="socket_mode")
+        req.context["client"] = WebClient(token="xoxb-test")
+        resp = BoltResponse(status=404)
+
+        resp = middleware.process(req=req, resp=resp, next=next)
+
+        assert resp.status == 200
+        for key in AGENT_KWARGS:
+            assert key in req.context, f"{key} should be set on context"
+        assert req.context["say"].thread_ts == "1726133698.626339"
+
+    def test_non_assistant_event_does_not_attach_kwargs(self):
+        middleware = AttachingAgentKwargs()
+        req = BoltRequest(body=channel_user_message_event_body, mode="socket_mode")
+        req.context["client"] = WebClient(token="xoxb-test")
+        resp = BoltResponse(status=404)
+
+        resp = middleware.process(req=req, resp=resp, next=next)
+
+        assert resp.status == 200
+        for key in AGENT_KWARGS:
+            assert key not in req.context, f"{key} should not be set on context"
+
+    def test_non_event_body_does_not_attach_kwargs(self):
+        middleware = AttachingAgentKwargs()
+        req = BoltRequest(body="payload={}", headers={})
+        resp = BoltResponse(status=404)
+
+        resp = middleware.process(req=req, resp=resp, next=next)
+
+        assert resp.status == 200
+        for key in AGENT_KWARGS:
+            assert key not in req.context, f"{key} should not be set on context"
+
+    def test_next_always_called(self):
+        middleware = AttachingAgentKwargs()
+        call_state = {"called": False}
+
+        def tracking_next():
+            call_state["called"] = True
+            return BoltResponse(status=200)
+
+        # With assistant event
+        req = BoltRequest(body=thread_started_event_body, mode="socket_mode")
+        req.context["client"] = WebClient(token="xoxb-test")
+        middleware.process(req=req, resp=BoltResponse(status=404), next=tracking_next)
+        assert call_state["called"] is True
+
+        # With non-event body
+        call_state["called"] = False
+        req = BoltRequest(body="payload={}", headers={})
+        middleware.process(req=req, resp=BoltResponse(status=404), next=tracking_next)
+        assert call_state["called"] is True

tests/slack_bolt/middleware/attaching_agent_kwargs/test_attaching_agent_kwargs.py

@@ -0,0 +1,94 @@
+import pytest
+from slack_sdk.web.async_client import AsyncWebClient
+
+from slack_bolt.middleware.attaching_agent_kwargs.async_attaching_agent_kwargs import AsyncAttachingAgentKwargs
+from slack_bolt.request.async_request import AsyncBoltRequest
+from slack_bolt.response import BoltResponse
+from tests.scenario_tests_async.test_events_assistant import (
+    thread_started_event_body,
+    user_message_event_body,
+    channel_user_message_event_body,
+)
+
+
+async def next():
+    return BoltResponse(status=200)
+
+
+AGENT_KWARGS = ("say", "set_status", "set_title", "set_suggested_prompts", "get_thread_context", "save_thread_context")
+
+
+class TestAsyncAttachingAgentKwargs:
+    @pytest.mark.asyncio
+    async def test_assistant_event_attaches_kwargs(self):
+        middleware = AsyncAttachingAgentKwargs()
+        req = AsyncBoltRequest(body=thread_started_event_body, mode="socket_mode")
+        req.context["client"] = AsyncWebClient(token="xoxb-test")
+        resp = BoltResponse(status=404)
+
+        resp = await middleware.async_process(req=req, resp=resp, next=next)
+
+        assert resp.status == 200
+        for key in AGENT_KWARGS:
+            assert key in req.context, f"{key} should be set on context"
+        assert req.context["say"].thread_ts == "1726133698.626339"
+
+    @pytest.mark.asyncio
+    async def test_user_message_assistant_event_attaches_kwargs(self):
+        middleware = AsyncAttachingAgentKwargs()
+        req = AsyncBoltRequest(body=user_message_event_body, mode="socket_mode")
+        req.context["client"] = AsyncWebClient(token="xoxb-test")
+        resp = BoltResponse(status=404)
+
+        resp = await middleware.async_process(req=req, resp=resp, next=next)
+
+        assert resp.status == 200
+        for key in AGENT_KWARGS:
+            assert key in req.context, f"{key} should be set on context"
+        assert req.context["say"].thread_ts == "1726133698.626339"
+
+    @pytest.mark.asyncio
+    async def test_non_assistant_event_does_not_attach_kwargs(self):
+        middleware = AsyncAttachingAgentKwargs()
+        req = AsyncBoltRequest(body=channel_user_message_event_body, mode="socket_mode")
+        req.context["client"] = AsyncWebClient(token="xoxb-test")
+        resp = BoltResponse(status=404)
+
+        resp = await middleware.async_process(req=req, resp=resp, next=next)
+
+        assert resp.status == 200
+        for key in AGENT_KWARGS:
+            assert key not in req.context, f"{key} should not be set on context"
+
+    @pytest.mark.asyncio
+    async def test_non_event_body_does_not_attach_kwargs(self):
+        middleware = AsyncAttachingAgentKwargs()
+        req = AsyncBoltRequest(body="payload={}", headers={})
+        resp = BoltResponse(status=404)
+
+        resp = await middleware.async_process(req=req, resp=resp, next=next)
+
+        assert resp.status == 200
+        for key in AGENT_KWARGS:
+            assert key not in req.context, f"{key} should not be set on context"
+
+    @pytest.mark.asyncio
+    async def test_next_always_called(self):
+        middleware = AsyncAttachingAgentKwargs()
+        call_state = {"called": False}
+
+        async def tracking_next():
+            call_state["called"] = True
+            return BoltResponse(status=200)
+
+        # With assistant event
+        req = AsyncBoltRequest(body=thread_started_event_body, mode="socket_mode")
+        req.context["client"] = AsyncWebClient(token="xoxb-test")
+        await middleware.async_process(req=req, resp=BoltResponse(status=404), next=tracking_next)
+        assert call_state["called"] is True
+
+        # With non-event body
+        call_state["called"] = False
+        req = AsyncBoltRequest(body="payload={}", headers={})
+        await middleware.async_process(req=req, resp=BoltResponse(status=404), next=tracking_next)
+        assert call_state["called"] is True