From a5ecd917b0dbf2b75376dd590d96c4633d63725f Mon Sep 17 00:00:00 2001 From: Mohith26 <109003762+Mohith26@users.noreply.github.com> Date: Wed, 4 Nov 2026 09:34:25 -0500 Subject: [PATCH 1/2] Add slack_bolt.testing for unit testing Bolt apps (#380) --- docs/english/_sidebar.json | 1 + docs/english/concepts/testing.md | 94 ++++++ slack_bolt/testing/__init__.py | 37 +++ slack_bolt/testing/async_client.py | 88 ++++++ slack_bolt/testing/client.py | 164 +++++++++++ slack_bolt/testing/mock_slack_server.py | 238 +++++++++++++++ slack_bolt/testing/payloads.py | 272 ++++++++++++++++++ tests/slack_bolt/testing/__init__.py | 0 tests/slack_bolt/testing/test_testing.py | 200 +++++++++++++ tests/slack_bolt_async/testing/__init__.py | 0 .../testing/test_async_testing.py | 86 ++++++ 11 files changed, 1180 insertions(+) create mode 100644 docs/english/concepts/testing.md create mode 100644 slack_bolt/testing/__init__.py create mode 100644 slack_bolt/testing/async_client.py create mode 100644 slack_bolt/testing/client.py create mode 100644 slack_bolt/testing/mock_slack_server.py create mode 100644 slack_bolt/testing/payloads.py create mode 100644 tests/slack_bolt/testing/__init__.py create mode 100644 tests/slack_bolt/testing/test_testing.py create mode 100644 tests/slack_bolt_async/testing/__init__.py create mode 100644 tests/slack_bolt_async/testing/test_async_testing.py diff --git a/docs/english/_sidebar.json b/docs/english/_sidebar.json index 22a8ba69d..fc7f02c04 100644 --- a/docs/english/_sidebar.json +++ b/docs/english/_sidebar.json @@ -77,6 +77,7 @@ ] }, "tools/bolt-python/concepts/lazy-listeners", + "tools/bolt-python/concepts/testing", { "type": "category", "label": "Adaptors", diff --git a/docs/english/concepts/testing.md b/docs/english/concepts/testing.md new file mode 100644 index 000000000..a1b5fb5bf --- /dev/null +++ b/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!"}] +``` diff --git a/slack_bolt/testing/__init__.py b/slack_bolt/testing/__init__.py new file mode 100644 index 000000000..68f7844f0 --- /dev/null +++ b/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", +] diff --git a/slack_bolt/testing/async_client.py b/slack_bolt/testing/async_client.py new file mode 100644 index 000000000..12cd150de --- /dev/null +++ b/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"] diff --git a/slack_bolt/testing/client.py b/slack_bolt/testing/client.py new file mode 100644 index 000000000..639cc02a5 --- /dev/null +++ b/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"] diff --git a/slack_bolt/testing/mock_slack_server.py b/slack_bolt/testing/mock_slack_server.py new file mode 100644 index 000000000..2e87a84dd --- /dev/null +++ b/slack_bolt/testing/mock_slack_server.py @@ -0,0 +1,238 @@ +import json +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Dict, List, Optional +from urllib.parse import parse_qsl + +from slack_sdk import WebClient + +DEFAULT_TEAM_ID = "T0000000000" +DEFAULT_USER_ID = "U0000000000" +DEFAULT_BOT_USER_ID = "U00000000BT" +DEFAULT_BOT_ID = "B0000000000" +DEFAULT_APP_ID = "A0000000000" + + +class APICall: + """A Web API call received by `MockSlackServer`.""" + + method: str + """The API method name, e.g. `chat.postMessage`.""" + args: Dict[str, Any] + """The arguments sent with the call, parsed from the JSON or form body.""" + headers: Dict[str, str] + """The request headers.""" + + def __init__(self, *, method: str, args: Dict[str, Any], headers: Dict[str, str]): + self.method = method + self.args = args + self.headers = headers + + @property + def token(self) -> Optional[str]: + """The token sent in the Authorization header, if any.""" + auth = self.headers.get("authorization") or "" + return auth[len("Bearer ") :] if auth.startswith("Bearer ") else None + + def __repr__(self) -> str: + return f"APICall(method={self.method!r}, args={self.args!r})" + + +class MockSlackServer: + """An in-process stand-in for the Slack Web API and `response_url` endpoints, for tests. + + It records every Web API call and every message posted to a `response_url`, and answers + them with canned responses. `auth.test` is answered with a bot installation by default, + and any other method returns `{"ok": true}` unless a response is set with `set_response()`. + + server = MockSlackServer() + server.start() + app = App(client=server.web_client("xoxb-test"), signing_secret="secret") + ... + server.stop() + + It can also be used as a context manager, which starts and stops the server. + """ + + def __init__( + self, + *, + team_id: str = DEFAULT_TEAM_ID, + bot_user_id: str = DEFAULT_BOT_USER_ID, + bot_id: str = DEFAULT_BOT_ID, + responses: Optional[Dict[str, Dict[str, Any]]] = None, + ): + self.team_id = team_id + self.bot_user_id = bot_user_id + self.bot_id = bot_id + self._responses: Dict[str, Dict[str, Any]] = dict(responses or {}) + self._calls: List[APICall] = [] + self._responds: List[Dict[str, Any]] = [] + self._lock = threading.Lock() + self._server: Optional[ThreadingHTTPServer] = None + self._thread: Optional[threading.Thread] = None + + # ------------------------- + # lifecycle + + def start(self) -> "MockSlackServer": + if self._server is not None: + return self + handler = _build_handler(self) + self._server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + self._server.daemon_threads = True + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + return self + + def stop(self) -> None: + if self._server is not None: + self._server.shutdown() + self._server.server_close() + self._server = None + if self._thread is not None: + self._thread.join() + self._thread = None + + def __enter__(self) -> "MockSlackServer": + return self.start() + + def __exit__(self, *args) -> None: + self.stop() + + # ------------------------- + # urls and clients + + @property + def url(self) -> str: + if self._server is None: + raise RuntimeError("MockSlackServer is not running. Call start() first.") + port = self._server.server_address[1] + return f"http://127.0.0.1:{port}" + + @property + def base_url(self) -> str: + """The Web API base URL to pass to `WebClient(base_url=...)`.""" + return f"{self.url}/api/" + + def response_url(self, path: str = "T0000000000/1/test") -> str: + """A `response_url` that records the messages posted to it.""" + return f"{self.url}/response_url/{path}" + + def web_client(self, token: Optional[str] = "xoxb-test-token") -> WebClient: + """A `WebClient` that sends its requests to this server.""" + return WebClient(token=token, base_url=self.base_url) + + # ------------------------- + # recorded requests + + @property + def calls(self) -> List[APICall]: + """All Web API calls received, in order.""" + with self._lock: + return list(self._calls) + + def calls_to(self, method: str) -> List[APICall]: + """The Web API calls received for the given method, e.g. `chat.postMessage`.""" + return [c for c in self.calls if c.method == method] + + @property + def responds(self) -> List[Dict[str, Any]]: + """The message bodies posted to `response_url`s (e.g. by `respond()`), in order.""" + with self._lock: + return list(self._responds) + + def set_response(self, method: str, body: Dict[str, Any]) -> None: + """Sets the response body returned for the given Web API method.""" + with self._lock: + self._responses[method] = body + + def reset(self) -> None: + """Clears the recorded calls and responds. Canned responses are kept.""" + with self._lock: + self._calls.clear() + self._responds.clear() + + # ------------------------- + # internals + + def _response_for(self, method: str) -> Dict[str, Any]: + with self._lock: + if method in self._responses: + return self._responses[method] + if method == "auth.test": + return { + "ok": True, + "url": "https://test.slack.com/", + "team": "Test Workspace", + "user": "bot", + "team_id": self.team_id, + "user_id": self.bot_user_id, + "bot_id": self.bot_id, + "is_enterprise_install": False, + } + return {"ok": True} + + def _record_call(self, call: APICall) -> None: + with self._lock: + self._calls.append(call) + + def _record_respond(self, body: Dict[str, Any]) -> None: + with self._lock: + self._responds.append(body) + + +def _parse_body(raw: bytes, content_type: str) -> Dict[str, Any]: + if not raw: + return {} + text = raw.decode("utf-8") + if "application/json" in content_type: + parsed = json.loads(text) + return parsed if isinstance(parsed, dict) else {"body": parsed} + if "application/x-www-form-urlencoded" in content_type: + return dict(parse_qsl(text, keep_blank_values=True)) + return {"body": text} + + +def _build_handler(server: MockSlackServer): + class Handler(BaseHTTPRequestHandler): + def log_message(self, format, *args): # silence the default stderr logging + pass + + def _read(self) -> Dict[str, Any]: + length = int(self.headers.get("Content-Length") or 0) + raw = self.rfile.read(length) if length else b"" + args = _parse_body(raw, self.headers.get("Content-Type") or "") + query = self.path.split("?", 1)[1] if "?" in self.path else "" + if query: + args = {**dict(parse_qsl(query, keep_blank_values=True)), **args} + return args + + def _send(self, status: int, body: Dict[str, Any]) -> None: + data = json.dumps(body).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json;charset=utf-8") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def _handle(self) -> None: + path = self.path.split("?", 1)[0] + if path.startswith("/api/"): + method = path[len("/api/") :] + headers = {k.lower(): v for k, v in self.headers.items()} + server._record_call(APICall(method=method, args=self._read(), headers=headers)) + self._send(200, server._response_for(method)) + elif path.startswith("/response_url/"): + server._record_respond(self._read()) + self._send(200, {"ok": True}) + else: + self._send(404, {"ok": False, "error": "not_found"}) + + def do_POST(self): + self._handle() + + def do_GET(self): + self._handle() + + return Handler diff --git a/slack_bolt/testing/payloads.py b/slack_bolt/testing/payloads.py new file mode 100644 index 000000000..121d91872 --- /dev/null +++ b/slack_bolt/testing/payloads.py @@ -0,0 +1,272 @@ +"""Builders for the request payloads Slack sends to Bolt apps, for use in tests. + +Each function returns the payload as a dict. Use them with `BoltTestClient.dispatch()`, +or with `build_request()` to get a signed `BoltRequest`. +""" + +import json +import time +import uuid +from typing import Any, Dict, List, Optional +from urllib.parse import urlencode + +from slack_sdk.signature import SignatureVerifier + +from .mock_slack_server import DEFAULT_APP_ID, DEFAULT_BOT_USER_ID, DEFAULT_TEAM_ID, DEFAULT_USER_ID + +DEFAULT_CHANNEL_ID = "C0000000000" + + +def _ts() -> str: + return f"{time.time():.6f}" + + +def event( + event: Dict[str, Any], + *, + team_id: str = DEFAULT_TEAM_ID, + api_app_id: str = DEFAULT_APP_ID, + authorized_user_id: str = DEFAULT_BOT_USER_ID, + event_id: Optional[str] = None, +) -> Dict[str, Any]: + """An Events API `event_callback` payload wrapping the given event.""" + event = {"event_ts": _ts(), **event} + return { + "token": "verification-token", + "team_id": team_id, + "api_app_id": api_app_id, + "event": event, + "type": "event_callback", + "event_id": event_id or f"Ev{uuid.uuid4().hex[:10].upper()}", + "event_time": int(time.time()), + "authorizations": [ + { + "enterprise_id": None, + "team_id": team_id, + "user_id": authorized_user_id, + "is_bot": True, + "is_enterprise_install": False, + } + ], + "is_ext_shared_channel": False, + } + + +def message( + text: str, + *, + user: str = DEFAULT_USER_ID, + channel: str = DEFAULT_CHANNEL_ID, + channel_type: str = "channel", + ts: Optional[str] = None, + thread_ts: Optional[str] = None, + team_id: str = DEFAULT_TEAM_ID, + **event_fields: Any, +) -> Dict[str, Any]: + """An Events API payload for a `message` event posted by a user.""" + msg: Dict[str, Any] = { + "type": "message", + "text": text, + "user": user, + "channel": channel, + "channel_type": channel_type, + "ts": ts or _ts(), + "team": team_id, + } + if thread_ts is not None: + msg["thread_ts"] = thread_ts + msg.update(event_fields) + return event(msg, team_id=team_id) + + +def slash_command( + command: str, + text: str = "", + *, + user_id: str = DEFAULT_USER_ID, + user_name: str = "test-user", + channel_id: str = DEFAULT_CHANNEL_ID, + channel_name: str = "general", + team_id: str = DEFAULT_TEAM_ID, + api_app_id: str = DEFAULT_APP_ID, + response_url: str = "https://hooks.slack.com/commands/T0000000000/1/test", + trigger_id: str = "111.222.test", +) -> Dict[str, Any]: + """A slash command payload.""" + return { + "token": "verification-token", + "team_id": team_id, + "team_domain": "test", + "channel_id": channel_id, + "channel_name": channel_name, + "user_id": user_id, + "user_name": user_name, + "command": command, + "text": text, + "api_app_id": api_app_id, + "is_enterprise_install": "false", + "response_url": response_url, + "trigger_id": trigger_id, + } + + +def _interactive_base( + type: str, + *, + user_id: str, + team_id: str, + api_app_id: str, + trigger_id: str, +) -> Dict[str, Any]: + return { + "type": type, + "token": "verification-token", + "api_app_id": api_app_id, + "team": {"id": team_id, "domain": "test"}, + "user": {"id": user_id, "username": "test-user", "name": "test-user", "team_id": team_id}, + "enterprise": None, + "is_enterprise_install": False, + "trigger_id": trigger_id, + } + + +def block_action( + action_id: str, + *, + value: Optional[str] = None, + block_id: str = "block", + action_type: str = "button", + action: Optional[Dict[str, Any]] = None, + user_id: str = DEFAULT_USER_ID, + channel_id: str = DEFAULT_CHANNEL_ID, + team_id: str = DEFAULT_TEAM_ID, + api_app_id: str = DEFAULT_APP_ID, + message: Optional[Dict[str, Any]] = None, + response_url: str = "https://hooks.slack.com/actions/T0000000000/1/test", + trigger_id: str = "111.222.test", +) -> Dict[str, Any]: + """A `block_actions` payload for an interaction with a Block Kit element in a message. + + Pass `action` to send a fully custom action object instead of the default button action. + """ + body = _interactive_base("block_actions", user_id=user_id, team_id=team_id, api_app_id=api_app_id, trigger_id=trigger_id) + if action is None: + action = {"type": action_type, "action_id": action_id, "block_id": block_id, "action_ts": _ts()} + if value is not None: + action["value"] = value + body.update( + { + "container": {"type": "message", "channel_id": channel_id, "is_ephemeral": False}, + "channel": {"id": channel_id, "name": "general"}, + "message": message or {"type": "message", "text": "", "ts": _ts()}, + "response_url": response_url, + "actions": [action], + } + ) + return body + + +def shortcut( + callback_id: str, + *, + user_id: str = DEFAULT_USER_ID, + team_id: str = DEFAULT_TEAM_ID, + api_app_id: str = DEFAULT_APP_ID, + trigger_id: str = "111.222.test", +) -> Dict[str, Any]: + """A global shortcut payload.""" + body = _interactive_base("shortcut", user_id=user_id, team_id=team_id, api_app_id=api_app_id, trigger_id=trigger_id) + body.update({"callback_id": callback_id, "action_ts": _ts()}) + return body + + +def message_shortcut( + callback_id: str, + *, + message: Optional[Dict[str, Any]] = None, + user_id: str = DEFAULT_USER_ID, + channel_id: str = DEFAULT_CHANNEL_ID, + team_id: str = DEFAULT_TEAM_ID, + api_app_id: str = DEFAULT_APP_ID, + response_url: str = "https://hooks.slack.com/app/T0000000000/1/test", + trigger_id: str = "111.222.test", +) -> Dict[str, Any]: + """A message shortcut payload.""" + body = _interactive_base( + "message_action", user_id=user_id, team_id=team_id, api_app_id=api_app_id, trigger_id=trigger_id + ) + body.update( + { + "callback_id": callback_id, + "action_ts": _ts(), + "message_ts": (message or {}).get("ts", "1.0"), + "message": message or {"type": "message", "text": "", "user": user_id, "ts": "1.0"}, + "channel": {"id": channel_id, "name": "general"}, + "response_url": response_url, + } + ) + return body + + +def view_submission( + callback_id: str, + values: Optional[Dict[str, Dict[str, Any]]] = None, + *, + private_metadata: str = "", + view_id: str = "V0000000000", + user_id: str = DEFAULT_USER_ID, + team_id: str = DEFAULT_TEAM_ID, + api_app_id: str = DEFAULT_APP_ID, + response_urls: Optional[List[Dict[str, Any]]] = None, + trigger_id: str = "111.222.test", +) -> Dict[str, Any]: + """A `view_submission` payload for a modal. + + `values` uses the same shape as `view["state"]["values"]`, e.g. + `{"block_id": {"action_id": {"type": "plain_text_input", "value": "hi"}}}`. + """ + body = _interactive_base( + "view_submission", user_id=user_id, team_id=team_id, api_app_id=api_app_id, trigger_id=trigger_id + ) + body.update( + { + "view": { + "id": view_id, + "team_id": team_id, + "type": "modal", + "callback_id": callback_id, + "private_metadata": private_metadata, + "state": {"values": values or {}}, + "hash": "1.test", + "app_id": api_app_id, + "bot_id": "B0000000000", + }, + "response_urls": response_urls or [], + } + ) + return body + + +def to_http_body(payload: Dict[str, Any]) -> str: + """Encodes a payload the way Slack sends it over HTTP. + + Events API payloads are sent as JSON, slash commands as form fields, and interactivity + payloads as a form with a single `payload` field. + """ + if "command" in payload and "type" not in payload: + return urlencode(payload) + if payload.get("type") in ("event_callback", "url_verification", "app_rate_limited"): + return json.dumps(payload) + return urlencode({"payload": json.dumps(payload)}) + + +def signed_headers(body: str, signing_secret: str, timestamp: Optional[int] = None) -> Dict[str, str]: + """The headers of a request signed with the given signing secret.""" + ts = str(timestamp if timestamp is not None else int(time.time())) + signature = SignatureVerifier(signing_secret).generate_signature(timestamp=ts, body=body) + content_type = "application/json" if body.startswith("{") else "application/x-www-form-urlencoded" + return { + "content-type": content_type, + "x-slack-request-timestamp": ts, + "x-slack-signature": signature or "", + } diff --git a/tests/slack_bolt/testing/__init__.py b/tests/slack_bolt/testing/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/slack_bolt/testing/test_testing.py b/tests/slack_bolt/testing/test_testing.py new file mode 100644 index 000000000..cb56d33f1 --- /dev/null +++ b/tests/slack_bolt/testing/test_testing.py @@ -0,0 +1,200 @@ +import json +import time + +import pytest +from slack_sdk import WebClient + +from slack_bolt import App +from slack_bolt.testing import BoltTestClient, MockSlackServer, payloads + + +@pytest.fixture +def server(): + with MockSlackServer() as s: + yield s + + +def build_app(server: MockSlackServer, **kwargs) -> App: + return App(client=server.web_client("xoxb-test"), signing_secret="test-secret", **kwargs) + + +class TestMockSlackServer: + def test_records_calls_and_returns_canned_responses(self, server: MockSlackServer): + client = server.web_client("xoxb-abc") + auth = client.auth_test() + assert auth["user_id"] == server.bot_user_id + assert auth["team_id"] == server.team_id + + server.set_response("conversations.info", {"ok": True, "channel": {"id": "C1", "name": "general"}}) + assert client.conversations_info(channel="C1")["channel"]["name"] == "general" + + client.chat_postMessage(channel="C1", text="hello", blocks=[{"type": "divider"}]) + call = server.calls_to("chat.postMessage")[0] + assert call.args["channel"] == "C1" + assert call.args["text"] == "hello" + assert call.token == "xoxb-abc" + assert [c.method for c in server.calls] == ["auth.test", "conversations.info", "chat.postMessage"] + + server.reset() + assert server.calls == [] + + def test_error_responses_raise_like_slack(self, server: MockSlackServer): + from slack_sdk.errors import SlackApiError + + server.set_response("chat.postMessage", {"ok": False, "error": "channel_not_found"}) + with pytest.raises(SlackApiError) as e: + server.web_client().chat_postMessage(channel="C404", text="hi") + assert e.value.response["error"] == "channel_not_found" + + def test_requires_start(self): + with pytest.raises(RuntimeError): + MockSlackServer().base_url + + +class TestBoltTestClient: + def test_slash_command_ack_and_respond(self, server: MockSlackServer): + app = build_app(server) + + @app.command("/hello") + def hello(ack, command, respond): + ack(f"Working on {command['text']}...") + respond(text=f"Hi <@{command['user_id']}>!", response_type="in_channel") + + result = BoltTestClient(app, server).slash_command("/hello", "the report", user_id="U123") + assert result.status == 200 + assert result.body == "Working on the report..." + assert result.responds == [{"text": "Hi <@U123>!", "response_type": "in_channel"}] + + def test_message_event_and_say(self, server: MockSlackServer): + app = build_app(server) + + @app.message("hello") + def greet(message, say): + say(f"Hey there <@{message['user']}>!") + + result = BoltTestClient(app, server).message("hello bot", user="U456", channel="C789") + assert result.status == 200 + posts = result.calls_to("chat.postMessage") + assert len(posts) == 1 + assert posts[0].args["channel"] == "C789" + assert posts[0].args["text"] == "Hey there <@U456>!" + + def test_waits_for_listeners_without_process_before_response(self, server: MockSlackServer): + app = build_app(server, process_before_response=False) + + @app.event("app_mention") + def slow(event, client): + time.sleep(0.2) + client.reactions_add(channel=event["channel"], timestamp=event["ts"], name="eyes") + + result = BoltTestClient(app, server).event({"type": "app_mention", "channel": "C1", "ts": "1.1", "text": "hi"}) + assert [c.method for c in result.api_calls] == ["reactions.add"] + + def test_lazy_listeners_run_before_returning(self, server: MockSlackServer): + app = build_app(server, process_before_response=True) + + def ack_fast(ack): + ack() + + def do_work(respond): + respond("done") + + app.command("/work")(ack=ack_fast, lazy=[do_work]) + + result = BoltTestClient(app, server).slash_command("/work") + assert result.status == 200 + assert result.responds == [{"text": "done"}] + + def test_block_action_and_view_submission(self, server: MockSlackServer): + app = build_app(server) + + @app.action("approve") + def approve(ack, body, client): + ack() + client.views_open(trigger_id=body["trigger_id"], view={"type": "modal", "callback_id": "reason"}) + + @app.view("reason") + def reason(ack, view): + value = view["state"]["values"]["reason"]["input"]["value"] + if not value: + ack(response_action="errors", errors={"reason": "Required"}) + else: + ack() + + tester = BoltTestClient(app, server) + result = tester.block_action("approve", value="42") + assert result.status == 200 + assert result.calls_to("views.open")[0].args["trigger_id"] == "111.222.test" + + values = {"reason": {"input": {"type": "plain_text_input", "value": ""}}} + result = tester.view_submission("reason", values) + assert result.json == {"response_action": "errors", "errors": {"reason": "Required"}} + + values["reason"]["input"]["value"] = "looks good" + assert tester.view_submission("reason", values).body == "" + + def test_shortcuts(self, server: MockSlackServer): + app = build_app(server) + seen = [] + + @app.shortcut("global_sc") + def global_sc(ack, shortcut): + ack() + seen.append(shortcut["type"]) + + @app.shortcut("message_sc") + def message_sc(ack, shortcut, respond): + ack() + respond(f"Got {shortcut['message']['text']}") + + tester = BoltTestClient(app, server) + assert tester.shortcut("global_sc").status == 200 + result = tester.message_shortcut("message_sc", message={"type": "message", "text": "hi", "ts": "1.2"}) + assert result.responds == [{"text": "Got hi"}] + assert seen == ["shortcut"] + + def test_unhandled_and_invalid_requests(self, server: MockSlackServer): + app = build_app(server) + tester = BoltTestClient(app, server) + assert tester.slash_command("/nobody-listens").status == 404 + + tester.signing_secret = "wrong-secret" + assert tester.slash_command("/nobody-listens").status == 401 + + def test_listener_errors_return_500(self, server: MockSlackServer): + app = build_app(server) + + @app.command("/boom") + def boom(ack): + raise ValueError("boom") + + assert BoltTestClient(app, server).slash_command("/boom").status == 500 + + def test_redirects_an_app_built_with_a_default_client(self, server: MockSlackServer): + # e.g. an app created at import time with a token from the environment + app = App(token="xoxb-env", signing_secret="test-secret", token_verification_enabled=False) + + @app.event("app_home_opened") + def home(event, client): + client.views_publish(user_id=event["user"], view={"type": "home", "blocks": []}) + + result = BoltTestClient(app, server).event({"type": "app_home_opened", "user": "U1", "tab": "home"}) + call = result.calls_to("views.publish")[0] + assert call.args["user_id"] == "U1" + assert call.token == "xoxb-env" + + +class TestPayloads: + def test_http_encoding(self): + assert payloads.to_http_body(payloads.slash_command("/a", "b")).startswith("token=") + assert json.loads(payloads.to_http_body(payloads.message("hi")))["event"]["text"] == "hi" + body = payloads.to_http_body(payloads.shortcut("x")) + assert body.startswith("payload=") + + def test_signed_headers_verify(self): + from slack_sdk.signature import SignatureVerifier + + body = payloads.to_http_body(payloads.message("hi")) + headers = payloads.signed_headers(body, "secret") + assert SignatureVerifier("secret").is_valid_request(body, headers) + assert headers["content-type"] == "application/json" diff --git a/tests/slack_bolt_async/testing/__init__.py b/tests/slack_bolt_async/testing/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/slack_bolt_async/testing/test_async_testing.py b/tests/slack_bolt_async/testing/test_async_testing.py new file mode 100644 index 000000000..3648e58ea --- /dev/null +++ b/tests/slack_bolt_async/testing/test_async_testing.py @@ -0,0 +1,86 @@ +import asyncio + +import pytest + +from slack_bolt.async_app import AsyncApp +from slack_bolt.testing import MockSlackServer +from slack_bolt.testing.async_client import AsyncBoltTestClient +from slack_sdk.web.async_client import AsyncWebClient + + +@pytest.fixture +def server(): + with MockSlackServer() as s: + yield s + + +def build_app(server: MockSlackServer, **kwargs) -> AsyncApp: + client = AsyncWebClient(token="xoxb-test", base_url=server.base_url) + return AsyncApp(client=client, signing_secret="test-secret", **kwargs) + + +class TestAsyncBoltTestClient: + @pytest.mark.asyncio + async def test_slash_command_ack_and_respond(self, server: MockSlackServer): + app = build_app(server) + + @app.command("/hello") + async def hello(ack, command, respond): + await ack("On it") + await respond(f"Hi <@{command['user_id']}>!") + + result = await AsyncBoltTestClient(app, server).slash_command("/hello", user_id="U123") + assert result.status == 200 + assert result.body == "On it" + assert result.responds == [{"text": "Hi <@U123>!"}] + + @pytest.mark.asyncio + async def test_waits_for_listener_tasks(self, server: MockSlackServer): + app = build_app(server, process_before_response=False) + + @app.message("hello") + async def greet(message, say): + await asyncio.sleep(0.2) + await say(f"Hey <@{message['user']}>") + + result = await AsyncBoltTestClient(app, server).message("hello", user="U9", channel="C9") + post = result.calls_to("chat.postMessage")[0] + assert (post.args["channel"], post.args["text"]) == ("C9", "Hey <@U9>") + + @pytest.mark.asyncio + async def test_lazy_listeners(self, server: MockSlackServer): + app = build_app(server, process_before_response=True) + + async def ack_fast(ack): + await ack() + + async def do_work(respond): + await asyncio.sleep(0.1) + await respond("done") + + app.command("/work")(ack=ack_fast, lazy=[do_work]) + result = await AsyncBoltTestClient(app, server).slash_command("/work") + assert result.responds == [{"text": "done"}] + + @pytest.mark.asyncio + async def test_view_submission_errors(self, server: MockSlackServer): + app = build_app(server) + + @app.view("reason") + async def reason(ack): + await ack(response_action="errors", errors={"reason": "Required"}) + + result = await AsyncBoltTestClient(app, server).view_submission("reason") + assert result.json == {"response_action": "errors", "errors": {"reason": "Required"}} + + @pytest.mark.asyncio + async def test_timeout(self, server: MockSlackServer): + app = build_app(server) + + @app.event("app_mention") + async def forever(): + await asyncio.sleep(1) + + tester = AsyncBoltTestClient(app, server, timeout=0.2) + with pytest.raises(asyncio.TimeoutError): + await tester.event({"type": "app_mention", "channel": "C1", "ts": "1.1", "text": "hi"}) From fdd0439327325bcb4528743ed3e24e59c3c5a608 Mon Sep 17 00:00:00 2001 From: Mohith26 <109003762+Mohith26@users.noreply.github.com> Date: Wed, 4 Nov 2026 09:46:29 -0500 Subject: [PATCH 2/2] Re-run CLA check