diff --git a/tests/test_event_loop.py b/tests/test_event_loop.py new file mode 100644 index 0000000..f288e71 --- /dev/null +++ b/tests/test_event_loop.py @@ -0,0 +1,228 @@ +"""Tests for event loop management in streaming display.""" + +import asyncio +import pytest +from unittest.mock import Mock, AsyncMock, patch + +from EvoScientist.stream.display import _get_event_loop, _create_event_loop + + +class TestCreateEventLoop: + """Tests for _create_event_loop helper.""" + + def test_creates_new_loop(self): + """Should create a new event loop and set it as current.""" + # Get initial loop (if any) + try: + initial_loop = asyncio.get_event_loop() + initial_loop.close() + except RuntimeError: + pass + + # Create new loop + loop = _create_event_loop() + + assert loop is not None + assert not loop.is_closed() + assert asyncio.get_event_loop() is loop + + # Cleanup + loop.close() + + def test_replaces_closed_loop(self): + """Should replace a closed loop.""" + old_loop = asyncio.new_event_loop() + asyncio.set_event_loop(old_loop) + old_loop.close() + + new_loop = _create_event_loop() + + assert new_loop is not old_loop + assert not new_loop.is_closed() + assert asyncio.get_event_loop() is new_loop + + # Cleanup + new_loop.close() + + +class TestGetEventLoop: + """Tests for _get_event_loop helper.""" + + def test_returns_existing_open_loop(self): + """Should return existing event loop if it's open.""" + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + result = _get_event_loop() + + assert result is loop + assert not result.is_closed() + + # Cleanup + loop.close() + + def test_creates_new_loop_when_closed(self): + """Should create new event loop if current one is closed.""" + old_loop = asyncio.new_event_loop() + asyncio.set_event_loop(old_loop) + old_loop.close() + + result = _get_event_loop() + + assert result is not old_loop + assert not result.is_closed() + + # Cleanup + result.close() + + def test_handles_no_event_loop(self): + """Should handle RuntimeError when no event loop exists (edge case).""" + # This test simulates what happens in a worker thread + # In practice, get_event_loop() returns a closed loop, not RuntimeError + # But we handle the RuntimeError case defensively + loop = _get_event_loop() + assert loop is not None + assert not loop.is_closed() + + # Cleanup + loop.close() + + +class TestMultipleStreamingCalls: + """Tests for the main bug fix: multiple _run_streaming calls.""" + + def test_sequential_streaming_calls(self): + """Multiple sequential calls should work without 'Event loop is closed' error.""" + from EvoScientist.stream.display import _run_streaming + + # Mock agent that returns simple events + mock_agent = Mock() + + async def mock_stream(*args, **kwargs): + """Mock event stream.""" + yield {"type": "text", "content": "test response"} + yield {"type": "done", "response": "test response"} + + # Clean up any existing event loop to start fresh + try: + existing_loop = asyncio.get_event_loop() + if not existing_loop.is_closed(): + existing_loop.close() + except RuntimeError: + pass + + # Patch the stream_agent_events function + with patch('EvoScientist.stream.display.stream_agent_events', side_effect=mock_stream): + # Patch Live to avoid terminal output during tests + with patch('EvoScientist.stream.display.Live'): + # First call + _run_streaming( + agent=mock_agent, + message="test message 1", + thread_id="thread1", + show_thinking=False, + interactive=True, + ) + + # Second call - this would fail with "Event loop is closed" before the fix + _run_streaming( + agent=mock_agent, + message="test message 2", + thread_id="thread1", + show_thinking=False, + interactive=True, + ) + + # Third call for good measure + _run_streaming( + agent=mock_agent, + message="test message 3", + thread_id="thread1", + show_thinking=False, + interactive=True, + ) + + def test_loop_reused_across_calls(self): + """Event loop should be reused across multiple calls.""" + # Create a fresh loop + loop = _create_event_loop() + + # Simulate multiple calls + for _ in range(3): + current_loop = _get_event_loop() + assert not current_loop.is_closed() + + # Run a simple coroutine + async def dummy(): + return "ok" + + result = current_loop.run_until_complete(dummy()) + assert result == "ok" + + # Loop should still be open + assert not loop.is_closed() + + # Cleanup + loop.close() + + def test_closed_loop_recovery(self): + """If loop gets closed, next call should create a new one.""" + # Create and close a loop + loop1 = _create_event_loop() + loop1.close() + + # Next call should detect closed loop and create new one + loop2 = _get_event_loop() + + assert loop2 is not loop1 + assert not loop2.is_closed() + + # Should be able to use the new loop + async def dummy(): + return "success" + + result = loop2.run_until_complete(dummy()) + assert result == "success" + + # Cleanup + loop2.close() + + +class TestEventLoopThreadSafety: + """Tests for thread safety edge cases.""" + + def test_main_thread_normal_case(self): + """Normal case in main thread should work.""" + loop = _get_event_loop() + assert loop is not None + assert not loop.is_closed() + + # Cleanup + loop.close() + + @pytest.mark.skipif( + True, # Skip by default as threading tests can be flaky + reason="Thread test can be flaky in CI" + ) + def test_worker_thread_creates_loop(self): + """Worker thread should be able to create its own loop.""" + import threading + + result = {"loop": None, "error": None} + + def thread_func(): + try: + # In a worker thread, get_event_loop() may raise RuntimeError + # Our code should handle this by creating a new loop + result["loop"] = _get_event_loop() + except Exception as e: + result["error"] = e + + thread = threading.Thread(target=thread_func) + thread.start() + thread.join() + + # Should either get a loop or handle the error gracefully + assert result["error"] is None or isinstance(result["error"], RuntimeError) + if result["loop"]: + assert not result["loop"].is_closed()