Add slack_bolt.testing for unit testing Bolt apps

docs/english/_sidebar.json

@@ -77,6 +77,7 @@
 		]
 	},
 	"tools/bolt-python/concepts/lazy-listeners",
+	"tools/bolt-python/concepts/testing",
 	{
 		"type": "category",
 		"label": "Adaptors",

docs/english/concepts/testing.md

@@ -0,0 +1,94 @@
+# Testing your app
+
+`slack_bolt.testing` lets you unit test listeners without connecting to Slack. It has two parts:
+
+* `MockSlackServer`: a small local HTTP server that stands in for the Slack Web API and for `response_url`. It records every API call your app makes and every message posted with `respond()`, and answers with canned responses. `auth.test` is answered with a bot installation by default, and every other method returns `{"ok": true}` unless you set a response.
+* `BoltTestClient` (and `AsyncBoltTestClient` for `AsyncApp`): builds signed requests for events, messages, slash commands, block actions, shortcuts and view submissions, sends them through your app, and returns what happened.
+
+Requests go through the app's full middleware chain, including request verification, authorization and listener matching. The test client runs listeners (and lazy listeners) to completion before returning, so you can assert on their side effects right away, whatever `process_before_response` is set to.
+
+```python
+import pytest
+from slack_bolt import App
+from slack_bolt.testing import BoltTestClient, MockSlackServer
+
+
+@pytest.fixture
+def server():
+    with MockSlackServer() as server:
+        yield server
+
+
+def create_app(client) -> App:
+    app = App(client=client, signing_secret="test-secret")
+
+    @app.command("/ticket")
+    def open_ticket(ack, command, respond, client):
+        ack(f"Creating a ticket for: {command['text']}")
+        client.chat_postMessage(channel="C_SUPPORT", text=command["text"])
+        respond("Your ticket was created")
+
+    return app
+
+
+def test_ticket_command(server):
+    app = create_app(server.web_client())
+    result = BoltTestClient(app, server).slash_command("/ticket", "printer is on fire")
+
+    assert result.status == 200
+    assert result.body == "Creating a ticket for: printer is on fire"
+    assert result.calls_to("chat.postMessage")[0].args["channel"] == "C_SUPPORT"
+    assert result.responds == [{"text": "Your ticket was created"}]
+```
+
+## Canned API responses
+
+```python
+server.set_response("users.info", {"ok": True, "user": {"id": "U123", "real_name": "Ada"}})
+server.set_response("chat.postMessage", {"ok": False, "error": "channel_not_found"})
+```
+
+An `"ok": false` response raises `SlackApiError` in your listener, the same as the real API.
+
+## Other request types
+
+```python
+tester = BoltTestClient(app, server)
+tester.message("hello", user="U123", channel="C123")
+tester.event({"type": "app_mention", "channel": "C123", "ts": "1.1", "text": "<@U0> hi"})
+tester.block_action("approve_button", value="request-42")
+tester.shortcut("open_form")
+tester.message_shortcut("save_message", message={"type": "message", "text": "hi", "ts": "1.2"})
+tester.view_submission("form", {"title_block": {"title": {"type": "plain_text_input", "value": "Hi"}}})
+```
+
+Each method returns a `DispatchResult` with `status`, `body`, `json` (the body parsed as JSON, for example `response_action` errors from a view submission), `api_calls`, `calls_to(method)` and `responds`. The payload builders are also available on their own in `slack_bolt.testing.payloads` if you want to customize a payload and send it with `tester.dispatch(payload)`.
+
+## Apps created at import time
+
+If your app is created at module level with a token from the environment, you can still test it: `BoltTestClient` points the app's Web API client at the mock server. Disable token verification so creating the app doesn't call Slack:
+
+```python
+app = App(token=os.environ["SLACK_BOT_TOKEN"], signing_secret=os.environ["SLACK_SIGNING_SECRET"],
+          token_verification_enabled=False)
+```
+
+## AsyncApp
+
+```python
+from slack_sdk.web.async_client import AsyncWebClient
+from slack_bolt.async_app import AsyncApp
+from slack_bolt.testing.async_client import AsyncBoltTestClient
+
+
+async def test_async_command(server):
+    app = AsyncApp(client=AsyncWebClient(token="xoxb-test", base_url=server.base_url), signing_secret="test-secret")
+
+    @app.command("/hello")
+    async def hello(ack, respond):
+        await ack()
+        await respond("Hi!")
+
+    result = await AsyncBoltTestClient(app, server).slash_command("/hello")
+    assert result.responds == [{"text": "Hi!"}]
+```

slack_bolt/testing/__init__.py

@@ -0,0 +1,37 @@
+"""Utilities for testing Bolt apps without connecting to Slack.
+
+`MockSlackServer` stands in for the Slack Web API and `response_url` endpoints and records
+what your app sends to them. `BoltTestClient` sends signed requests (events, messages, slash
+commands, actions, shortcuts and view submissions) to your app and returns what it did.
+
+    from slack_bolt import App
+    from slack_bolt.testing import BoltTestClient, MockSlackServer
+
+    def test_hello_command():
+        with MockSlackServer() as server:
+            app = App(client=server.web_client(), signing_secret="test-secret")
+
+            @app.command("/hello")
+            def hello(ack, command, respond):
+                ack()
+                respond(f"Hi <@{command['user_id']}>!")
+
+            result = BoltTestClient(app, server).slash_command("/hello")
+            assert result.status == 200
+            assert result.responds[0]["text"] == "Hi <@U0000000000>!"
+
+For `AsyncApp`, use `slack_bolt.testing.async_client.AsyncBoltTestClient`.
+"""
+
+from . import payloads
+from .mock_slack_server import APICall, MockSlackServer
+from .client import BoltTestClient, DispatchResult, InlineExecutor
+
+__all__ = [
+    "APICall",
+    "BoltTestClient",
+    "DispatchResult",
+    "InlineExecutor",
+    "MockSlackServer",
+    "payloads",
+]

slack_bolt/testing/async_client.py

@@ -0,0 +1,88 @@
+import asyncio
+from typing import Any, Dict, Optional, Sequence
+
+from slack_bolt.app.async_app import AsyncApp
+from slack_bolt.request.async_request import AsyncBoltRequest
+
+from . import payloads
+from .mock_slack_server import MockSlackServer
+from .client import DispatchResult, _BaseTestClient
+
+
+class AsyncBoltTestClient(_BaseTestClient):
+    """Sends signed requests to a Bolt `AsyncApp` in tests and reports what the app did.
+
+    This is the `AsyncApp` counterpart of `BoltTestClient`. After the app acknowledges a request,
+    the client waits for the listener and lazy listener tasks it started, so all of their side
+    effects are done when a call returns.
+    """
+
+    def __init__(self, app: AsyncApp, server: MockSlackServer, *, timeout: float = 10.0):
+        self.app = app
+        self.server = server
+        self.signing_secret = app._signing_secret
+        self.timeout = timeout
+        app.client.base_url = server.base_url
+
+    async def dispatch(self, payload: Dict[str, Any], *, headers: Optional[Dict[str, str]] = None) -> DispatchResult:
+        """Dispatches a payload (e.g. built with `slack_bolt.testing.payloads`) to the app."""
+        body, request_headers = self._prepare(payload, headers)
+        calls_before = len(self.server.calls)
+        responds_before = len(self.server.responds)
+        existing = asyncio.all_tasks()
+        response = await self.app.async_dispatch(AsyncBoltRequest(body=body, headers=request_headers))
+        await self._wait_for_started_tasks(existing)
+        return DispatchResult(
+            response=response,
+            api_calls=self.server.calls[calls_before:],
+            responds=self.server.responds[responds_before:],
+        )
+
+    async def _wait_for_started_tasks(self, existing) -> None:
+        loop = asyncio.get_event_loop()
+        deadline = loop.time() + self.timeout
+        current = asyncio.current_task()
+        while True:
+            pending = [t for t in asyncio.all_tasks() if t not in existing and t is not current and not t.done()]
+            if not pending:
+                return
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                raise asyncio.TimeoutError(f"Listener tasks did not finish within {self.timeout} seconds")
+            await asyncio.wait(pending, timeout=remaining)
+
+    async def event(self, event: Dict[str, Any], **kwargs: Any) -> DispatchResult:
+        """Sends an Events API request with the given event, e.g. `{"type": "app_mention", ...}`."""
+        return await self.dispatch(payloads.event(event, **kwargs))
+
+    async def message(self, text: str, **kwargs: Any) -> DispatchResult:
+        """Sends a `message` event."""
+        return await self.dispatch(payloads.message(text, **kwargs))
+
+    async def slash_command(self, command: str, text: str = "", **kwargs: Any) -> DispatchResult:
+        """Sends a slash command. `respond()` calls are recorded in the result."""
+        return await self.dispatch(payloads.slash_command(command, text, **self._with_response_url(kwargs)))
+
+    async def block_action(self, action_id: str, **kwargs: Any) -> DispatchResult:
+        """Sends a `block_actions` request. `respond()` calls are recorded in the result."""
+        return await self.dispatch(payloads.block_action(action_id, **self._with_response_url(kwargs)))
+
+    async def shortcut(self, callback_id: str, **kwargs: Any) -> DispatchResult:
+        """Sends a global shortcut."""
+        return await self.dispatch(payloads.shortcut(callback_id, **kwargs))
+
+    async def message_shortcut(self, callback_id: str, **kwargs: Any) -> DispatchResult:
+        """Sends a message shortcut. `respond()` calls are recorded in the result."""
+        return await self.dispatch(payloads.message_shortcut(callback_id, **self._with_response_url(kwargs)))
+
+    async def view_submission(
+        self,
+        callback_id: str,
+        values: Optional[Dict[str, Dict[str, Any]]] = None,
+        **kwargs: Any,
+    ) -> DispatchResult:
+        """Sends a `view_submission` request for a modal."""
+        return await self.dispatch(payloads.view_submission(callback_id, values, **kwargs))
+
+
+__all__: Sequence[str] = ["AsyncBoltTestClient"]

slack_bolt/testing/client.py

@@ -0,0 +1,164 @@
+import json
+from concurrent.futures import Executor, Future
+from typing import Any, Callable, Dict, List, Optional, Sequence
+
+from slack_bolt.app import App
+from slack_bolt.request import BoltRequest
+from slack_bolt.response import BoltResponse
+
+from . import payloads
+from .mock_slack_server import APICall, MockSlackServer
+
+
+class InlineExecutor(Executor):
+    """An executor that runs each submitted function immediately in the calling thread."""
+
+    def submit(self, fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Future:  # type: ignore[override]
+        future: Future = Future()
+        try:
+            future.set_result(fn(*args, **kwargs))
+        except BaseException as e:
+            future.set_exception(e)
+        return future
+
+
+class DispatchResult:
+    """The outcome of dispatching one request to a Bolt app in a test."""
+
+    response: Optional[BoltResponse]
+    """The HTTP response the app returned to Slack, or None if no listener acknowledged the request."""
+    api_calls: List[APICall]
+    """The Web API calls the app made while handling this request."""
+    responds: List[Dict[str, Any]]
+    """The messages the app posted to the request's `response_url` (e.g. with `respond()`)."""
+
+    def __init__(
+        self,
+        *,
+        response: Optional[BoltResponse],
+        api_calls: List[APICall],
+        responds: List[Dict[str, Any]],
+    ):
+        self.response = response
+        self.api_calls = api_calls
+        self.responds = responds
+
+    @property
+    def status(self) -> Optional[int]:
+        """The HTTP status of the response, or None if no listener acknowledged the request."""
+        return self.response.status if self.response is not None else None
+
+    @property
+    def body(self) -> str:
+        """The raw response body, e.g. the text passed to `ack()`."""
+        return self.response.body if self.response is not None else ""
+
+    @property
+    def json(self) -> Optional[Dict[str, Any]]:
+        """The response body parsed as JSON, or None if it is empty or not JSON."""
+        try:
+            parsed = json.loads(self.body)
+        except ValueError:
+            return None
+        return parsed if isinstance(parsed, dict) else None
+
+    def calls_to(self, method: str) -> List[APICall]:
+        """The Web API calls made for the given method while handling this request."""
+        return [c for c in self.api_calls if c.method == method]
+
+    def __repr__(self) -> str:
+        return f"DispatchResult(status={self.status!r}, body={self.body!r}, api_calls={self.api_calls!r})"
+
+
+class _BaseTestClient:
+    server: MockSlackServer
+    signing_secret: str
+
+    def _prepare(self, payload: Dict[str, Any], headers: Optional[Dict[str, str]]):
+        body = payloads.to_http_body(payload)
+        request_headers = payloads.signed_headers(body, self.signing_secret)
+        request_headers.update(headers or {})
+        return body, request_headers
+
+    def _with_response_url(self, kwargs: Dict[str, Any], key: str = "response_url") -> Dict[str, Any]:
+        kwargs.setdefault(key, self.server.response_url())
+        return kwargs
+
+
+class BoltTestClient(_BaseTestClient):
+    """Sends signed requests to a Bolt `App` in tests and reports what the app did.
+
+        server = MockSlackServer().start()
+        app = App(client=server.web_client("xoxb-test"), signing_secret="secret")
+
+        @app.command("/hello")
+        def hello(ack, respond):
+            ack()
+            respond("Hi!")
+
+        result = BoltTestClient(app, server).slash_command("/hello")
+        assert result.status == 200
+        assert result.responds == [{"text": "Hi!"}]
+
+    Requests run through the app's full middleware chain, including request verification
+    and authorization. The test client points the app's Web API client at `server`, and runs
+    listeners and lazy listeners in the calling thread, so all of their side effects are done
+    when a call returns, even when `process_before_response` is False.
+    """
+
+    def __init__(self, app: App, server: MockSlackServer):
+        self.app = app
+        self.server = server
+        self.signing_secret = app._signing_secret
+        app.client.base_url = server.base_url
+        executor = InlineExecutor()
+        app.listener_runner.listener_executor = executor
+        app.listener_runner.lazy_listener_runner.executor = executor  # type: ignore[attr-defined]
+
+    def dispatch(self, payload: Dict[str, Any], *, headers: Optional[Dict[str, str]] = None) -> DispatchResult:
+        """Dispatches a payload (e.g. built with `slack_bolt.testing.payloads`) to the app."""
+        body, request_headers = self._prepare(payload, headers)
+        calls_before = len(self.server.calls)
+        responds_before = len(self.server.responds)
+        response = self.app.dispatch(BoltRequest(body=body, headers=request_headers))
+        return DispatchResult(
+            response=response,
+            api_calls=self.server.calls[calls_before:],
+            responds=self.server.responds[responds_before:],
+        )
+
+    def event(self, event: Dict[str, Any], **kwargs: Any) -> DispatchResult:
+        """Sends an Events API request with the given event, e.g. `{"type": "app_mention", ...}`."""
+        return self.dispatch(payloads.event(event, **kwargs))
+
+    def message(self, text: str, **kwargs: Any) -> DispatchResult:
+        """Sends a `message` event."""
+        return self.dispatch(payloads.message(text, **kwargs))
+
+    def slash_command(self, command: str, text: str = "", **kwargs: Any) -> DispatchResult:
+        """Sends a slash command. `respond()` calls are recorded in the result."""
+        return self.dispatch(payloads.slash_command(command, text, **self._with_response_url(kwargs)))
+
+    def block_action(self, action_id: str, **kwargs: Any) -> DispatchResult:
+        """Sends a `block_actions` request. `respond()` calls are recorded in the result."""
+        return self.dispatch(payloads.block_action(action_id, **self._with_response_url(kwargs)))
+
+    def shortcut(self, callback_id: str, **kwargs: Any) -> DispatchResult:
+        """Sends a global shortcut."""
+        return self.dispatch(payloads.shortcut(callback_id, **kwargs))
+
+    def message_shortcut(self, callback_id: str, **kwargs: Any) -> DispatchResult:
+        """Sends a message shortcut. `respond()` calls are recorded in the result."""
+        return self.dispatch(payloads.message_shortcut(callback_id, **self._with_response_url(kwargs)))
+
+    def view_submission(
+        self,
+        callback_id: str,
+        values: Optional[Dict[str, Dict[str, Any]]] = None,
+        **kwargs: Any,
+    ) -> DispatchResult:
+        """Sends a `view_submission` request for a modal."""
+        return self.dispatch(payloads.view_submission(callback_id, values, **kwargs))
+
+
+__all__: Sequence[str] = ["BoltTestClient", "DispatchResult", "InlineExecutor"]