110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
import pytest
|
|
from unittest.mock import patch, ANY
|
|
import asyncio
|
|
import time
|
|
from gui_2 import App
|
|
from events import UserRequestEvent
|
|
from api_hook_client import ApiHookClient
|
|
|
|
@pytest.mark.timeout(10)
|
|
@pytest.mark.asyncio
|
|
async 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. Call the handler directly since start_services is mocked (no event loop thread)
|
|
app.controller._handle_request_event(event)
|
|
# 3. Verify ai_client.send was called
|
|
assert mock_send.called, "ai_client.send was not called"
|
|
mock_send.assert_called_once_with(
|
|
"Context", "Hello AI", ".", [], "History",
|
|
pre_tool_callback=ANY,
|
|
qa_callback=ANY,
|
|
stream=ANY,
|
|
stream_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
|
|
await asyncio.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)
|
|
@pytest.mark.asyncio
|
|
async 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")),
|
|
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="."
|
|
)
|
|
app.controller._handle_request_event(event)
|
|
# 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
|
|
await asyncio.sleep(0.1)
|
|
assert success, f"Error state was not reflected in UI. status: {app.ai_status}, response: {app.ai_response}"
|
|
|
|
def test_api_gui_state_live(live_gui) -> None:
|
|
client = ApiHookClient()
|
|
client.set_value('current_provider', 'anthropic')
|
|
client.set_value('current_model', 'claude-3-haiku-20240307')
|
|
|
|
start_time = time.time()
|
|
success = False
|
|
while time.time() - start_time < 10:
|
|
state = client.get_gui_state()
|
|
if state and state.get('current_provider') == 'anthropic' and state.get('current_model') == 'claude-3-haiku-20240307':
|
|
success = True
|
|
break
|
|
time.sleep(0.5)
|
|
|
|
assert success, f"GUI state did not update. Got: {client.get_gui_state()}"
|
|
final_state = client.get_gui_state()
|
|
assert final_state['current_provider'] == 'anthropic'
|
|
assert final_state['current_model'] == 'claude-3-haiku-20240307'
|