125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
"""
|
|
ANTI-SIMPLIFICATION: These tests verify internal queue synchronization and end-to-end event loops.
|
|
They MUST NOT be simplified. They ensure that requests hit the AI client, return to the event queue,
|
|
and ultimately end up processed by the GUI render loop.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import patch
|
|
import time
|
|
from src.gui_2 import App
|
|
from src.events import UserRequestEvent
|
|
from src.api_hook_client import ApiHookClient
|
|
|
|
@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.
|
|
ANTI-SIMPLIFICATION: This verifies the full cross-thread boundary.
|
|
"""
|
|
app = mock_app
|
|
# Mock all ai_client methods called during _handle_request_event
|
|
mock_response = "This is a test AI response"
|
|
with (
|
|
patch('src.ai_client.send', return_value=mock_response) as mock_send,
|
|
patch('src.ai_client.set_custom_system_prompt'),
|
|
patch('src.ai_client.set_model_params'),
|
|
patch('src.ai_client.set_agent_tools'),
|
|
patch('src.app_controller.AppController._update_gcli_adapter')
|
|
):
|
|
# 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)
|
|
# But _handle_request_event itself puts a 'response' event in the queue.
|
|
# Our mock_app fixture mocks start_services, so _process_event_queue is NOT running.
|
|
# We need to call it manually or not mock start_services.
|
|
|
|
# Let's call the handler
|
|
app.controller._handle_request_event(event)
|
|
|
|
# 3. Verify ai_client.send was called
|
|
assert mock_send.called, "ai_client.send was not called"
|
|
|
|
# 4. Now the 'response' event is in app.controller.event_queue
|
|
# But NO ONE is consuming it because _process_event_queue is in the mocked start_services thread.
|
|
# Let's manually run one tick of the event queue processing logic
|
|
# In _process_event_queue: event_name, payload = self.event_queue.get()
|
|
event_name, payload = app.controller.event_queue.get()
|
|
assert event_name == "response"
|
|
|
|
# Manually push it to _pending_gui_tasks as _process_event_queue would
|
|
app.controller._pending_gui_tasks.append({
|
|
"action": "handle_ai_response",
|
|
"payload": payload
|
|
})
|
|
|
|
# 5. Process the GUI tasks
|
|
app.controller._process_pending_gui_tasks()
|
|
|
|
assert app.controller.ai_response == mock_response
|
|
assert app.controller.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('src.ai_client.send', side_effect=Exception("API Failure")),
|
|
patch('src.ai_client.set_custom_system_prompt'),
|
|
patch('src.ai_client.set_model_params'),
|
|
patch('src.ai_client.set_agent_tools'),
|
|
patch('src.app_controller.AppController._update_gcli_adapter')
|
|
):
|
|
event = UserRequestEvent(
|
|
prompt="Trigger Error",
|
|
stable_md="",
|
|
file_items=[],
|
|
disc_text="",
|
|
base_dir="."
|
|
)
|
|
app.controller._handle_request_event(event)
|
|
|
|
# Manually consume from queue
|
|
event_name, payload = app.controller.event_queue.get()
|
|
assert event_name == "response"
|
|
assert payload["status"] == "error"
|
|
|
|
# Manually push to GUI tasks
|
|
app.controller._pending_gui_tasks.append({
|
|
"action": "handle_ai_response",
|
|
"payload": payload
|
|
})
|
|
|
|
app.controller._process_pending_gui_tasks()
|
|
assert app.controller.ai_status == "error"
|
|
assert "ERROR: API Failure" in app.controller.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'
|