ed/manual_slop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP: I HATE PYTHON

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-03-05 13:55:40 -05:00
parent 107608cd76
commit 5e69617f88
43 changed files with 1854 additions and 1671 deletions
Download Patch File Download Diff File Expand all files Collapse all files

235
tests/test_gemini_cli_edge_cases.py
View File

@@ -1,179 +1,64 @@
import time
import os
import sys
import requests
from typing import Any
from api_hook_client import ApiHookClient
import json
import subprocess
from unittest.mock import patch, MagicMock
from src.gemini_cli_adapter import GeminiCliAdapter
from src import mcp_client
def test_gemini_cli_context_bleed_prevention(live_gui: Any) -> None:
"""
Test that the GeminiCliAdapter correctly filters out echoed 'user' messages
and only shows assistant content in the GUI history.
"""
client = ApiHookClient("http://127.0.0.1:8999")
client.click("btn_reset")
_start = time.time()
while time.time() - _start < 8.0:
s = client.get_session()
if not s or not s.get('session', {}).get('entries'):
break
time.sleep(0.2)
client.set_value("auto_add_history", True)
# Create a specialized mock for context bleed
bleed_mock = os.path.abspath("tests/mock_context_bleed.py")
with open(bleed_mock, "w") as f:
f.write('''import sys, json
print(json.dumps({"type": "init", "session_id": "bleed-test"}), flush=True)
print(json.dumps({"type": "message", "role": "user", "content": "I am echoing you"}), flush=True)
print(json.dumps({"type": "message", "role": "assistant", "content": "Actual AI Response"}), flush=True)
print(json.dumps({"type": "result", "stats": {"total_tokens": 10}}), flush=True)
''')
cli_cmd = f'"{sys.executable}" "{bleed_mock}"'
client.set_value("current_provider", "gemini_cli")
client.set_value("gcli_path", cli_cmd)
client.set_value("ai_input", "Test context bleed")
client.click("btn_gen_send")
# Wait for completion
_start = time.time()
while time.time() - _start < 15.0:
s = client.get_session()
if any(e.get('role') == 'AI' for e in s.get('session', {}).get('entries', [])):
break
time.sleep(0.3)
session = client.get_session()
entries = session.get("session", {}).get("entries", [])
# Verify: We expect exactly one AI entry, and it must NOT contain the echoed user message
ai_entries = [e for e in entries if e.get("role") == "AI"]
assert len(ai_entries) == 1
assert ai_entries[0].get("content") == "Actual AI Response"
assert "echoing you" not in ai_entries[0].get("content")
os.remove(bleed_mock)
def test_gemini_cli_parameter_resilience(live_gui: Any) -> None:
"""
Test that mcp_client correctly handles 'file_path' and 'dir_path' aliases
sent by the AI instead of 'path'.
"""
client = ApiHookClient("http://127.0.0.1:8999")
client.click("btn_reset")
time.sleep(1.0)
client.set_value("auto_add_history", True)
client.set_value("manual_approve", True)
client.select_list_item("proj_files", "manual_slop")
# Create a mock that uses dir_path for list_directory
alias_mock = os.path.abspath("tests/mock_alias_tool.py")
bridge_path = os.path.abspath("scripts/cli_tool_bridge.py")
# Avoid backslashes in f-string expression part
if sys.platform == "win32":
bridge_path_str = bridge_path.replace("\\", "/")
else:
bridge_path_str = bridge_path
with open("tests/mock_alias_tool.py", "w") as f:
f.write(f'''import sys, json, os, subprocess
prompt = sys.stdin.read()
if '"role": "tool"' in prompt:
print(json.dumps({{"type": "message", "role": "assistant", "content": "Tool worked!"}}), flush=True)
print(json.dumps({{"type": "result", "stats": {{"total_tokens": 20}}}}), flush=True)
else:
# We must call the bridge to trigger the GUI approval!
tool_call = {{"name": "list_directory", "input": {{"dir_path": "."}}}}
bridge_cmd = [sys.executable, "{bridge_path_str}"]
proc = subprocess.Popen(bridge_cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
stdout, _ = proc.communicate(input=json.dumps(tool_call))
def test_gemini_cli_context_bleed_prevention(monkeypatch) -> None:
"""Test that the GeminiCliAdapter correctly filters out echoed 'user' messages
from the streaming JSON if they were to occur (safety check)."""
adapter = GeminiCliAdapter()
# Even if bridge says allow, we emit the tool_use to the adapter
print(json.dumps({{"type": "message", "role": "assistant", "content": "I will list the directory."}}), flush=True)
print(json.dumps({{
"type": "tool_use",
"name": "list_directory",
"id": "alias_call",
"args": {{"dir_path": "."}}
}}), flush=True)
print(json.dumps({{"type": "result", "stats": {{"total_tokens": 10}}}}), flush=True)
''')
cli_cmd = f'"{sys.executable}" "{alias_mock}"'
client.set_value("current_provider", "gemini_cli")
client.set_value("gcli_path", cli_cmd)
client.set_value("ai_input", "Test parameter aliases")
client.click("btn_gen_send")
# Handle approval
timeout = 60
start_time = time.time()
approved = False
while time.time() - start_time < timeout:
for ev in client.get_events():
etype = ev.get("type")
eid = ev.get("request_id") or ev.get("action_id")
if etype == "ask_received":
requests.post("http://127.0.0.1:8999/api/ask/respond",
json={"request_id": eid, "response": {"approved": True}})
approved = True
elif etype == "script_confirmation_required":
requests.post(f"http://127.0.0.1:8999/api/confirm/{eid}", json={"approved": True})
approved = True
if approved: break
time.sleep(0.5)
assert approved, "Tool approval event never received"
# Verify tool result in history
time.sleep(2)
session = client.get_session()
entries = session.get("session", {}).get("entries", [])
# Check for "Tool worked!" which implies the tool execution was successful
found = any("Tool worked!" in e.get("content", "") for e in entries)
assert found, "Tool result indicating success not found in history"
os.remove(alias_mock)
mock_process = MagicMock()
# Simulate a stream that includes a message from 'user' (should be ignored)
# and a message from 'model'.
mock_process.stdout = [
b'{"kind": "message", "role": "user", "payload": "Echoed user prompt"}\n',
b'{"kind": "message", "role": "model", "payload": "Model response"}\n'
]
mock_process.stderr = []
mock_process.returncode = 0
with patch('subprocess.Popen', return_value=mock_process):
result = adapter.send("msg")
# Should only contain the model response
assert result["text"] == "Model response"
def test_gemini_cli_loop_termination(live_gui: Any) -> None:
"""
Test that multi-round tool calling correctly terminates and preserves
payload (session context) between rounds.
"""
client = ApiHookClient("http://127.0.0.1:8999")
client.click("btn_reset")
time.sleep(1.0)
client.set_value("auto_add_history", True)
client.set_value("manual_approve", True)
client.select_list_item("proj_files", "manual_slop")
# This uses the existing mock_gemini_cli.py which is already designed for 2 rounds
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)
client.set_value("ai_input", "Perform multi-round tool test")
client.click("btn_gen_send")
# Handle approvals (mock does one tool call)
timeout = 60
start_time = time.time()
approved = False
while time.time() - start_time < timeout:
for ev in client.get_events():
etype = ev.get("type")
eid = ev.get("request_id") or ev.get("action_id")
if etype == "ask_received":
requests.post("http://127.0.0.1:8999/api/ask/respond",
json={"request_id": eid, "response": {"approved": True}})
approved = True
elif etype == "script_confirmation_required":
requests.post(f"http://127.0.0.1:8999/api/confirm/{eid}", json={"approved": True})
approved = True
if approved: break
time.sleep(0.5)
# Wait for the second round and final answer
found_final = False
start_time = time.time()
while time.time() - start_time < 30:
session = client.get_session()
entries = session.get("session", {}).get("entries", [])
print(f"DEBUG: Session entries: {[e.get('content', '')[:30] for e in entries]}")
for e in entries:
content = e.get("content", "")
success_markers = ["processed the tool results", "Here are the files", "Here are the lines", "Script hello.ps1 created successfully"]
if any(marker in content for marker in success_markers):
found_final = True
break
if found_final: break
time.sleep(1)
assert found_final, "Final message after multi-round tool loop not found"
def test_gemini_cli_parameter_resilience() -> None:
"""Test that mcp_client correctly handles 'file_path' and 'dir_path' aliases
if the AI provides them instead of 'path'."""
from src import mcp_client
# Mock dispatch to see what it receives
with patch('src.mcp_client.read_file', return_value="content") as mock_read:
mcp_client.dispatch("read_file", {"file_path": "aliased.txt"})
mock_read.assert_called_once_with("aliased.txt")
with patch('src.mcp_client.list_directory', return_value="files") as mock_list:
mcp_client.dispatch("list_directory", {"dir_path": "aliased_dir"})
mock_list.assert_called_once_with("aliased_dir")
def test_gemini_cli_loop_termination() -> None:
"""Test that multi-round tool calling correctly terminates and preserves
the final text."""
from src import ai_client
ai_client.set_provider("gemini_cli", "gemini-2.0-flash")
# Round 1: Tool call
mock_resp1 = {"text": "Calling tool", "tool_calls": [{"name": "read_file", "args": {"path": "f.txt"}}]}
# Round 2: Final response
mock_resp2 = {"text": "Final answer", "tool_calls": []}
with patch('src.ai_client.GeminiCliAdapter') as MockAdapter:
instance = MockAdapter.return_value
instance.send.side_effect = [mock_resp1, mock_resp2]
instance.last_usage = {"total_tokens": 10}
instance.last_latency = 0.1
instance.session_id = "s1"
# We need to mock mcp_client.dispatch too
with patch('src.mcp_client.dispatch', return_value="content"):
result = ai_client.send("context", "prompt")
assert result == "Final answer"
assert instance.send.call_count == 2