manual_slop/tests/test_gemini_cli_integration.py

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.
    """
    client = ApiHookClient("http://127.0.0.1:8999")
    
    # 1. Setup paths and configure the GUI
    mock_script = os.path.abspath("tests/mock_gemini_cli.py")
    # Wrap in quotes for shell execution if path has spaces
    cli_cmd = f'"{sys.executable}" "{mock_script}"'
    
    # Set provider and binary path via GUI hooks
    # Note: Using set_value which now triggers the property setter in gui_2.py
    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 were applied
    assert client.get_value("current_provider") == "gemini_cli"
    assert client.get_value("gcli_path") == cli_cmd

    # 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 the 'ask_received' event
    print("[TEST] Waiting for ask_received event...")
    request_id = None
    timeout = 30
    start_time = time.time()
    while time.time() - start_time < timeout:
        events = client.get_events()
        if events:
            print(f"[TEST] Received {len(events)} events: {[e.get('type') for e in events]}")
        for ev in events:
            if ev.get("type") == "ask_received":
                request_id = ev.get("request_id")
                print(f"[TEST] Found request_id: {request_id}")
                break
        if request_id:
            break
        time.sleep(0.5)

    assert request_id is not None, "Timed out waiting for 'ask_received' event from the bridge"

    # 4. Respond to the permission request
    print("[TEST] Responding to ask with approval")
    resp = requests.post(
        "http://127.0.0.1:8999/api/ask/respond",
        json={
            "request_id": request_id,
            "response": {"approved": True}
        }
    )
    assert resp.status_code == 200

    # 5. Verify that the final response is displayed in the GUI
    print("[TEST] Waiting for final message in history...")
    final_message_received = False
    start_time = time.time()
    while time.time() - start_time < timeout:
        session = client.get_session()
        entries = session.get("session", {}).get("entries", [])
        for entry in entries:
            content = entry.get("content", "")
            if "Hello from mock!" in content:
                print(f"[TEST] Success! Found message: {content[:50]}...")
                final_message_received = True
                break
        if final_message_received:
            break
        time.sleep(1.0)

    assert final_message_received, "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")
    
    # 1. Setup paths and configure the GUI
    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 that will be denied
    print("[TEST] Sending user message (to be denied)...")
    client.set_value("ai_input", "Deny me")
    client.click("btn_gen_send")
    
    # 3. Wait for 'ask_received' and respond with rejection
    request_id = None
    timeout = 15
    start_time = time.time()
    while time.time() - start_time < timeout:
        for ev in client.get_events():
            if ev.get("type") == "ask_received":
                request_id = ev.get("request_id")
                break
        if request_id: break
        time.sleep(0.5)

    assert request_id is not None

    print("[TEST] Responding to ask with REJECTION")
    requests.post("http://127.0.0.1:8999/api/ask/respond",
                 json={"request_id": request_id, "response": {"approved": False}})

    # 4. Verify rejection message in history
    print("[TEST] Waiting for rejection message in history...")
    rejection_found = False
    start_time = time.time()
    while time.time() - start_time < timeout:
        session = client.get_session()
        entries = session.get("session", {}).get("entries", [])
        for entry in entries:
            if "Tool execution was denied. Decision: deny" 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"

    # 5. Send a follow-up message and verify history grows
    print("[TEST] Sending follow-up message...")
    client.set_value("ai_input", "What happened?")
    client.click("btn_gen_send")
    
    # Wait for mock to finish (it will just return a message)
    time.sleep(2) 
    session = client.get_session()
    entries = session.get("session", {}).get("entries", [])
    # Should have: 
    # 1. User: Deny me
    # 2. AI: Tool execution was denied...
    # 3. User: What happened?
    # 4. AI: ...
    print(f"[TEST] Final history length: {len(entries)}")
    for i, entry in enumerate(entries):
        print(f"  {i}: {entry.get('role')} - {entry.get('content')[:30]}...")
    
    assert len(entries) >= 4