[FIX] Socket adapters that use asyncio.Transport and asyncio.Datagram…

…Transport do not use is_closing() anymore

src/easynetwork/serializers/cbor.py

@@ -88,8 +88,8 @@ def __init__(
             raise ModuleNotFoundError("cbor dependencies are missing. Consider adding 'cbor' extra") from exc
 
         super().__init__(expected_load_error=(cbor2.CBORDecodeError, UnicodeError))
-        self.__encoder_cls: Callable[[IO[bytes]], cbor2.CBOREncoder]  # type: ignore[no-any-unimported]
-        self.__decoder_cls: Callable[[IO[bytes]], cbor2.CBORDecoder]  # type: ignore[no-any-unimported]
+        self.__encoder_cls: Callable[[IO[bytes]], cbor2.CBOREncoder]  # type: ignore[no-any-unimported,unused-ignore]
+        self.__decoder_cls: Callable[[IO[bytes]], cbor2.CBORDecoder]  # type: ignore[no-any-unimported,unused-ignore]
 
         if encoder_config is None:
             encoder_config = CBOREncoderConfig()

src/easynetwork_asyncio/datagram/listener.py

@@ -42,6 +42,7 @@ class AsyncioTransportDatagramListenerSocketAdapter(transports.AsyncDatagramList
     __slots__ = (
         "__endpoint",
         "__socket",
+        "__closing",
     )
 
     def __init__(self, endpoint: DatagramEndpoint) -> None:
@@ -52,11 +53,16 @@ def __init__(self, endpoint: DatagramEndpoint) -> None:
         assert socket is not None, "transport must be a socket transport"  # nosec assert_used
 
         self.__socket: asyncio.trsock.TransportSocket = socket
+        # asyncio.DatagramTransport.is_closing() can suddently become true if there is something wrong with the socket
+        # even if transport.close() was never called.
+        # To bypass this side effect, we use our own flag.
+        self.__closing: bool = False
 
     def is_closing(self) -> bool:
-        return self.__endpoint.is_closing()
+        return self.__closing
 
     async def aclose(self) -> None:
+        self.__closing = True
         self.__endpoint.close()
         try:
             await self.__endpoint.wait_closed()

src/easynetwork_asyncio/datagram/socket.py

@@ -41,6 +41,7 @@ class AsyncioTransportDatagramSocketAdapter(transports.AsyncDatagramTransport):
     __slots__ = (
         "__endpoint",
         "__socket",
+        "__closing",
     )
 
     def __init__(self, endpoint: DatagramEndpoint) -> None:
@@ -51,8 +52,13 @@ def __init__(self, endpoint: DatagramEndpoint) -> None:
         assert socket is not None, "transport must be a socket transport"  # nosec assert_used
 
         self.__socket: asyncio.trsock.TransportSocket = socket
+        # asyncio.DatagramTransport.is_closing() can suddently become true if there is something wrong with the socket
+        # even if transport.close() was never called.
+        # To bypass this side effect, we use our own flag.
+        self.__closing: bool = False
 
     async def aclose(self) -> None:
+        self.__closing = True
         self.__endpoint.close()
         try:
             return await self.__endpoint.wait_closed()
@@ -61,7 +67,7 @@ async def aclose(self) -> None:
             raise
 
     def is_closing(self) -> bool:
-        return self.__endpoint.is_closing()
+        return self.__closing
 
     async def recv(self) -> bytes:
         data, _ = await self.__endpoint.recvfrom()

src/easynetwork_asyncio/stream/socket.py

@@ -41,6 +41,7 @@ class AsyncioTransportStreamSocketAdapter(transports.AsyncStreamTransport):
         "__reader",
         "__writer",
         "__socket",
+        "__closing",
     )
 
     def __init__(
@@ -56,7 +57,13 @@ def __init__(
         assert socket is not None, "Writer transport must be a socket transport"  # nosec assert_used
         self.__socket: asyncio.trsock.TransportSocket = socket
 
+        # asyncio.Transport.is_closing() can suddently become true if there is something wrong with the socket
+        # even if transport.close() was never called.
+        # To bypass this side effect, we use our own flag.
+        self.__closing: bool = False
+
     async def aclose(self) -> None:
+        self.__closing = True
         if not self.__writer.is_closing():
             try:
                 if self.__writer.can_write_eof():
@@ -74,7 +81,7 @@ async def aclose(self) -> None:
             raise
 
     def is_closing(self) -> bool:
-        return self.__writer.is_closing()
+        return self.__closing
 
     async def recv(self, bufsize: int) -> bytes:
         if bufsize < 0:

tests/unit_test/test_async/test_asyncio_backend/test_datagram.py

@@ -730,21 +730,25 @@ async def test____aclose____abort_transport_if_cancelled(
         mock_endpoint.wait_closed.assert_awaited_once_with()
         mock_endpoint.transport.abort.assert_called_once_with()
 
-    async def test____is_closing____return_endpoint_state(
+    @pytest.mark.parametrize("transport_closed", [False, True], ids=lambda p: f"transport_closed=={p}")
+    async def test____is_closing____return_internal_flag(
         self,
+        transport_closed: bool,
         socket: AsyncBaseTransport,
         mock_endpoint: MagicMock,
-        mocker: MockerFixture,
     ) -> None:
         # Arrange
-        mock_endpoint.is_closing.return_value = mocker.sentinel.is_closing
+        if transport_closed:
+            await socket.aclose()
+            mock_endpoint.reset_mock()
+        mock_endpoint.is_closing.side_effect = AssertionError
 
         # Act
         state = socket.is_closing()
 
         # Assert
-        mock_endpoint.is_closing.assert_called_once_with()
-        assert state is mocker.sentinel.is_closing
+        mock_endpoint.is_closing.assert_not_called()
+        assert state is transport_closed
 
 
 @pytest.mark.asyncio

tests/unit_test/test_async/test_asyncio_backend/test_stream.py

@@ -149,21 +149,25 @@ async def test____aclose____abort_transport_if_cancelled(
         mock_asyncio_writer.wait_closed.assert_awaited_once_with()
         mock_asyncio_writer.transport.abort.assert_called_once_with()
 
-    async def test____is_closing____return_writer_state(
+    @pytest.mark.parametrize("transport_closed", [False, True], ids=lambda p: f"transport_closed=={p}")
+    async def test____is_closing____return_internal_flag(
         self,
+        transport_closed: bool,
         socket: AsyncioTransportStreamSocketAdapter,
         mock_asyncio_writer: MagicMock,
-        mocker: MockerFixture,
     ) -> None:
         # Arrange
-        mock_asyncio_writer.is_closing.return_value = mocker.sentinel.is_closing
+        if transport_closed:
+            await socket.aclose()
+            mock_asyncio_writer.reset_mock()
+        mock_asyncio_writer.is_closing.side_effect = AssertionError
 
         # Act
         state = socket.is_closing()
 
         # Assert
-        mock_asyncio_writer.is_closing.assert_called_once_with()
-        assert state is mocker.sentinel.is_closing
+        mock_asyncio_writer.is_closing.assert_not_called()
+        assert state is transport_closed
 
     async def test____recv____read_from_reader(
         self,