[FIX] Fixed flaky tests on uvloop

tests/functional_test/test_communication/test_async/test_client/test_tcp.py

@@ -15,6 +15,8 @@
 import pytest
 import pytest_asyncio
 
+from .....tools import is_uvloop_event_loop
+
 
 async def readline(loop: asyncio.AbstractEventLoop, sock: Socket) -> bytes:
     buf: list[bytes] = []
@@ -72,42 +74,50 @@ async def test____send_packet____default(
 
     async def test____send_packet____connection_error____fresh_connection_closed_by_server(
         self,
+        event_loop: asyncio.AbstractEventLoop,
         client: AsyncTCPNetworkClient[str, str],
         server: Socket,
     ) -> None:
+        if is_uvloop_event_loop(event_loop):
+            pytest.skip("It is not mandadtory for uvloop to raise ConnectionAbortedError")
         server.close()
         with pytest.raises(ConnectionAbortedError):
             for _ in range(3):  # Windows and macOS catch the issue after several send()
                 await client.send_packet("ABCDEF")
-                await asyncio.sleep(0)
+                await asyncio.sleep(0.01)
 
     async def test____send_packet____connection_error____after_previous_successful_try(
         self,
         event_loop: asyncio.AbstractEventLoop,
         client: AsyncTCPNetworkClient[str, str],
         server: Socket,
     ) -> None:
+        if is_uvloop_event_loop(event_loop):
+            pytest.skip("It is not mandadtory for uvloop to raise ConnectionAbortedError")
+
         await client.send_packet("ABCDEF")
         assert await readline(event_loop, server) == b"ABCDEF\n"
         server.close()
         with pytest.raises(ConnectionAbortedError):
             for _ in range(3):  # Windows and macOS catch the issue after several send()
                 await client.send_packet("ABCDEF")
-                await asyncio.sleep(0)
+                await asyncio.sleep(0.01)
 
     async def test____send_packet____connection_error____partial_read_then_close(
         self,
         event_loop: asyncio.AbstractEventLoop,
         client: AsyncTCPNetworkClient[str, str],
         server: Socket,
     ) -> None:
+        if is_uvloop_event_loop(event_loop):
+            pytest.skip("It is not mandadtory for uvloop to raise ConnectionAbortedError")
         await client.send_packet("ABC")
         assert await event_loop.sock_recv(server, 1) == b"A"
         server.close()
         with pytest.raises(ConnectionAbortedError):
             for _ in range(3):  # Windows and macOS catch the issue after several send()
                 await client.send_packet("DEF")
-                await asyncio.sleep(0)
+                await asyncio.sleep(0.01)
 
     async def test____send_packet____closed_client(self, client: AsyncTCPNetworkClient[str, str]) -> None:
         await client.aclose()

tests/tools.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import importlib
 import sys
 import time
 from collections.abc import Generator
@@ -70,6 +71,14 @@ def is_proactor_event_loop(event_loop: asyncio.AbstractEventLoop) -> bool:
     return isinstance(event_loop, ProactorEventLoop)
 
 
+def is_uvloop_event_loop(event_loop: asyncio.AbstractEventLoop) -> bool:
+    try:
+        uvloop = importlib.import_module("uvloop")
+    except ModuleNotFoundError:
+        return False
+    return isinstance(event_loop, uvloop.Loop)
+
+
 _TooShortBufferBehavior: TypeAlias = Literal["error", "fill_at_most", "xfail"]