123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
import json
|
|
import subprocess
|
|
import io
|
|
import sys
|
|
import os
|
|
|
|
# Ensure the project root is in sys.path to resolve imports correctly
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
from gemini_cli_adapter import GeminiCliAdapter
|
|
|
|
class TestGeminiCliAdapter(unittest.TestCase):
|
|
def setUp(self):
|
|
self.adapter = GeminiCliAdapter(binary_path="gemini")
|
|
|
|
@patch('subprocess.Popen')
|
|
def test_send_starts_subprocess_with_correct_args(self, mock_popen):
|
|
"""
|
|
Verify that send(message) correctly starts the subprocess with
|
|
--output-format stream-json and the provided message.
|
|
"""
|
|
# Setup mock process with a minimal valid JSONL termination
|
|
process_mock = MagicMock()
|
|
process_mock.stdout = io.StringIO(json.dumps({"type": "result", "usage": {}}) + "\n")
|
|
process_mock.poll.return_value = 0
|
|
process_mock.wait.return_value = 0
|
|
mock_popen.return_value = process_mock
|
|
|
|
message = "Hello Gemini CLI"
|
|
self.adapter.send(message)
|
|
|
|
# Verify subprocess.Popen call
|
|
mock_popen.assert_called_once()
|
|
args, kwargs = mock_popen.call_args
|
|
cmd = args[0]
|
|
|
|
# Check mandatory CLI components
|
|
self.assertIn("gemini", cmd)
|
|
self.assertIn("--output-format", cmd)
|
|
self.assertIn("stream-json", cmd)
|
|
self.assertIn(message, cmd)
|
|
|
|
# Check process configuration
|
|
self.assertEqual(kwargs.get('stdout'), subprocess.PIPE)
|
|
self.assertEqual(kwargs.get('text'), True)
|
|
|
|
@patch('subprocess.Popen')
|
|
def test_send_parses_jsonl_output(self, mock_popen):
|
|
"""
|
|
Verify that it correctly parses multiple JSONL 'message' events
|
|
and returns the combined text.
|
|
"""
|
|
jsonl_output = [
|
|
json.dumps({"type": "message", "text": "The quick brown "}),
|
|
json.dumps({"type": "message", "text": "fox jumps."}),
|
|
json.dumps({"type": "result", "usage": {"prompt_tokens": 5, "candidates_tokens": 5}})
|
|
]
|
|
stdout_content = "\n".join(jsonl_output) + "\n"
|
|
|
|
process_mock = MagicMock()
|
|
process_mock.stdout = io.StringIO(stdout_content)
|
|
# Mock poll sequence: running, running, finished
|
|
process_mock.poll.side_effect = [None, None, 0]
|
|
process_mock.wait.return_value = 0
|
|
mock_popen.return_value = process_mock
|
|
|
|
result = self.adapter.send("test message")
|
|
|
|
self.assertEqual(result, "The quick brown fox jumps.")
|
|
|
|
@patch('subprocess.Popen')
|
|
def test_send_handles_tool_use_events(self, mock_popen):
|
|
"""
|
|
Verify that it correctly handles 'tool_use' events in the stream
|
|
by continuing to read until the final 'result' event.
|
|
"""
|
|
jsonl_output = [
|
|
json.dumps({"type": "message", "text": "Calling tool..."}),
|
|
json.dumps({"type": "tool_use", "name": "read_file", "args": {"path": "test.txt"}}),
|
|
json.dumps({"type": "message", "text": "\nFile read successfully."}),
|
|
json.dumps({"type": "result", "usage": {}})
|
|
]
|
|
stdout_content = "\n".join(jsonl_output) + "\n"
|
|
|
|
process_mock = MagicMock()
|
|
process_mock.stdout = io.StringIO(stdout_content)
|
|
process_mock.poll.side_effect = [None, None, None, 0]
|
|
process_mock.wait.return_value = 0
|
|
mock_popen.return_value = process_mock
|
|
|
|
result = self.adapter.send("read test.txt")
|
|
|
|
# Result should contain the combined text from all 'message' events
|
|
self.assertEqual(result, "Calling tool...\nFile read successfully.")
|
|
|
|
@patch('subprocess.Popen')
|
|
def test_send_captures_usage_metadata(self, mock_popen):
|
|
"""
|
|
Verify that usage data is extracted from the 'result' event.
|
|
"""
|
|
usage_data = {"total_tokens": 42}
|
|
jsonl_output = [
|
|
json.dumps({"type": "message", "text": "Finalizing"}),
|
|
json.dumps({"type": "result", "usage": usage_data})
|
|
]
|
|
stdout_content = "\n".join(jsonl_output) + "\n"
|
|
|
|
process_mock = MagicMock()
|
|
process_mock.stdout = io.StringIO(stdout_content)
|
|
process_mock.poll.side_effect = [None, 0]
|
|
process_mock.wait.return_value = 0
|
|
mock_popen.return_value = process_mock
|
|
|
|
self.adapter.send("usage test")
|
|
|
|
# Verify the usage was captured in the adapter instance
|
|
self.assertEqual(self.adapter.last_usage, usage_data)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|