bolt-python/tests/slack_bolt/context/test_say_stream.py at 44bcd8c7e2ea11376cc3932a7543165e896c10e8 · slackapi/bolt-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
import pytest
from slack_sdk import WebClient

from slack_bolt.context.say_stream.say_stream import SayStream
from tests.mock_web_api_server import cleanup_mock_web_api_server, setup_mock_web_api_server


class TestSayStream:
    default_chat_stream_buffer_size = WebClient.chat_stream.__kwdefaults__["buffer_size"]

    def setup_method(self):
        setup_mock_web_api_server(self)
        valid_token = "xoxb-valid"
        mock_api_server_base_url = "http://localhost:8888"
        self.web_client = WebClient(token=valid_token, base_url=mock_api_server_base_url)

    def teardown_method(self):
        cleanup_mock_web_api_server(self)

    def test_missing_channel_raises(self):
        say_stream = SayStream(client=self.web_client, channel=None, thread_ts="111.222")
        with pytest.raises(ValueError, match="channel"):
            say_stream()

    def test_missing_thread_ts_raises(self):
        say_stream = SayStream(client=self.web_client, channel="C111", thread_ts=None)
        with pytest.raises(ValueError, match="thread_ts"):
            say_stream()

    def test_default_params(self):
        say_stream = SayStream(
            client=self.web_client,
            channel="C111",
            recipient_team_id="T111",
            recipient_user_id="U111",
            thread_ts="111.222",
        )
        stream = say_stream()

        assert stream._buffer_size == self.default_chat_stream_buffer_size
        assert stream._stream_args == {
            "channel": "C111",
            "thread_ts": "111.222",
            "recipient_team_id": "T111",
            "recipient_user_id": "U111",
            "task_display_mode": None,
        }

    def test_parameter_overrides(self):
        say_stream = SayStream(
            client=self.web_client,
            channel="C111",
            recipient_team_id="T111",
            recipient_user_id="U111",
            thread_ts="111.222",
        )
        stream = say_stream(channel="C222", thread_ts="333.444", recipient_team_id="T222", recipient_user_id="U222")

        assert stream._buffer_size == self.default_chat_stream_buffer_size
        assert stream._stream_args == {
            "channel": "C222",
            "thread_ts": "333.444",
            "recipient_team_id": "T222",
            "recipient_user_id": "U222",
            "task_display_mode": None,
        }

    def test_buffer_size_overrides(self):
        say_stream = SayStream(
            client=self.web_client,
            channel="C111",
            recipient_team_id="T111",
            recipient_user_id="U111",
            thread_ts="111.222",
        )
        stream = say_stream(
            buffer_size=100,
            channel="C222",
            thread_ts="333.444",
            recipient_team_id="T222",
            recipient_user_id="U222",
        )

        assert stream._buffer_size == 100
        assert stream._stream_args == {
            "channel": "C222",
            "thread_ts": "333.444",
            "recipient_team_id": "T222",
            "recipient_user_id": "U222",
            "task_display_mode": None,
        }