Add tests for Pyramid framework

seratch · seratch · commit 8566095fab8d · 2020-09-12T18:00:19.000+09:00
diff --git a/slack_bolt/adapter/pyramid/handler.py b/slack_bolt/adapter/pyramid/handler.py
@@ -8,10 +8,14 @@
 
 
 def to_bolt_request(request: Request) -> BoltRequest:
+    body: str = ""
+    if request.body is not None:
+        if isinstance(request.body, bytes):
+            body = request.body.decode("utf-8")
+        else:
+            body = request.body
     bolt_req = BoltRequest(
-        body=request.body.decode("utf-8"),
-        query=request.query_string,
-        headers=request.headers,
+        body=body, query=request.query_string, headers=request.headers,
     )
     return bolt_req
 
diff --git a/tests/adapter_tests/test_pyramid.py b/tests/adapter_tests/test_pyramid.py
@@ -0,0 +1,178 @@
+import json
+from time import time
+from unittest import TestCase
+from urllib.parse import quote
+
+from pyramid import testing
+from pyramid.request import Request
+from pyramid.response import Response
+from slack_sdk.signature import SignatureVerifier
+from slack_sdk.web import WebClient
+
+from slack_bolt.adapter.pyramid import SlackRequestHandler
+from slack_bolt.app import App
+from slack_bolt.oauth.oauth_settings import OAuthSettings
+from tests.mock_web_api_server import (
+    setup_mock_web_api_server,
+    cleanup_mock_web_api_server,
+)
+from tests.utils import remove_os_env_temporarily, restore_os_env
+
+
+class TestPyramid(TestCase):
+    signing_secret = "secret"
+    valid_token = "xoxb-valid"
+    mock_api_server_base_url = "http://localhost:8888"
+    signature_verifier = SignatureVerifier(signing_secret)
+    web_client = WebClient(token=valid_token, base_url=mock_api_server_base_url,)
+
+    def setUp(self):
+        self.config = testing.setUp()
+        self.old_os_env = remove_os_env_temporarily()
+        setup_mock_web_api_server(self)
+
+    def tearDown(self):
+        testing.tearDown()
+        cleanup_mock_web_api_server(self)
+        restore_os_env(self.old_os_env)
+
+    def generate_signature(self, body: str, timestamp: str):
+        return self.signature_verifier.generate_signature(
+            body=body, timestamp=timestamp,
+        )
+
+    def build_headers(self, timestamp: str, body: str):
+        content_type = (
+            "application/json"
+            if body.startswith("{")
+            else "application/x-www-form-urlencoded"
+        )
+        return {
+            "content-type": [content_type],
+            "x-slack-signature": [self.generate_signature(body, timestamp)],
+            "x-slack-request-timestamp": [timestamp],
+        }
+
+    def test_events(self):
+        app = App(client=self.web_client, signing_secret=self.signing_secret,)
+
+        def event_handler():
+            pass
+
+        app.event("app_mention")(event_handler)
+
+        input = {
+            "token": "verification_token",
+            "team_id": "T111",
+            "enterprise_id": "E111",
+            "api_app_id": "A111",
+            "event": {
+                "client_msg_id": "9cbd4c5b-7ddf-4ede-b479-ad21fca66d63",
+                "type": "app_mention",
+                "text": "<@W111> Hi there!",
+                "user": "W222",
+                "ts": "1595926230.009600",
+                "team": "T111",
+                "channel": "C111",
+                "event_ts": "1595926230.009600",
+            },
+            "type": "event_callback",
+            "event_id": "Ev111",
+            "event_time": 1595926230,
+            "authed_users": ["W111"],
+        }
+        timestamp, body = str(int(time())), json.dumps(input)
+
+        request: Request = testing.DummyRequest()
+        request.path = "/slack/events"
+        request.method = "POST"
+        request.body = body.encode("utf-8")
+        request.headers = self.build_headers(timestamp, body)
+        response: Response = SlackRequestHandler(app).handle(request)
+        assert response.status_code == 200
+        assert self.mock_received_requests["/auth.test"] == 1
+
+    def test_shortcuts(self):
+        app = App(client=self.web_client, signing_secret=self.signing_secret,)
+
+        def shortcut_handler(ack):
+            ack()
+
+        app.shortcut("test-shortcut")(shortcut_handler)
+
+        input = {
+            "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",
+        }
+
+        timestamp, body = str(int(time())), f"payload={quote(json.dumps(input))}"
+
+        request: Request = testing.DummyRequest()
+        request.path = "/slack/events"
+        request.method = "POST"
+        request.body = body.encode("utf-8")
+        request.headers = self.build_headers(timestamp, body)
+        response: Response = SlackRequestHandler(app).handle(request)
+        assert response.status_code == 200
+        assert self.mock_received_requests["/auth.test"] == 1
+
+    def test_commands(self):
+        app = App(client=self.web_client, signing_secret=self.signing_secret,)
+
+        def command_handler(ack):
+            ack()
+
+        app.command("/hello-world")(command_handler)
+
+        input = (
+            "token=verification_token"
+            "&team_id=T111"
+            "&team_domain=test-domain"
+            "&channel_id=C111"
+            "&channel_name=random"
+            "&user_id=W111"
+            "&user_name=primary-owner"
+            "&command=%2Fhello-world"
+            "&text=Hi"
+            "&enterprise_id=E111"
+            "&enterprise_name=Org+Name"
+            "&response_url=https%3A%2F%2Fhooks.slack.com%2Fcommands%2FT111%2F111%2Fxxxxx"
+            "&trigger_id=111.111.xxx"
+        )
+        timestamp, body = str(int(time())), input
+
+        request: Request = testing.DummyRequest()
+        request.path = "/slack/events"
+        request.method = "POST"
+        request.body = body.encode("utf-8")
+        request.headers = self.build_headers(timestamp, body)
+        response: Response = SlackRequestHandler(app).handle(request)
+        assert response.status_code == 200
+        assert self.mock_received_requests["/auth.test"] == 1
+
+    def test_oauth(self):
+        app = App(
+            client=self.web_client,
+            signing_secret=self.signing_secret,
+            oauth_settings=OAuthSettings(
+                client_id="111.111",
+                client_secret="xxx",
+                scopes=["chat:write", "commands"],
+            ),
+        )
+
+        request: Request = testing.DummyRequest()
+        request.path = "/slack/install"
+        request.method = "GET"
+        response: Response = SlackRequestHandler(app).handle(request)
+        assert response.status_code == 302
