95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
import pytest
|
|
import time
|
|
import sys
|
|
import os
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src")))
|
|
|
|
from src import api_hook_client
|
|
|
|
@pytest.mark.integration
|
|
def test_phase4_final_verify(live_gui):
|
|
client = api_hook_client.ApiHookClient()
|
|
assert client.wait_for_server(timeout=15), "Hook server did not start"
|
|
|
|
# 1. Setup mock project data
|
|
workspace_dir = Path("tests/artifacts/live_gui_workspace")
|
|
workspace_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create dummy files
|
|
(workspace_dir / "final_test_1.txt").write_text("Manual Slop RAG is great.")
|
|
(workspace_dir / "final_test_2.py").write_text("def test_func():\n return 'Manual Slop RAG result'")
|
|
|
|
try:
|
|
# 2. Configure project through Hook API
|
|
client.set_value('files', ['final_test_1.txt', 'final_test_2.py'])
|
|
client.set_value('rag_enabled', True)
|
|
client.set_value('rag_source', 'chroma')
|
|
client.set_value('rag_emb_provider', 'local')
|
|
client.set_value('auto_add_history', True)
|
|
client.set_value('current_provider', 'gemini_cli')
|
|
client.set_value('gcli_path', os.path.abspath(os.path.join(os.path.dirname(__file__), "mock_gcli.bat")))
|
|
|
|
# 3. Trigger Initial Indexing
|
|
print("[VERIFY] Triggering indexing...")
|
|
client.click('btn_rebuild_rag_index')
|
|
|
|
# Wait for ready
|
|
success = False
|
|
for _ in range(50):
|
|
status = client.get_value('rag_status')
|
|
if status == 'ready':
|
|
success = True
|
|
break
|
|
time.sleep(0.5)
|
|
assert success, f"Indexing failed. Status: {status}"
|
|
|
|
# 4. Verify Retrieval and Visualization
|
|
print("[VERIFY] Triggering retrieval turn...")
|
|
client.set_value('ai_input', "What makes RAG great?")
|
|
client.click('btn_gen_send')
|
|
|
|
# Wait for completion
|
|
success = False
|
|
for _ in range(50):
|
|
state = client.get_gui_state()
|
|
if state.get('ai_status') == 'done':
|
|
success = True
|
|
break
|
|
time.sleep(0.5)
|
|
assert success, "AI request timed out"
|
|
|
|
# 5. Verify discussion history has the context
|
|
session = client.get_session()
|
|
entries = session.get('session', {}).get('entries', [])
|
|
|
|
found_rag = False
|
|
for entry in entries:
|
|
if entry.get('role') == 'User' and '## Retrieved Context' in entry.get('content', ''):
|
|
found_rag = True
|
|
print(f"[VERIFY] Found RAG context: {entry.get('content')[:100]}...")
|
|
assert "Manual Slop RAG is great" in entry.get('content')
|
|
break
|
|
assert found_rag, "RAG context not found in history"
|
|
|
|
# 6. Verify Incremental Indexing (no changes)
|
|
print("[VERIFY] Verifying incrementality...")
|
|
start = time.time()
|
|
client.click('btn_rebuild_rag_index')
|
|
for _ in range(50):
|
|
if client.get_value('rag_status') == 'ready': break
|
|
time.sleep(0.1)
|
|
duration = time.time() - start
|
|
print(f"[VERIFY] Incremental indexing took {duration:.2f}s")
|
|
assert duration < 1.0, "Incremental indexing too slow (expected < 1s for 2 files)"
|
|
|
|
print("[VERIFY] Phase 4 final verification COMPLETED successfully.")
|
|
|
|
except Exception as e:
|
|
print(f"[VERIFY] ERROR in final verification: {e}")
|
|
raise
|