154 lines
4.9 KiB
Python
154 lines
4.9 KiB
Python
"""
|
|
ANTI-SIMPLIFICATION: These tests verify the end-to-end full live workflow.
|
|
They MUST NOT be simplified. They depend on exact execution states and timing
|
|
through the actual GUI and ApiHookClient interface.
|
|
"""
|
|
import pytest
|
|
import time
|
|
import sys
|
|
import os
|
|
|
|
# Ensure project root is in path
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from src.api_hook_client import ApiHookClient
|
|
|
|
def wait_for_value(client, field, expected, timeout=10):
|
|
"""
|
|
Helper to poll the GUI state until a field matches the expected value.
|
|
"""
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
state = client.get_gui_state()
|
|
val = state.get(field)
|
|
if val == expected:
|
|
return True
|
|
time.sleep(0.5)
|
|
return False
|
|
|
|
@pytest.mark.integration
|
|
def test_full_live_workflow(live_gui) -> None:
|
|
"""
|
|
Integration test that drives the GUI through a full workflow.
|
|
ANTI-SIMPLIFICATION: Asserts exact AI behavior, thinking state tracking,
|
|
and response logging in discussion history.
|
|
"""
|
|
client = ApiHookClient()
|
|
assert client.wait_for_server(timeout=10)
|
|
client.post_session(session_entries=[])
|
|
|
|
# 1. Reset
|
|
print("\n[TEST] Clicking Reset...")
|
|
client.click("btn_reset")
|
|
time.sleep(1)
|
|
|
|
# 2. Project Setup
|
|
temp_project_path = os.path.abspath("tests/artifacts/temp_project.toml")
|
|
if os.path.exists(temp_project_path):
|
|
try: os.remove(temp_project_path)
|
|
except: pass
|
|
print(f"[TEST] Creating new project at {temp_project_path}...")
|
|
client.click("btn_project_new_automated", user_data=temp_project_path)
|
|
|
|
# Wait for project to be active
|
|
success = False
|
|
for _ in range(10):
|
|
proj = client.get_project()
|
|
if proj.get('project', {}).get('project', {}).get('name') == 'temp_project':
|
|
success = True
|
|
break
|
|
time.sleep(1)
|
|
assert success, "Project failed to activate"
|
|
|
|
test_git = os.path.abspath(".")
|
|
print(f"[TEST] Setting project_git_dir to {test_git}...")
|
|
client.set_value("project_git_dir", test_git)
|
|
assert wait_for_value(client, "project_git_dir", test_git)
|
|
|
|
client.click("btn_project_save")
|
|
time.sleep(1)
|
|
|
|
# Enable auto-add so the response ends up in history
|
|
client.set_value("auto_add_history", True)
|
|
client.set_value("current_provider", "gemini")
|
|
|
|
# USE gemini-2.0-flash-lite (Actual current model)
|
|
client.set_value("current_model", "gemini-2.5-flash-lite")
|
|
time.sleep(1)
|
|
|
|
# 3. Discussion Turn
|
|
print("[TEST] Sending AI request...")
|
|
client.set_value("ai_input", "Hello! This is an automated test. Just say 'Acknowledged'.")
|
|
client.click("btn_gen_send")
|
|
|
|
# Verify thinking indicator appears or ai_status changes
|
|
print("[TEST] Polling for thinking indicator...")
|
|
success = False
|
|
for i in range(20):
|
|
mma = client.get_mma_status()
|
|
ai_status = mma.get('ai_status')
|
|
print(f" Poll {i}: ai_status='{ai_status}'")
|
|
if ai_status == 'error':
|
|
state = client.get_gui_state()
|
|
pytest.fail(f"AI Status went to error during thinking poll. Response: {state.get('ai_response')}")
|
|
|
|
if ai_status == 'sending...' or ai_status == 'streaming...':
|
|
print(f" AI is sending/streaming at poll {i}")
|
|
success = True
|
|
# Don't break, keep watching for a bit
|
|
|
|
indicator = client.get_indicator_state("thinking_indicator")
|
|
if indicator.get('shown'):
|
|
print(f" Thinking indicator seen at poll {i}")
|
|
success = True
|
|
break
|
|
time.sleep(0.5)
|
|
|
|
# 4. Wait for response in session
|
|
success = False
|
|
print("[TEST] Waiting for AI response in session history...")
|
|
for i in range(60):
|
|
session = client.get_session()
|
|
entries = session.get('session', {}).get('entries', [])
|
|
# Check for AI role. The entries are objects with a 'role' key.
|
|
found_ai = any(str(e.get('role', '')).upper() == 'AI' for e in entries)
|
|
if found_ai:
|
|
success = True
|
|
print(f" AI response found in history after {i}s")
|
|
break
|
|
|
|
mma = client.get_mma_status()
|
|
if mma.get('ai_status') == 'error':
|
|
state = client.get_gui_state()
|
|
pytest.fail(f"AI Status went to error during response wait. Response: {state.get('ai_response')}")
|
|
|
|
time.sleep(1)
|
|
|
|
# FALLBACK: if not in entries yet, check if ai_response is populated and status is done
|
|
if not success:
|
|
mma = client.get_mma_status()
|
|
if mma.get('ai_status') == 'done' or mma.get('ai_status') == 'idle':
|
|
state = client.get_gui_state()
|
|
if state.get('ai_response'):
|
|
print("[TEST] AI response found in ai_response field (fallback)")
|
|
success = True
|
|
|
|
assert success, f"AI failed to respond. Entries: {client.get_session()}, Status: {client.get_mma_status()}"
|
|
|
|
# 5. Switch Discussion
|
|
print("[TEST] Creating new discussion 'AutoDisc'...")
|
|
client.set_value("disc_new_name_input", "AutoDisc")
|
|
client.click("btn_disc_create")
|
|
time.sleep(1.0)
|
|
|
|
print("[TEST] Switching to 'AutoDisc'...")
|
|
client.select_list_item("disc_listbox", "AutoDisc")
|
|
time.sleep(1.0)
|
|
|
|
# Verify session is empty in new discussion
|
|
session = client.get_session()
|
|
entries = session.get('session', {}).get('entries', [])
|
|
print(f" New discussion history length: {len(entries)}")
|
|
assert len(entries) == 0
|
|
print("[TEST] Workflow completed successfully.")
|