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

never ends

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-03-05 20:39:56 -05:00
parent 2c5476dc5d
commit d2481b2de7
7 changed files with 129 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
tests/test_gemini_cli_adapter_parity.py
View File

@@ -5,50 +5,46 @@ from unittest.mock import patch, MagicMock
from src.gemini_cli_adapter import GeminiCliAdapter
class TestGeminiCliAdapterParity(unittest.TestCase):
def setUp(self) -> None:
self.adapter = GeminiCliAdapter(binary_path="gemini")
def setUp(self) -> None:
self.adapter = GeminiCliAdapter(binary_path="gemini")
def tearDown(self) -> None:
pass
def tearDown(self) -> None:
pass
def test_count_tokens_fallback(self) -> None:
contents = ["Hello", "world!"]
estimated = self.adapter.count_tokens(contents)
self.assertEqual(estimated, 3)
def test_count_tokens_fallback(self) -> None:
contents = ["Hello", "world!"]
estimated = self.adapter.count_tokens(contents)
self.assertEqual(estimated, 3)
@patch('src.gemini_cli_adapter.subprocess.Popen')
def test_send_starts_subprocess_with_model(self, mock_popen: MagicMock) -> None:
mock_process = MagicMock()
mock_process.communicate.return_value = ('{"type": "message", "content": "hi"}', '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
self.adapter.send("test", model="gemini-2.0-flash")
args, _ = mock_popen.call_args
cmd_list = args[0]
self.assertIn("-m", cmd_list)
self.assertIn("gemini-2.0-flash", cmd_list)
@patch('src.gemini_cli_adapter.subprocess.Popen')
def test_send_starts_subprocess_with_model(self, mock_popen: MagicMock) -> None:
mock_process = MagicMock()
mock_process.communicate.return_value = ('{"type": "message", "content": "hi"}', '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
self.adapter.send("test", model="gemini-2.0-flash")
args, _ = mock_popen.call_args
cmd_list = args[0]
self.assertIn("-m", cmd_list)
self.assertIn("gemini-2.0-flash", cmd_list)
@patch('src.gemini_cli_adapter.subprocess.Popen')
def test_send_parses_tool_calls_from_streaming_json(self, mock_popen: MagicMock) -> None:
tool_call_json = {
"type": "tool_use",
"tool_name": "list_directory",
"parameters": {"path": "."},
"tool_id": "call_abc"
}
mock_process = MagicMock()
stdout_output = (
json.dumps(tool_call_json) + "\n" +
'{"type": "message", "content": "I listed the files."}'
)
mock_process.communicate.return_value = (stdout_output, '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = self.adapter.send("msg")
self.assertEqual(len(result["tool_calls"]), 1)
self.assertEqual(result["tool_calls"][0]["name"], "list_directory")
self.assertEqual(result["text"], "I listed the files.")
@patch('src.gemini_cli_adapter.subprocess.Popen')
def test_send_parses_tool_calls_from_streaming_json(self, mock_popen: MagicMock) -> None:
tool_call_json = {
"type": "tool_use",
"tool_name": "list_directory",
"parameters": {"path": "."},
"tool_id": "call_abc"
}
mock_process = MagicMock()
stdout_output = (
json.dumps(tool_call_json) + "\n" +
'{"type": "message", "content": "I listed the files."}'
)
mock_process.communicate.return_value = (stdout_output, '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = self.adapter.send("msg")
self.assertEqual(len(result["tool_calls"]), 1)
self.assertEqual(result["tool_calls"][0]["name"], "list_directory")
self.assertEqual(result["text"], "I listed the files.")

70
tests/test_gemini_cli_edge_cases.py
View File

@@ -5,46 +5,38 @@ from src.gemini_cli_adapter import GeminiCliAdapter
from src import mcp_client
def test_gemini_cli_context_bleed_prevention() -> None:
import src.ai_client as ai_client
ai_client._gemini_cli_adapter = None
with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen:
adapter = GeminiCliAdapter()
mock_process = MagicMock()
stdout_output = (
'{"type": "message", "role": "user", "content": "Echoed user prompt"}' + "\n" +
'{"type": "message", "role": "model", "content": "Model response"}'
)
mock_process.communicate.return_value = (stdout_output, '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = adapter.send("msg")
assert result["text"] == "Model response"
import src.ai_client as ai_client
ai_client._gemini_cli_adapter = None
with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen:
adapter = GeminiCliAdapter()
mock_process = MagicMock()
stdout_output = (
'{"type": "message", "role": "user", "content": "Echoed user prompt"}' + "\n" +
'{"type": "message", "role": "model", "content": "Model response"}'
)
mock_process.communicate.return_value = (stdout_output, '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
result = adapter.send("msg")
assert result["text"] == "Model response"
def test_gemini_cli_parameter_resilience() -> None:
from src import mcp_client
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")
from src import mcp_client
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:
import src.ai_client as ai_client
ai_client._gemini_cli_adapter = None
with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen:
mock_process = MagicMock()
mock_process.communicate.return_value = ('{"type": "message", "content": "Final answer", "tool_calls": []}', "")
mock_process.returncode = 0
mock_popen.return_value = mock_process
ai_client.set_provider("gemini_cli", "gemini-2.0-flash")
result = ai_client.send("context", "prompt")
assert result == "Final answer"
import src.ai_client as ai_client
ai_client._gemini_cli_adapter = None
with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen:
mock_process = MagicMock()
mock_process.communicate.return_value = ('{"type": "message", "content": "Final answer", "tool_calls": []}', "")
mock_process.returncode = 0
mock_popen.return_value = mock_process
ai_client.set_provider("gemini_cli", "gemini-2.0-flash")
result = ai_client.send("context", "prompt")
assert result == "Final answer"

46
tests/test_gemini_cli_parity_regression.py
View File

@@ -2,30 +2,26 @@ from typing import Any
from unittest.mock import patch, MagicMock
def test_send_invokes_adapter_send() -> None:
import src.ai_client as ai_client
ai_client._gemini_cli_adapter = None
with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen:
mock_process = MagicMock()
mock_process.communicate.return_value = ('{"type": "message", "content": "Hello from mock adapter"}', '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
ai_client.set_provider("gemini_cli", "gemini-2.0-flash")
res = ai_client.send("context", "msg")
assert res == "Hello from mock adapter"
import src.ai_client as ai_client
ai_client._gemini_cli_adapter = None
with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen:
mock_process = MagicMock()
mock_process.communicate.return_value = ('{"type": "message", "content": "Hello from mock adapter"}', '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
ai_client.set_provider("gemini_cli", "gemini-2.0-flash")
res = ai_client.send("context", "msg")
assert res == "Hello from mock adapter"
def test_get_history_bleed_stats() -> None:
import src.ai_client as ai_client
ai_client._gemini_cli_adapter = None
with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen:
mock_process = MagicMock()
mock_process.communicate.return_value = ('{"type": "message", "content": "txt"}', '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
ai_client.set_provider("gemini_cli", "gemini-2.0-flash")
ai_client.send("context", "msg")
stats = ai_client.get_history_bleed_stats()
assert stats["provider"] == "gemini_cli"
import src.ai_client as ai_client
ai_client._gemini_cli_adapter = None
with patch('src.gemini_cli_adapter.subprocess.Popen') as mock_popen:
mock_process = MagicMock()
mock_process.communicate.return_value = ('{"type": "message", "content": "txt"}', '')
mock_process.returncode = 0
mock_popen.return_value = mock_process
ai_client.set_provider("gemini_cli", "gemini-2.0-flash")
ai_client.send("context", "msg")
stats = ai_client.get_history_bleed_stats()
assert stats["provider"] == "gemini_cli"

70
tests/test_token_usage.py
View File

@@ -10,43 +10,33 @@ sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src import ai_client
def test_token_usage_tracking() -> None:
ai_client.reset_session()
with patch("src.ai_client._ensure_gemini_client"), \
patch("src.ai_client._gemini_client") as mock_client:
mock_chat = MagicMock()
mock_client.chats.create.return_value = mock_chat
mock_usage = SimpleNamespace(
prompt_token_count=100,
candidates_token_count=50,
total_token_count=150,
cached_content_token_count=20
)
mock_part = SimpleNamespace(text="Mock Response", function_call=None)
mock_content = SimpleNamespace(parts=[mock_part])
mock_candidate = SimpleNamespace()
mock_candidate.content = mock_content
mock_candidate.finish_reason = SimpleNamespace(name="STOP")
mock_response = SimpleNamespace()
mock_response.candidates = [mock_candidate]
mock_response.usage_metadata = mock_usage
mock_response.text = "Mock Response"
mock_chat.send_message.return_value = mock_response
ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
ai_client.send("Context", "Hello")
comms = ai_client.get_comms_log()
response_entries = [e for e in comms if e.get("direction") == "IN" and e["kind"] == "response"]
assert len(response_entries) > 0
usage = response_entries[0]["payload"]["usage"]
assert usage["input_tokens"] == 100
assert usage["output_tokens"] == 50
assert usage["cache_read_input_tokens"] == 20
ai_client.reset_session()
with patch("src.ai_client._ensure_gemini_client"), \
patch("src.ai_client._gemini_client") as mock_client:
mock_chat = MagicMock()
mock_client.chats.create.return_value = mock_chat
mock_usage = SimpleNamespace(
prompt_token_count=100,
candidates_token_count=50,
total_token_count=150,
cached_content_token_count=20
)
mock_part = SimpleNamespace(text="Mock Response", function_call=None)
mock_content = SimpleNamespace(parts=[mock_part])
mock_candidate = SimpleNamespace()
mock_candidate.content = mock_content
mock_candidate.finish_reason = SimpleNamespace(name="STOP")
mock_response = SimpleNamespace()
mock_response.candidates = [mock_candidate]
mock_response.usage_metadata = mock_usage
mock_response.text = "Mock Response"
mock_chat.send_message.return_value = mock_response
ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
ai_client.send("Context", "Hello")
comms = ai_client.get_comms_log()
response_entries = [e for e in comms if e.get("direction") == "IN" and e["kind"] == "response"]
assert len(response_entries) > 0
usage = response_entries[0]["payload"]["usage"]
assert usage["input_tokens"] == 100
assert usage["output_tokens"] == 50
assert usage["cache_read_input_tokens"] == 20