chore(tests): Refine test separation, keep feature tests in main tests folder
This commit is contained in:
141
tests/test_gemini_cli_integration.py
Normal file
141
tests/test_gemini_cli_integration.py
Normal file
@@ -0,0 +1,141 @@
|
||||
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"
|
||||
Reference in New Issue
Block a user