142 lines
5.2 KiB
Python
142 lines
5.2 KiB
Python
import pytest
|
|
import time
|
|
import os
|
|
import sys
|
|
import requests
|
|
from api_hook_client import ApiHookClient
|
|
|
|
def test_gemini_cli_full_integration(live_gui):
|
|
"""
|
|
Integration test for the Gemini CLI provider and tool bridge.
|
|
Handles 'ask_received' events from the bridge and any other approval requests.
|
|
"""
|
|
client = ApiHookClient("http://127.0.0.1:8999")
|
|
|
|
# 0. Reset session and enable history
|
|
client.click("btn_reset")
|
|
client.set_value("auto_add_history", True)
|
|
# Switch to manual_slop project explicitly
|
|
client.select_list_item("proj_files", "manual_slop")
|
|
|
|
# 1. Setup paths and configure the GUI
|
|
# Use the real gemini CLI if available, otherwise use mock
|
|
# For CI/testing we prefer mock
|
|
mock_script = os.path.abspath("tests/mock_gemini_cli.py")
|
|
cli_cmd = f'"{sys.executable}" "{mock_script}"'
|
|
|
|
print(f"[TEST] Setting current_provider to gemini_cli")
|
|
client.set_value("current_provider", "gemini_cli")
|
|
print(f"[TEST] Setting gcli_path to {cli_cmd}")
|
|
client.set_value("gcli_path", cli_cmd)
|
|
|
|
# Verify settings
|
|
assert client.get_value("current_provider") == "gemini_cli"
|
|
|
|
# Clear events
|
|
client.get_events()
|
|
|
|
# 2. Trigger a message in the GUI
|
|
print("[TEST] Sending user message...")
|
|
client.set_value("ai_input", "Please read test.txt")
|
|
client.click("btn_gen_send")
|
|
|
|
# 3. Monitor for approval events
|
|
print("[TEST] Waiting for approval events...")
|
|
timeout = 45
|
|
start_time = time.time()
|
|
approved_count = 0
|
|
|
|
while time.time() - start_time < timeout:
|
|
events = client.get_events()
|
|
if events:
|
|
for ev in events:
|
|
etype = ev.get("type")
|
|
eid = ev.get("request_id") or ev.get("action_id")
|
|
print(f"[TEST] Received event: {etype} (ID: {eid})")
|
|
|
|
if etype in ["ask_received", "glob_approval_required", "script_confirmation_required"]:
|
|
print(f"[TEST] Approving {etype} {eid}")
|
|
if etype == "script_confirmation_required":
|
|
resp = requests.post(f"http://127.0.0.1:8999/api/confirm/{eid}", json={"approved": True})
|
|
else:
|
|
resp = requests.post("http://127.0.0.1:8999/api/ask/respond",
|
|
json={"request_id": eid, "response": {"approved": True}})
|
|
assert resp.status_code == 200
|
|
approved_count += 1
|
|
|
|
# Check if we got a final response in history
|
|
session = client.get_session()
|
|
entries = session.get("session", {}).get("entries", [])
|
|
found_final = False
|
|
for entry in entries:
|
|
content = entry.get("content", "")
|
|
if "Hello from mock!" in content or "processed the tool results" in content:
|
|
print(f"[TEST] Success! Found final message in history.")
|
|
found_final = True
|
|
break
|
|
|
|
if found_final:
|
|
break
|
|
|
|
time.sleep(1.0)
|
|
|
|
assert approved_count > 0, "No approval events were processed"
|
|
assert found_final, "Final message from mock CLI was not found in the GUI history"
|
|
|
|
def test_gemini_cli_rejection_and_history(live_gui):
|
|
"""
|
|
Integration test for the Gemini CLI provider: Rejection flow and history.
|
|
"""
|
|
client = ApiHookClient("http://127.0.0.1:8999")
|
|
|
|
# 0. Reset session
|
|
client.click("btn_reset")
|
|
client.set_value("auto_add_history", True)
|
|
client.select_list_item("proj_files", "manual_slop")
|
|
|
|
mock_script = os.path.abspath("tests/mock_gemini_cli.py")
|
|
cli_cmd = f'"{sys.executable}" "{mock_script}"'
|
|
client.set_value("current_provider", "gemini_cli")
|
|
client.set_value("gcli_path", cli_cmd)
|
|
|
|
# 2. Trigger a message
|
|
print("[TEST] Sending user message (to be denied)...")
|
|
client.set_value("ai_input", "Deny me")
|
|
client.click("btn_gen_send")
|
|
|
|
# 3. Wait for event and reject
|
|
timeout = 20
|
|
start_time = time.time()
|
|
denied = False
|
|
while time.time() - start_time < timeout:
|
|
for ev in client.get_events():
|
|
etype = ev.get("type")
|
|
eid = ev.get("request_id")
|
|
print(f"[TEST] Received event: {etype}")
|
|
if etype == "ask_received":
|
|
print(f"[TEST] Denying request {eid}")
|
|
requests.post("http://127.0.0.1:8999/api/ask/respond",
|
|
json={"request_id": eid, "response": {"approved": False}})
|
|
denied = True
|
|
break
|
|
if denied: break
|
|
time.sleep(0.5)
|
|
|
|
assert denied, "No ask_received event to deny"
|
|
|
|
# 4. Verify rejection in history
|
|
print("[TEST] Waiting for rejection in history...")
|
|
rejection_found = False
|
|
start_time = time.time()
|
|
while time.time() - start_time < 20:
|
|
session = client.get_session()
|
|
entries = session.get("session", {}).get("entries", [])
|
|
for entry in entries:
|
|
if "Tool execution was denied" in entry.get("content", ""):
|
|
rejection_found = True
|
|
break
|
|
if rejection_found: break
|
|
time.sleep(1.0)
|
|
|
|
assert rejection_found, "Rejection message not found in history"
|