test_web_client_customization.py

import json
import logging
from time import time
from urllib.parse import quote

from slack_sdk import WebClient
from slack_sdk.http_retry import all_builtin_retry_handlers
from slack_sdk.signature import SignatureVerifier

from slack_bolt import BoltRequest
from slack_bolt.app import App
from tests.mock_web_api_server import (
    setup_mock_web_api_server,
    cleanup_mock_web_api_server,
    assert_auth_test_count,
)
from tests.utils import remove_os_env_temporarily, restore_os_env


class TestWebClientCustomization:
    valid_token = "xoxb-valid"
    signing_secret = "secret"
    mock_api_server_base_url = "http://localhost:8888"
    signature_verifier = SignatureVerifier(signing_secret)
    test_logger = logging.getLogger("test.logger")
    web_client = WebClient(
        token=valid_token,
        base_url=mock_api_server_base_url,
    )

    def setup_method(self):
        self.old_os_env = remove_os_env_temporarily()
        setup_mock_web_api_server(self)

    def teardown_method(self):
        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):
        return {
            "content-type": ["application/x-www-form-urlencoded"],
            "x-slack-signature": [self.generate_signature(body, timestamp)],
            "x-slack-request-timestamp": [timestamp],
        }

    def build_valid_request(self) -> BoltRequest:
        timestamp = str(int(time()))
        return BoltRequest(body=raw_body, headers=self.build_headers(timestamp, raw_body))

    def test_web_client_customization(self):
        self.web_client.retry_handlers = all_builtin_retry_handlers()
        app = App(
            client=self.web_client,
            signing_secret=self.signing_secret,
        )

        @app.action("a")
        def listener(ack, client):
            assert len(client.retry_handlers) == 2
            ack()

        request = self.build_valid_request()
        response = app.dispatch(request)
        assert response.status == 200
        assert response.body == ""
        assert_auth_test_count(self, 1)

    def test_web_client_logger_is_default_app_logger(self):
        app = App(token=self.valid_token, signing_secret=self.signing_secret, token_verification_enabled=False)
        assert app.client.logger == app.logger

    def test_web_client_logger_is_app_logger(self):
        app = App(
            token=self.valid_token,
            signing_secret=self.signing_secret,
            logger=self.test_logger,
            token_verification_enabled=False,
        )
        assert app.client.logger == app.logger
        assert app.client.logger == self.test_logger

    def test_default_web_client_uses_bolt_framework_logger(self):
        app = App(token=self.valid_token, signing_secret=self.signing_secret, token_verification_enabled=False)
        app.client.base_url = self.mock_api_server_base_url

        @app.action("a")
        def listener(ack, client: WebClient):
            assert client.logger == app.logger
            ack()

        request = self.build_valid_request()
        response = app.dispatch(request)
        assert response.status == 200
        assert response.body == ""
        assert_auth_test_count(self, 1)

    def test_default_web_client_uses_bolt_app_custom_logger(self):
        app = App(
            token=self.valid_token,
            signing_secret=self.signing_secret,
            token_verification_enabled=False,
            logger=self.test_logger,
        )
        app.client.base_url = self.mock_api_server_base_url

        assert app.client.logger == app.logger

        @app.action("a")
        def listener(ack, client: WebClient):
            assert client.logger == app.logger
            assert client.logger == self.test_logger
            ack()

        request = self.build_valid_request()
        response = app.dispatch(request)
        assert response.status == 200
        assert response.body == ""
        assert_auth_test_count(self, 1)

    def test_custom_web_client_logger_is_used_instead_of_bolt_app_logger(self):
        web_client = WebClient(token=self.valid_token, base_url=self.mock_api_server_base_url, logger=self.test_logger)
        app = App(
            client=web_client,
            signing_secret=self.signing_secret,
        )

        @app.action("a")
        def listener(ack, client: WebClient):
            assert client.logger == self.test_logger
            assert app.logger != self.test_logger
            ack()

        request = self.build_valid_request()
        response = app.dispatch(request)
        assert response.status == 200
        assert response.body == ""
        assert_auth_test_count(self, 1)


block_actions_body = {
    "type": "block_actions",
    "user": {
        "id": "W99999",
        "username": "primary-owner",
        "name": "primary-owner",
        "team_id": "T111",
    },
    "api_app_id": "A111",
    "token": "verification_token",
    "container": {
        "type": "message",
        "message_ts": "111.222",
        "channel_id": "C111",
        "is_ephemeral": True,
    },
    "trigger_id": "111.222.valid",
    "team": {
        "id": "T111",
        "domain": "workspace-domain",
        "enterprise_id": "E111",
        "enterprise_name": "Sandbox Org",
    },
    "channel": {"id": "C111", "name": "test-channel"},
    "response_url": "https://hooks.slack.com/actions/T111/111/random-value",
    "actions": [
        {
            "action_id": "a",
            "block_id": "b",
            "text": {"type": "plain_text", "text": "Button", "emoji": True},
            "value": "click_me_123",
            "type": "button",
            "action_ts": "1596530385.194939",
        }
    ],
}

raw_body = f"payload={quote(json.dumps(block_actions_body))}"