fix

Author: Lendemor

reflex/app.py

@@ -2140,12 +2140,3 @@ 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,15 +313,6 @@ 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):
@@ -385,6 +376,17 @@ async def _reset_backend_state_manager(self):
                 msg = "Failed to reset state manager."
                 raise RuntimeError(msg)
 
+            # Also reset the TokenManager to avoid loop affinity issues
+            if (
+                hasattr(self.app_instance, "event_namespace")
+                and self.app_instance.event_namespace is not None
+                and hasattr(self.app_instance.event_namespace, "_token_manager")
+            ):
+                # Import here to avoid circular imports
+                from reflex.utils.token_manager import TokenManager
+
+                self.app_instance.event_namespace._token_manager = TokenManager.create()
+
     def _start_frontend(self):
         # Set up the frontend.
         with chdir(self.app_path):

reflex/utils/token_manager.py

@@ -52,13 +52,6 @@ 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.
@@ -114,13 +107,6 @@ 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.
@@ -229,17 +215,3 @@ 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,15 +174,6 @@ 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."""
@@ -411,28 +402,3 @@ 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)