fix closed loop

Author: Lendemor

reflex/app.py

@@ -2140,3 +2140,12 @@ async def link_token_to_sid(self, sid: str, token: str):
         if new_token:
             # Duplicate detected, emit new token to client
             await self.emit("new_token", new_token, to=sid)
+
+    async def close(self) -> None:
+        """Close any resources used by the event namespace.
+
+        This is necessary in testing scenarios to close between asyncio test cases
+        to avoid having lingering connections associated with event loops.
+        """
+        if hasattr(self, "_token_manager"):
+            await self._token_manager.close()

reflex/testing.py

@@ -313,6 +313,15 @@ async def _shutdown(*args, **kwargs) -> None:
                 with contextlib.suppress(ValueError):
                     await self.app_instance._state_manager.close()
 
+            # ensure token manager redis connections are closed
+            if (
+                self.app_instance is not None
+                and hasattr(self.app_instance, "event_namespace")
+                and self.app_instance.event_namespace is not None
+            ):
+                with contextlib.suppress(Exception):
+                    await self.app_instance.event_namespace.close()
+
             # socketio shutdown handler
             if self.app_instance is not None and self.app_instance.sio is not None:
                 with contextlib.suppress(TypeError):

reflex/utils/token_manager.py

@@ -52,6 +52,13 @@ async def disconnect_token(self, token: str, sid: str) -> None:
             sid: The Socket.IO session ID.
         """
 
+    @abstractmethod
+    async def close(self) -> None:
+        """Close any resources used by the token manager.
+
+        Subclasses must implement this method.
+        """
+
     @classmethod
     def create(cls) -> TokenManager:
         """Factory method to create appropriate TokenManager implementation.
@@ -107,6 +114,13 @@ async def disconnect_token(self, token: str, sid: str) -> None:
         self.token_to_sid.pop(token, None)
         self.sid_to_token.pop(sid, None)
 
+    async def close(self) -> None:
+        """Close any resources used by the token manager.
+
+        LocalTokenManager has no resources to close.
+        """
+        # No resources to clean up for local manager
+
 
 class RedisTokenManager(LocalTokenManager):
     """Token manager using Redis for distributed multi-worker support.
@@ -215,3 +229,17 @@ async def disconnect_token(self, token: str, sid: str) -> None:
 
             # Clean up local dicts (always do this)
             await super().disconnect_token(token, sid)
+
+    async def close(self) -> None:
+        """Close Redis connection to prevent closed loop errors in tests.
+
+        It is necessary in testing scenarios to close between asyncio test cases
+        to avoid having lingering redis connections associated with event loops
+        that will be closed (each test case uses its own event loop).
+
+        Note: Connections will be automatically reopened when needed.
+        """
+        try:
+            await self.redis.aclose(close_connection_pool=True)
+        except Exception as e:
+            console.error(f"Redis close error: {e}")

tests/units/utils/test_token_manager.py

@@ -174,6 +174,15 @@ async def test_disconnect_nonexistent_token(self, manager):
         assert len(manager.token_to_sid) == 0
         assert len(manager.sid_to_token) == 0
 
+    async def test_close_does_nothing(self, manager):
+        """Test close method does nothing for LocalTokenManager.
+
+        Args:
+            manager: LocalTokenManager fixture instance.
+        """
+        await manager.close()
+        # Should not raise any errors
+
 
 class TestRedisTokenManager:
     """Tests for RedisTokenManager."""
@@ -402,3 +411,28 @@ def test_inheritance_from_local_manager(self, manager):
         assert isinstance(manager, LocalTokenManager)
         assert hasattr(manager, "token_to_sid")
         assert hasattr(manager, "sid_to_token")
+
+    async def test_close_calls_redis_aclose(self, manager, mock_redis):
+        """Test close method calls Redis aclose with proper parameters.
+
+        Args:
+            manager: RedisTokenManager fixture instance.
+            mock_redis: Mock Redis client fixture.
+        """
+        await manager.close()
+
+        mock_redis.aclose.assert_called_once_with(close_connection_pool=True)
+
+    async def test_close_handles_redis_error(self, manager, mock_redis):
+        """Test close method handles Redis errors gracefully.
+
+        Args:
+            manager: RedisTokenManager fixture instance.
+            mock_redis: Mock Redis client fixture.
+        """
+        mock_redis.aclose.side_effect = Exception("Redis close error")
+
+        # Should not raise an error
+        await manager.close()
+
+        mock_redis.aclose.assert_called_once_with(close_connection_pool=True)