From 0c942cd24a5a77dbe6840178ae88078ccebe0d0d Mon Sep 17 00:00:00 2001 From: Kazuhiro Sera Date: Mon, 25 May 2020 10:04:16 +0900 Subject: [PATCH 1/2] Fix #701 by reverting the change for simultaneous request handling --- integration_tests/rtm/test_issue_558.py | 2 +- integration_tests/rtm/test_issue_701.py | 135 ++++++++++++++++++++++++ slack/rtm/client.py | 45 +++----- 3 files changed, 151 insertions(+), 31 deletions(-) create mode 100644 integration_tests/rtm/test_issue_701.py diff --git a/integration_tests/rtm/test_issue_558.py b/integration_tests/rtm/test_issue_558.py index e20b43d45..cd115c880 100644 --- a/integration_tests/rtm/test_issue_558.py +++ b/integration_tests/rtm/test_issue_558.py @@ -27,7 +27,7 @@ def tearDown(self): # Reset the decorators by @RTMClient.run_on RTMClient._callbacks = collections.defaultdict(list) - @pytest.mark.skipif(condition=is_not_specified(), reason="To avoid rate limited errors") + @pytest.mark.skipif(condition=is_not_specified(), reason="Still unfixed") @async_test async def test_issue_558(self): channel_id = os.environ[SLACK_SDK_TEST_RTM_TEST_CHANNEL_ID] diff --git a/integration_tests/rtm/test_issue_701.py b/integration_tests/rtm/test_issue_701.py new file mode 100644 index 000000000..9e0bb4d59 --- /dev/null +++ b/integration_tests/rtm/test_issue_701.py @@ -0,0 +1,135 @@ +import asyncio +import collections +import logging +import os +import threading +import time +import unittest + +import pytest + +from integration_tests.env_variable_names import SLACK_SDK_TEST_CLASSIC_APP_BOT_TOKEN +from integration_tests.helpers import async_test, is_not_specified +from slack import RTMClient, WebClient + + +class TestRTMClient(unittest.TestCase): + """Runs integration tests with real Slack API + + https://github.com/slackapi/python-slackclient/issues/701 + """ + + def setUp(self): + self.logger = logging.getLogger(__name__) + self.bot_token = os.environ[SLACK_SDK_TEST_CLASSIC_APP_BOT_TOKEN] + + def tearDown(self): + # Reset the decorators by @RTMClient.run_on + RTMClient._callbacks = collections.defaultdict(list) + + # @pytest.mark.skipif(condition=is_not_specified(), reason="to avoid rate_limited errors") + @pytest.mark.skip() + def test_receiving_all_messages(self): + self.rtm_client = RTMClient(token=self.bot_token, loop=asyncio.new_event_loop()) + self.web_client = WebClient(token=self.bot_token) + + self.call_count = 0 + + @RTMClient.run_on(event="message") + def send_reply(**payload): + self.logger.debug(payload) + web_client, data = payload["web_client"], payload["data"] + web_client.reactions_add(channel=data["channel"], timestamp=data["ts"], name="eyes") + self.call_count += 1 + + def connect(): + self.logger.debug("Starting RTM Client...") + self.rtm_client.start() + + rtm = threading.Thread(target=connect) + rtm.setDaemon(True) + + rtm.start() + time.sleep(3) + + total_num = 10 + + sender_completion = [] + + def sent_bulk_message(): + for i in range(total_num): + text = f"Sent by ! ({i})" + self.web_client.chat_postMessage(channel="#random", text=text) + time.sleep(0.1) + sender_completion.append(True) + + num_of_senders = 3 + senders = [] + for sender_num in range(num_of_senders): + sender = threading.Thread(target=sent_bulk_message) + sender.setDaemon(True) + sender.start() + senders.append(sender) + + while len(sender_completion) < num_of_senders: + time.sleep(1) + + expected_call_count = total_num * num_of_senders + wait_seconds = 0 + max_wait = 20 + while self.call_count < expected_call_count and wait_seconds < max_wait: + time.sleep(1) + wait_seconds += 1 + + self.assertEqual(total_num * num_of_senders, self.call_count, "The RTM handler failed") + + @pytest.mark.skipif(condition=is_not_specified(), reason="to avoid rate_limited errors") + @async_test + async def test_receiving_all_messages_async(self): + self.rtm_client = RTMClient(token=self.bot_token, run_async=True) + self.web_client = WebClient(token=self.bot_token, run_async=False) + + self.call_count = 0 + + @RTMClient.run_on(event="message") + async def send_reply(**payload): + self.logger.debug(payload) + web_client, data = payload["web_client"], payload["data"] + await web_client.reactions_add(channel=data["channel"], timestamp=data["ts"], name="eyes") + self.call_count += 1 + + # intentionally not waiting here + self.rtm_client.start() + + await asyncio.sleep(3) + + total_num = 10 + + sender_completion = [] + + def sent_bulk_message(): + for i in range(total_num): + text = f"Sent by ! ({i})" + self.web_client.chat_postMessage(channel="#random", text=text) + time.sleep(0.1) + sender_completion.append(True) + + num_of_senders = 3 + senders = [] + for sender_num in range(num_of_senders): + sender = threading.Thread(target=sent_bulk_message) + sender.setDaemon(True) + sender.start() + senders.append(sender) + + while len(sender_completion) < num_of_senders: + await asyncio.sleep(1) + + expected_call_count = total_num * num_of_senders + wait_seconds = 0 + max_wait = 20 + while self.call_count < expected_call_count and wait_seconds < max_wait: + await asyncio.sleep(1) + wait_seconds += 1 + + self.assertEqual(total_num * num_of_senders, self.call_count, "The RTM handler failed") diff --git a/slack/rtm/client.py b/slack/rtm/client.py index 0a8928a9b..d75670d28 100644 --- a/slack/rtm/client.py +++ b/slack/rtm/client.py @@ -394,12 +394,7 @@ async def _connect_and_read(self): async def _read_messages(self): """Process messages received on the WebSocket connection.""" - text_message_callback_executions: List[Future] = [] while not self._stopped and self._websocket is not None: - for future in text_message_callback_executions: - if future.done(): - text_message_callback_executions.remove(future) - try: # Wait for a message to be received, but timeout after a second so that # we can check if the socket has been closed, or if self._stopped is @@ -419,34 +414,22 @@ async def _read_messages(self): ) self._websocket = None await self._dispatch_event(event="close") - num_of_running_callbacks = len(text_message_callback_executions) - if num_of_running_callbacks > 0: - self._logger.info( - "WebSocket connection has been closed " - f"though {num_of_running_callbacks} callback executions were still in progress" - ) return if message.type == aiohttp.WSMsgType.TEXT: - payload = message.json() - event = payload.pop("type", "Unknown") - - async def run_dispatch_event(): - try: - await self._dispatch_event(event, data=payload) - except Exception as err: - data = message.data if message else message - self._logger.info( - f"Caught a raised exception ({err}) while dispatching a TEXT message ({data})" - ) - # Raised exceptions here happen in users' code and were just unhandled. - # As they're not intended for closing current WebSocket connection, - # this exception should not be propagated to higher level (#_connect_and_read()). - return - - # Asynchronously run callbacks to handle simultaneous incoming messages from Slack - f = asyncio.ensure_future(run_dispatch_event()) - text_message_callback_executions.append(f) + try: + payload = message.json() + event = payload.pop("type", "Unknown") + await self._dispatch_event(event, data=payload) + except Exception as err: + data = message.data if message else message + self._logger.info( + f"Caught a raised exception ({err}) while dispatching a TEXT message ({data})" + ) + # Raised exceptions here happen in users' code and were just unhandled. + # As they're not intended for closing current WebSocket connection, + # this exception should not be propagated to higher level (#_connect_and_read()). + continue elif message.type == aiohttp.WSMsgType.ERROR: self._logger.error("Received an error on the websocket: %r", message) await self._dispatch_event(event="error", data=message) @@ -479,6 +462,8 @@ async def _dispatch_event(self, event, data=None): } } """ + if self._logger.level <= logging.DEBUG: + self._logger.debug("Received an event: '%s' - %s", event, data) for callback in self._callbacks[event]: self._logger.debug( "Running %s callbacks for event: '%s'", From 10563efcce9c40795069cf8cdc5ec926080ef1fc Mon Sep 17 00:00:00 2001 From: Kazuhiro Sera Date: Mon, 25 May 2020 10:19:49 +0900 Subject: [PATCH 2/2] Exclude integration tests from DeepSource analysis --- .deepsource.toml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.deepsource.toml b/.deepsource.toml index 00bf68bbd..3d884cae7 100644 --- a/.deepsource.toml +++ b/.deepsource.toml @@ -8,7 +8,9 @@ test_patterns = [ exclude_patterns = [ 'setup.py', 'docs-src/**/*.py', - 'tutorial/**/*.py' + 'tutorial/**/*.py', + 'integration_tests/web/*.py', + 'integration_tests/rtm/*.py' ] [[analyzers]]