Replace Flask-Sockets with aiohttp for testing

.github/workflows/tests.yml

@@ -23,7 +23,7 @@ jobs:
         run: |
           pip install -U pip
           pip install -r requirements.txt
-          pip install -r requirements/testing_without_asyncio.txt
+          pip install -r requirements/testing.txt
       - name: Run tests without aiohttp
         run: |
           pytest tests/slack_bolt/

requirements/adapter_testing.txt

@@ -1,8 +1,5 @@
 # pip install -r requirements/adapter_testing.txt
-moto>=3,<4                               # For AWS tests
+moto>=3,<5                               # For AWS tests
 docker>=5,<6                             # Used by moto
 boddle>=0.2,<0.3                         # For Bottle app tests
-Flask>=1,<2                              # TODO: Flask-Sockets is not yet compatible with Flask 2.x
-Werkzeug>=1,<2                           # TODO: Flask-Sockets is not yet compatible with Flask 2.x
 sanic-testing>=0.7; python_version>"3.6"
-requests>=2,<3                           # For Starlette's TestClient

requirements/testing.txt

@@ -1,6 +1,7 @@
 # pip install -r requirements/testing.txt
--r testing_without_asyncio.txt
-
-pytest-asyncio>=0.16.0; python_version=="3.6"
-pytest-asyncio>=0.18.2,<1; python_version>"3.6"
-aiohttp>=3,<4
+pytest>=6.2.5,<7
+pytest-cov>=3,<5
+aiohttp<4
+black==22.8.0             # Until we drop Python 3.6 support, we have to stay with this version
+click<=8.0.4              # black is affected by https://github.com/pallets/click/issues/2225
+pytest-asyncio<1;

tests/adapter_tests/socket_mode/mock_socket_mode_server.py

@@ -1,78 +1,107 @@
-import json
+import asyncio
 import logging
 import threading
 import time
-import requests
-from typing import List
 from unittest import TestCase
+from urllib.error import URLError
+from urllib.request import urlopen
+
+from aiohttp import WSMsgType, web
 
 socket_mode_envelopes = [
     """{"envelope_id":"57d6a792-4d35-4d0b-b6aa-3361493e1caf","payload":{"type":"shortcut","token":"xxx","action_ts":"1610198080.300836","team":{"id":"T111","domain":"seratch"},"user":{"id":"U111","username":"seratch","team_id":"T111"},"is_enterprise_install":false,"enterprise":null,"callback_id":"do-something","trigger_id":"111.222.xxx"},"type":"interactive","accepts_response_payload":false}""",
     """{"envelope_id":"1d3c79ab-0ffb-41f3-a080-d19e85f53649","payload":{"token":"xxx","team_id":"T111","team_domain":"xxx","channel_id":"C111","channel_name":"random","user_id":"U111","user_name":"seratch","command":"/hello-socket-mode","text":"","api_app_id":"A111","response_url":"https://hooks.slack.com/commands/T111/111/xxx","trigger_id":"111.222.xxx"},"type":"slash_commands","accepts_response_payload":true}""",
     """{"envelope_id":"08cfc559-d933-402e-a5c1-79e135afaae4","payload":{"token":"xxx","team_id":"T111","api_app_id":"A111","event":{"client_msg_id":"c9b466b5-845c-49c6-a371-57ae44359bf1","type":"message","text":"<@W111>","user":"U111","ts":"1610197986.000300","team":"T111","blocks":[{"type":"rich_text","block_id":"1HBPc","elements":[{"type":"rich_text_section","elements":[{"type":"user","user_id":"U111"}]}]}],"channel":"C111","event_ts":"1610197986.000300","channel_type":"channel"},"type":"event_callback","event_id":"Ev111","event_time":1610197986,"authorizations":[{"enterprise_id":null,"team_id":"T111","user_id":"U111","is_bot":true,"is_enterprise_install":false}],"is_ext_shared_channel":false,"event_context":"1-message-T111-C111"},"type":"events_api","accepts_response_payload":false,"retry_attempt":1,"retry_reason":"timeout"}""",
 ]
 
-from flask import Flask
-from flask_sockets import Sockets
-
 
 def start_thread_socket_mode_server(test: TestCase, port: int):
-    def _start_thread_socket_mode_server():
-        logger = logging.getLogger(__name__)
-        app: Flask = Flask(__name__)
+    logger = logging.getLogger(__name__)
+    state = {}
+
+    def reset_server_state():
+        state.update(
+            envelopes_to_consume=list(socket_mode_envelopes),
+        )
+
+    test.reset_server_state = reset_server_state
+
+    async def health(request: web.Request):
+        wr = web.Response()
+        await wr.prepare(request)
+        wr.set_status(200)
+        return wr
+
+    async def link(request: web.Request):
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+
+        async for msg in ws:
+            if msg.type != WSMsgType.TEXT:
+                continue
 
-        @app.route("/state")
-        def state():
-            return json.dumps({"success": True}), 200, {"ContentType": "application/json"}
+            if state["envelopes_to_consume"]:
+                e = state["envelopes_to_consume"].pop(0)
+                logger.debug(f"Send an envelope: {e}")
+                await ws.send_str(e)
 
-        sockets: Sockets = Sockets(app)
+            message = msg.data
+            logger.debug(f"Server received a message: {message}")
 
-        envelopes_to_consume: List[str] = list(socket_mode_envelopes)
+            await ws.send_str(message)
 
-        @sockets.route("/link")
-        def link(ws):
-            while not ws.closed:
-                message = ws.read_message()
-                if message is not None:
-                    if len(envelopes_to_consume) > 0:
-                        e = envelopes_to_consume.pop(0)
-                        logger.debug(f"Send an envelope: {e}")
-                        ws.send(e)
+        return ws
 
-                    logger.debug(f"Server received a message: {message}")
-                    ws.send(message)
+    app = web.Application()
+    app.add_routes(
+        [
+            web.get("/link", link),
+            web.get("/health", health),
+        ]
+    )
+    runner = web.AppRunner(app)
 
-        from gevent import pywsgi
-        from geventwebsocket.handler import WebSocketHandler
+    def run_server():
+        reset_server_state()
 
-        server = pywsgi.WSGIServer(("", port), app, handler_class=WebSocketHandler)
-        test.server = server
-        server.serve_forever(stop_timeout=1)
+        test.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(test.loop)
+        test.loop.run_until_complete(runner.setup())
+        site = web.TCPSite(runner, "127.0.0.1", port, reuse_port=True)
+        test.loop.run_until_complete(site.start())
 
-    return _start_thread_socket_mode_server
+        # run until it's stopped from the main thread
+        test.loop.run_forever()
+
+        test.loop.run_until_complete(runner.cleanup())
+        test.loop.close()
+
+    return run_server
 
 
 def start_socket_mode_server(test, port: int):
     test.sm_thread = threading.Thread(target=start_thread_socket_mode_server(test, port))
     test.sm_thread.daemon = True
     test.sm_thread.start()
-    wait_for_socket_mode_server(port, 4)  # wait for the server
+    wait_for_socket_mode_server(port, 4)
 
 
-def wait_for_socket_mode_server(port: int, secs: int):
+def wait_for_socket_mode_server(port: int, timeout: int):
     start_time = time.time()
-    while (time.time() - start_time) < secs:
-        response = requests.get(url=f"http://localhost:{port}/state")
-        if response.ok:
-            break
-        time.sleep(0.01)
-
-
-def stop_socket_mode_server(test):
-    test.server.stop()
-    test.server.close()
-
-
-async def stop_socket_mode_server_async(test: TestCase):
-    test.server.stop()
-    test.server.close()
+    while (time.time() - start_time) < timeout:
+        try:
+            urlopen(f"http://127.0.0.1:{port}/health")
+            return
+        except URLError:
+            time.sleep(0.01)
+
+
+def stop_socket_mode_server(test: TestCase):
+    # An event loop runs in a thread and executes all callbacks and Tasks in
+    # its thread. While a Task is running in the event loop, no other Tasks
+    # can run in the same thread. When a Task executes an await expression, the
+    # running Task gets suspended, and the event loop executes the next Task.
+    # To schedule a callback from another OS thread, the loop.call_soon_threadsafe() method should be used.
+    # https://docs.python.org/3/library/asyncio-dev.html#asyncio-multithreading
+    test.loop.call_soon_threadsafe(test.loop.stop)
+    test.sm_thread.join(timeout=5)

tests/adapter_tests_async/socket_mode/test_async_aiohttp.py

@@ -12,7 +12,7 @@
 from tests.utils import remove_os_env_temporarily, restore_os_env, get_event_loop
 from ...adapter_tests.socket_mode.mock_socket_mode_server import (
     start_socket_mode_server,
-    stop_socket_mode_server_async,
+    stop_socket_mode_server,
 )
 
 
@@ -71,4 +71,4 @@ async def command_handler(ack):
             assert result["command"] is True
         finally:
             await handler.client.close()
-            await stop_socket_mode_server_async(self)
+            stop_socket_mode_server(self)

tests/adapter_tests_async/socket_mode/test_async_lazy_listeners.py

@@ -12,7 +12,7 @@
 from tests.utils import remove_os_env_temporarily, restore_os_env, get_event_loop
 from ...adapter_tests.socket_mode.mock_socket_mode_server import (
     start_socket_mode_server,
-    stop_socket_mode_server_async,
+    stop_socket_mode_server,
 )
 
 
@@ -80,4 +80,4 @@ async def lazy_func(body):
 
         finally:
             await handler.client.close()
-            await stop_socket_mode_server_async(self)
+            stop_socket_mode_server(self)

tests/adapter_tests_async/socket_mode/test_async_websockets.py

@@ -12,7 +12,7 @@
 from tests.utils import remove_os_env_temporarily, restore_os_env, get_event_loop
 from ...adapter_tests.socket_mode.mock_socket_mode_server import (
     start_socket_mode_server,
-    stop_socket_mode_server_async,
+    stop_socket_mode_server,
 )
 
 
@@ -71,4 +71,4 @@ async def command_handler(ack):
             assert result["command"] is True
         finally:
             await handler.client.close()
-            await stop_socket_mode_server_async(self)
+            stop_socket_mode_server(self)