feat(ipc): implement cli_tool_bridge as BeforeTool hook
This commit is contained in:
74
tests/test_cli_tool_bridge.py
Normal file
74
tests/test_cli_tool_bridge.py
Normal file
@@ -0,0 +1,74 @@
|
||||
import unittest
|
||||
from unittest.mock import patch, MagicMock
|
||||
import io
|
||||
import json
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add project root to sys.path
|
||||
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
||||
|
||||
# Import after path fix
|
||||
from scripts.cli_tool_bridge import main
|
||||
|
||||
class TestCliToolBridge(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tool_call = {
|
||||
'tool_name': 'read_file',
|
||||
'tool_input': {'path': 'test.txt'}
|
||||
}
|
||||
|
||||
@patch('sys.stdin', new_callable=io.StringIO)
|
||||
@patch('sys.stdout', new_callable=io.StringIO)
|
||||
@patch('api_hook_client.ApiHookClient.request_confirmation')
|
||||
def test_allow_decision(self, mock_request, mock_stdout, mock_stdin):
|
||||
# 1. Mock stdin with a JSON string tool call
|
||||
mock_stdin.write(json.dumps(self.tool_call))
|
||||
mock_stdin.seek(0)
|
||||
|
||||
# 2. Mock ApiHookClient to return approved
|
||||
mock_request.return_value = {'approved': True}
|
||||
|
||||
# Run main
|
||||
main()
|
||||
|
||||
# 3. Capture stdout and assert allow
|
||||
output = json.loads(mock_stdout.getvalue().strip())
|
||||
self.assertEqual(output.get('decision'), 'allow')
|
||||
|
||||
@patch('sys.stdin', new_callable=io.StringIO)
|
||||
@patch('sys.stdout', new_callable=io.StringIO)
|
||||
@patch('api_hook_client.ApiHookClient.request_confirmation')
|
||||
def test_deny_decision(self, mock_request, mock_stdout, mock_stdin):
|
||||
# Mock stdin
|
||||
mock_stdin.write(json.dumps(self.tool_call))
|
||||
mock_stdin.seek(0)
|
||||
|
||||
# 4. Mock ApiHookClient to return denied
|
||||
mock_request.return_value = {'approved': False}
|
||||
|
||||
main()
|
||||
|
||||
# Assert deny
|
||||
output = json.loads(mock_stdout.getvalue().strip())
|
||||
self.assertEqual(output.get('decision'), 'deny')
|
||||
|
||||
@patch('sys.stdin', new_callable=io.StringIO)
|
||||
@patch('sys.stdout', new_callable=io.StringIO)
|
||||
@patch('api_hook_client.ApiHookClient.request_confirmation')
|
||||
def test_unreachable_hook_server(self, mock_request, mock_stdout, mock_stdin):
|
||||
# Mock stdin
|
||||
mock_stdin.write(json.dumps(self.tool_call))
|
||||
mock_stdin.seek(0)
|
||||
|
||||
# 5. Test case where hook server is unreachable (exception)
|
||||
mock_request.side_effect = Exception("Connection refused")
|
||||
|
||||
main()
|
||||
|
||||
# Assert deny on error
|
||||
output = json.loads(mock_stdout.getvalue().strip())
|
||||
self.assertEqual(output.get('decision'), 'deny')
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user