123 lines
4.1 KiB
Python
123 lines
4.1 KiB
Python
from typing import Generator
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch, AsyncMock, ANY
|
|
import asyncio
|
|
import time
|
|
from gui_2 import App
|
|
from events import UserRequestEvent
|
|
import ai_client
|
|
|
|
@pytest.fixture
|
|
def mock_app() -> Generator[App, None, None]:
|
|
with (
|
|
patch('gui_2.load_config', return_value={
|
|
"ai": {"provider": "gemini", "model": "model-1", "temperature": 0.0, "max_tokens": 100, "history_trunc_limit": 1000},
|
|
"projects": {"paths": [], "active": ""},
|
|
"gui": {"show_windows": {}}
|
|
}),
|
|
patch('gui_2.project_manager.load_project', return_value={
|
|
"project": {"name": "test_proj"},
|
|
"discussion": {"active": "main", "discussions": {"main": {"history": []}}},
|
|
"files": {"paths": [], "base_dir": "."},
|
|
"screenshots": {"paths": [], "base_dir": "."},
|
|
"agent": {"tools": {}}
|
|
}),
|
|
patch('gui_2.project_manager.migrate_from_legacy_config', return_value={}),
|
|
patch('gui_2.project_manager.save_project'),
|
|
patch('gui_2.session_logger.open_session'),
|
|
patch('gui_2.App._init_ai_and_hooks'),
|
|
patch('gui_2.App._fetch_models')
|
|
):
|
|
app = App()
|
|
yield app
|
|
# We don't have a clean way to stop the loop thread in gui_2.py App
|
|
# so we just let it daemon-exit.
|
|
|
|
@pytest.mark.timeout(10)
|
|
def test_user_request_integration_flow(mock_app: App) -> None:
|
|
"""
|
|
Verifies that pushing a UserRequestEvent to the event_queue:
|
|
1. Triggers ai_client.send
|
|
2. Results in a 'response' event back to the queue
|
|
3. Eventually updates the UI state (ai_response, ai_status) after processing GUI tasks.
|
|
"""
|
|
app = mock_app
|
|
# Mock all ai_client methods called during _handle_request_event
|
|
mock_response = "This is a test AI response"
|
|
with (
|
|
patch('ai_client.send', return_value=mock_response) as mock_send,
|
|
patch('ai_client.set_custom_system_prompt'),
|
|
patch('ai_client.set_model_params'),
|
|
patch('ai_client.set_agent_tools')
|
|
):
|
|
# 1. Create and push a UserRequestEvent
|
|
event = UserRequestEvent(
|
|
prompt="Hello AI",
|
|
stable_md="Context",
|
|
file_items=[],
|
|
disc_text="History",
|
|
base_dir="."
|
|
)
|
|
# 2. Push event to the app's internal loop
|
|
asyncio.run_coroutine_threadsafe(
|
|
app.event_queue.put("user_request", event),
|
|
app._loop
|
|
)
|
|
# 3. Wait for ai_client.send to be called (polling background thread)
|
|
start_time = time.time()
|
|
while not mock_send.called and time.time() - start_time < 5:
|
|
time.sleep(0.1)
|
|
assert mock_send.called, "ai_client.send was not called within timeout"
|
|
mock_send.assert_called_once_with(
|
|
"Context", "Hello AI", ".", [], "History",
|
|
pre_tool_callback=ANY,
|
|
qa_callback=ANY
|
|
)
|
|
# 4. Wait for the response to propagate to _pending_gui_tasks and update UI
|
|
# We call _process_pending_gui_tasks manually to simulate a GUI frame update.
|
|
start_time = time.time()
|
|
success = False
|
|
while time.time() - start_time < 3:
|
|
app._process_pending_gui_tasks()
|
|
if app.ai_response == mock_response and app.ai_status == "done":
|
|
success = True
|
|
break
|
|
time.sleep(0.1)
|
|
assert success, f"UI state was not updated. ai_response: '{app.ai_response}', status: '{app.ai_status}'"
|
|
assert app.ai_response == mock_response
|
|
assert app.ai_status == "done"
|
|
|
|
@pytest.mark.timeout(10)
|
|
def test_user_request_error_handling(mock_app: App) -> None:
|
|
"""
|
|
Verifies that if ai_client.send raises an exception, the UI is updated with the error state.
|
|
"""
|
|
app = mock_app
|
|
with (
|
|
patch('ai_client.send', side_effect=Exception("API Failure")) as mock_send,
|
|
patch('ai_client.set_custom_system_prompt'),
|
|
patch('ai_client.set_model_params'),
|
|
patch('ai_client.set_agent_tools')
|
|
):
|
|
event = UserRequestEvent(
|
|
prompt="Trigger Error",
|
|
stable_md="",
|
|
file_items=[],
|
|
disc_text="",
|
|
base_dir="."
|
|
)
|
|
asyncio.run_coroutine_threadsafe(
|
|
app.event_queue.put("user_request", event),
|
|
app._loop
|
|
)
|
|
# Poll for error state by processing GUI tasks
|
|
start_time = time.time()
|
|
success = False
|
|
while time.time() - start_time < 5:
|
|
app._process_pending_gui_tasks()
|
|
if app.ai_status == "error" and "ERROR: API Failure" in app.ai_response:
|
|
success = True
|
|
break
|
|
time.sleep(0.1)
|
|
assert success, f"Error state was not reflected in UI. status: {app.ai_status}, response: {app.ai_response}"
|