fixing ci tests

Author: eadwinCode

ellar/socket_io/testing/module.py

@@ -1,9 +1,17 @@
 import typing as t
 from contextlib import asynccontextmanager
+from pathlib import Path
 
 import socketio
 from ellar.testing.module import Test, TestingModule
 from ellar.testing.uvicorn_server import EllarUvicornServer
+from starlette.routing import Host, Mount
+
+if t.TYPE_CHECKING:  # pragma: no cover
+    from ellar.common import ControllerBase, GuardCanActivate, ModuleRouter
+    from ellar.core import ModuleBase
+    from ellar.core.routing import EllarControllerMount
+    from ellar.di import ProviderConfig
 
 
 class RunWithServerContext:
@@ -62,3 +70,37 @@ async def run_with_server(
 
 class TestGateway(Test):
     TESTING_MODULE = SocketIOTestingModule
+
+    @classmethod
+    def create_test_module(
+        cls,
+        controllers: t.Sequence[t.Union[t.Type["ControllerBase"], t.Type]] = (),
+        routers: t.Sequence[
+            t.Union["ModuleRouter", "EllarControllerMount", Mount, Host, t.Callable]
+        ] = (),
+        providers: t.Sequence[t.Union[t.Type, "ProviderConfig"]] = (),
+        template_folder: t.Optional[str] = "templates",
+        base_directory: t.Optional[t.Union[Path, str]] = None,
+        static_folder: str = "static",
+        modules: t.Sequence[t.Union[t.Type, t.Any]] = (),
+        application_module: t.Optional[t.Union[t.Type["ModuleBase"], str]] = None,
+        global_guards: t.Optional[
+            t.List[t.Union[t.Type["GuardCanActivate"], "GuardCanActivate"]]
+        ] = None,
+        config_module: t.Optional[t.Union[str, t.Dict]] = None,
+    ) -> SocketIOTestingModule:
+        return t.cast(
+            SocketIOTestingModule,
+            super().create_test_module(
+                controllers=controllers,
+                routers=routers,
+                providers=providers,
+                template_folder=template_folder,
+                base_directory=base_directory,
+                static_folder=static_folder,
+                modules=modules,
+                application_module=application_module,
+                global_guards=global_guards,
+                config_module=config_module,
+            ),
+        )

tests/conftest.py

@@ -1,4 +1,6 @@
+import contextlib
 import functools
+import socket
 from pathlib import PurePath, PurePosixPath, PureWindowsPath
 from uuid import uuid4
 
@@ -52,3 +54,17 @@ def reflect_context():
 async def async_reflect_context():
     async with reflect.async_context():
         yield
+
+
+def _unused_port(socket_type: int) -> int:
+    """Find an unused localhost port from 1024-65535 and return it."""
+    with contextlib.closing(socket.socket(type=socket_type)) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+# This was copied from pytest-asyncio.
+# Ref.: https://github.com/pytest-dev/pytest-asyncio/blob/25d9592286682bc6dbfbf291028ff7a9594cf283/pytest_asyncio/plugin.py#L525-L527
+@pytest.fixture
+def unused_tcp_port() -> int:
+    return _unused_port(socket.SOCK_STREAM)

tests/test_socket_io/test_operation.py

@@ -20,12 +20,12 @@
 class TestEventGateway:
     test_client = TestGateway.create_test_module(controllers=[EventGateway])
 
-    async def test_socket_connection_work(self):
+    async def test_socket_connection_work(self, unused_tcp_port):
         my_response_message = []
         connected_called = False
         disconnected_called = False
 
-        async with self.test_client.run_with_server() as ctx:
+        async with self.test_client.run_with_server(port=unused_tcp_port) as ctx:
 
             @ctx.sio.event
             async def my_response(message):
@@ -52,11 +52,11 @@ async def connect(*args):
         ]
         assert disconnected_called and connected_called
 
-    async def test_broadcast_work(self):
+    async def test_broadcast_work(self, unused_tcp_port):
         sio_1_response_message = []
         sio_2_response_message = []
 
-        async with self.test_client.run_with_server() as ctx:
+        async with self.test_client.run_with_server(port=unused_tcp_port) as ctx:
             ctx_2 = ctx.new_socket_client_context()
 
             @ctx.sio.on("my_response")
@@ -94,10 +94,10 @@ async def my_response_case_2(message):
 class TestGatewayWithGuards:
     test_client = TestGateway.create_test_module(controllers=[GatewayWithGuards])
 
-    async def test_socket_connection_work(self):
+    async def test_socket_connection_work(self, unused_tcp_port):
         my_response_message = []
 
-        async with self.test_client.run_with_server() as ctx:
+        async with self.test_client.run_with_server(port=unused_tcp_port) as ctx:
 
             @ctx.sio.event
             async def my_response(message):
@@ -113,10 +113,10 @@ async def my_response(message):
             {"auth-key": "supersecret", "data": "Testing Broadcast"}
         ]
 
-    async def test_event_with_header_work(self):
+    async def test_event_with_header_work(self, unused_tcp_port):
         my_response_message = []
 
-        async with self.test_client.run_with_server() as ctx:
+        async with self.test_client.run_with_server(port=unused_tcp_port) as ctx:
 
             @ctx.sio.event
             async def my_response(message):
@@ -132,10 +132,10 @@ async def my_response(message):
             {"data": "Testing Broadcast", "x_auth_key": "supersecret"}
         ]
 
-    async def test_event_with_plain_response(self):
+    async def test_event_with_plain_response(self, unused_tcp_port):
         my_response_message = []
 
-        async with self.test_client.run_with_server() as ctx:
+        async with self.test_client.run_with_server(port=unused_tcp_port) as ctx:
 
             @ctx.sio.on("my_plain_response")
             async def message_receive(message):
@@ -151,10 +151,10 @@ async def message_receive(message):
             {"data": "Testing Broadcast", "x_auth_key": "supersecret"}
         ]
 
-    async def test_failed_to_connect(self):
+    async def test_failed_to_connect(self, unused_tcp_port):
         my_response_message = []
 
-        async with self.test_client.run_with_server() as ctx:
+        async with self.test_client.run_with_server(port=unused_tcp_port) as ctx:
             ctx = typing.cast(RunWithServerContext, ctx)
 
             @ctx.sio.on("error")
@@ -169,10 +169,10 @@ async def error(message):
 
         assert my_response_message == [{"code": 1011, "reason": "Authorization Failed"}]
 
-    async def test_failed_process_message_sent(self):
+    async def test_failed_process_message_sent(self, unused_tcp_port):
         my_response_message = []
 
-        async with self.test_client.run_with_server() as ctx:
+        async with self.test_client.run_with_server(port=unused_tcp_port) as ctx:
             ctx = typing.cast(RunWithServerContext, ctx)
 
             @ctx.sio.on("error")
@@ -224,13 +224,15 @@ class TestGatewayExceptions:
             ),
         ],
     )
-    async def test_exception_handling_works_debug_true_or_false(self, debug, result):
+    async def test_exception_handling_works_debug_true_or_false(
+        self, debug, result, unused_tcp_port
+    ):
         test_client = TestGateway.create_test_module(
             controllers=[GatewayOthers], config_module={"DEBUG": debug}
         )
         my_response_message = []
 
-        async with test_client.run_with_server() as ctx:
+        async with test_client.run_with_server(port=unused_tcp_port) as ctx:
             ctx = typing.cast(RunWithServerContext, ctx)
             ctx2 = ctx.new_socket_client_context()
 
@@ -253,11 +255,11 @@ async def error_2(message):
 
         assert my_response_message == result
 
-    async def test_message_with_extra_args(self):
+    async def test_message_with_extra_args(self, unused_tcp_port):
         test_client = TestGateway.create_test_module(controllers=[GatewayOthers])
         my_response_message = []
 
-        async with test_client.run_with_server() as ctx:
+        async with test_client.run_with_server(port=unused_tcp_port) as ctx:
             ctx = typing.cast(RunWithServerContext, ctx)
 
             @ctx.sio.on("error")