Added socketio test client

Author: eadwinCode

ellar/socket_io/testing/__init__.py

@@ -0,0 +1,3 @@
+from .module import SocketIOTestingModule, TestGateway
+
+__all__ = ["SocketIOTestingModule", "TestGateway"]

ellar/socket_io/testing/module.py

@@ -0,0 +1,59 @@
+import typing as t
+
+import socketio
+
+from ellar.compatible import asynccontextmanager
+from ellar.testing.module import Test, TestingModule
+from ellar.testing.uvicorn_server import EllarUvicornServer
+
+
+class RunWithServerContext:
+    __slots__ = ("sio", "base_url")
+
+    def __init__(self, sio: socketio.AsyncClient, base_url: str) -> None:
+        self.sio = sio
+        self.base_url: str = base_url
+
+    async def connect(
+        self,
+        path: str = "",
+        namespaces: str = "/",
+        socketio_path: str = "socket.io",
+        **kwargs: t.Any,
+    ) -> None:
+        assert path == "" or path.startswith("/"), "Routed paths must start with '/'"
+        await self.sio.connect(
+            self.base_url + path,
+            namespaces=namespaces,
+            socketio_path=socketio_path,
+            **kwargs,
+        )
+
+    async def wait(self, seconds: float = 0.5) -> None:
+        await self.sio.sleep(seconds)
+
+    def new_socket_client_context(self) -> "RunWithServerContext":
+        sio = socketio.AsyncClient()
+        return self.__class__(sio=sio, base_url=self.base_url)
+
+
+class SocketIOTestingModule(TestingModule):
+    @asynccontextmanager
+    async def run_with_server(
+        self, host: str = "127.0.0.1", port: int = 8000
+    ) -> t.AsyncIterator[RunWithServerContext]:
+        base_url = f"http://{host}:{port}"
+
+        server = EllarUvicornServer(app=self.create_application(), host=host, port=port)
+
+        await server.start_up()
+        sio = socketio.AsyncClient()
+
+        yield RunWithServerContext(sio=sio, base_url=base_url)
+
+        await sio.disconnect()
+        await server.tear_down()
+
+
+class TestGateway(Test):
+    TESTING_MODULE = SocketIOTestingModule

ellar/testing/uvicorn_server.py

@@ -0,0 +1,42 @@
+import asyncio
+import typing as t
+
+import uvicorn
+from starlette.types import ASGIApp
+
+
+class EllarUvicornServer(uvicorn.Server):
+    """
+    Uvicorn Server for testing purpose.
+
+
+    @pytest_asyncio.fixture
+    async def start_server():
+        server = EllarUvicornServer(asgi_app)
+        await server.start_up()
+        yield
+        await server.tear_down()
+
+    """
+
+    def __init__(self, app: ASGIApp, host: str = "127.0.0.1", port: int = 8000):
+        self._startup_done = asyncio.Event()
+        self._serve_task: t.Optional[t.Awaitable[t.Any]] = None
+        super().__init__(config=uvicorn.Config(app, host=host, port=port))
+
+    async def startup(self, sockets: list = None) -> None:
+        """Override uvicorn startup"""
+        await super().startup(sockets)
+        self.config.setup_event_loop()
+        self._startup_done.set()
+
+    async def start_up(self) -> None:
+        """Start up server asynchronously"""
+        self._serve_task = asyncio.create_task(self.serve())
+        await self._startup_done.wait()
+
+    async def tear_down(self) -> None:
+        """Shut down server asynchronously"""
+        self.should_exit = True
+        if self._serve_task:
+            await self._serve_task