import unittest from unittest.mock import MagicMock, patch from constants import BIND_HINT from public_gateway import PublicGatewayApp class TestPublicGatewayApp(unittest.TestCase): def setUp(self): self.mock_session_store = MagicMock() self.mock_api_client = MagicMock() self.mock_auth_client = MagicMock() self.app = PublicGatewayApp( session_store=self.mock_session_store, api_client=self.mock_api_client, auth_client=self.mock_auth_client ) def test_list_tools_returns_metadata(self): tools = self.app.list_tools() self.assertIsInstance(tools, list) self.assertGreater(len(tools), 0) # Check that each tool has required fields for tool in tools: self.assertIn('name', tool) self.assertIn('description', tool) self.assertIn('input_schema', tool) def test_build_request_id_generates_id_when_empty(self): request_id = self.app.build_request_id('') self.assertTrue(request_id.startswith('rq_')) self.assertEqual(len(request_id), 3 + 16) # 'rq_' + 16 hex chars def test_build_request_id_uses_provided_id(self): provided_id = 'custom_request_123' request_id = self.app.build_request_id(provided_id) self.assertEqual(request_id, provided_id) def test_bind_auth_code_success(self): gateway_session_id = 'GWS_test123' auth_code = 'AC_xyz789' self.mock_auth_client.exchange.return_value = { 'code': 'MCP_0000', 'msg': 'success', 'data': {'mcp_token': 'MT_token'} } self.mock_session_store.get.return_value = { 'admin_id': 123, 'company_id': 100 } result = self.app.bind_auth_code(gateway_session_id, auth_code) self.mock_auth_client.exchange.assert_called_once_with(gateway_session_id, auth_code) self.assertEqual(result['code'], 'MCP_0000') def test_bind_auth_code_log_does_not_include_auth_code_value(self): gateway_session_id = 'GWS_test123' auth_code = 'AC_secret_value_123' self.mock_auth_client.exchange.return_value = {'code': 'MCP_9999', 'msg': 'failed'} with self.assertLogs('public_gateway', level='INFO') as logs: self.app.bind_auth_code(gateway_session_id, auth_code) joined_logs = '\n'.join(logs.output) self.assertIn('bind_auth_code', joined_logs) self.assertNotIn(auth_code, joined_logs) self.assertNotIn('AC_secre', joined_logs) def test_bind_auth_code_log_does_not_include_auth_code_value(self): gateway_session_id = 'GWS_test123' auth_code = 'AC_secret_value_123' self.mock_auth_client.exchange.return_value = {'code': 'MCP_9999', 'msg': 'failed'} with self.assertLogs('public_gateway', level='INFO') as logs: self.app.bind_auth_code(gateway_session_id, auth_code) joined_logs = '\n'.join(logs.output) self.assertIn('bind_auth_code', joined_logs) self.assertNotIn(auth_code, joined_logs) self.assertNotIn('AC_secre', joined_logs) def test_bind_auth_code_raises_when_no_auth_client(self): app_no_auth = PublicGatewayApp( session_store=self.mock_session_store, api_client=self.mock_api_client, auth_client=None ) with self.assertRaises(RuntimeError) as context: app_no_auth.bind_auth_code('GWS_test', 'AC_code') self.assertIn('auth client is required', str(context.exception)) def test_call_tool_bind_auth_code(self): gateway_session_id = 'GWS_bind' arguments = {'auth_code': 'AC_test'} self.mock_auth_client.exchange.return_value = { 'code': 'MCP_0000', 'data': {} } self.mock_session_store.get.return_value = {'admin_id': 1} result = self.app.call_tool(gateway_session_id, 'bind_auth_code', arguments) self.mock_auth_client.exchange.assert_called_once_with(gateway_session_id, 'AC_test') def test_call_tool_raises_on_unregistered_tool(self): with self.assertRaises(KeyError) as context: self.app.call_tool('GWS_test', 'nonexistent_tool', {}) self.assertIn('tool not registered', str(context.exception)) def test_call_tool_raises_when_session_not_found(self): self.mock_session_store.get.return_value = None with self.assertRaises(RuntimeError) as context: self.app.call_tool('GWS_nosession', 'query_order', {}) self.assertEqual(str(context.exception), BIND_HINT) def test_call_tool_raises_when_no_token_in_session(self): self.mock_session_store.get.return_value = { 'admin_id': 123, 'company_id': 100 # Missing mcp_token } with self.assertRaises(RuntimeError) as context: self.app.call_tool('GWS_notoken', 'query_order', {}) self.assertEqual(str(context.exception), BIND_HINT) def test_call_tool_success(self): gateway_session_id = 'GWS_valid' tool_name = 'query_order' arguments = {'order_no': 'ABC123'} self.mock_session_store.get.return_value = { 'mcp_token': 'MT_token123', 'admin_id': 456, 'company_id': 200 } self.mock_api_client.call_tool.return_value = { 'code': '0', 'msg': 'success', 'data': {'order': 'details'} } result = self.app.call_tool(gateway_session_id, tool_name, arguments, 'rq_test') # Verify api_client.call_tool was called correctly self.mock_api_client.call_tool.assert_called_once() call_args = self.mock_api_client.call_tool.call_args[1] self.assertEqual(call_args['token'], 'MT_token123') self.assertEqual(call_args['tool_code'], 'query_order') self.assertEqual(call_args['payload'], arguments) self.assertTrue(call_args['request_id'].startswith('rq_')) # Verify result self.assertEqual(result['code'], '0') def test_call_tool_generates_request_id_when_not_provided(self): self.mock_session_store.get.return_value = { 'mcp_token': 'MT_token', 'admin_id': 1, 'company_id': 1 } self.mock_api_client.call_tool.return_value = {'code': '0'} self.app.call_tool('GWS_test', 'query_order', {}, '') call_args = self.mock_api_client.call_tool.call_args[1] self.assertTrue(call_args['request_id'].startswith('rq_')) def test_call_tool_uses_provided_request_id(self): self.mock_session_store.get.return_value = { 'mcp_token': 'MT_token', 'admin_id': 1, 'company_id': 1 } self.mock_api_client.call_tool.return_value = {'code': '0'} custom_request_id = 'rq_custom_123' self.app.call_tool('GWS_test', 'query_order', {}, custom_request_id) call_args = self.mock_api_client.call_tool.call_args[1] self.assertEqual(call_args['request_id'], custom_request_id) def test_call_tool_logs_error_on_exception(self): """Test that tool call exceptions are logged and re-raised""" gateway_session_id = 'GWS_error_test' tool_name = 'query_order' self.mock_session_store.get.return_value = { 'mcp_token': 'MT_token', 'admin_id': 999, 'company_id': 888 } # Make API client raise an exception self.mock_api_client.call_tool.side_effect = RuntimeError("API connection failed") with self.assertRaises(RuntimeError) as context: self.app.call_tool(gateway_session_id, tool_name, {}, 'rq_err') self.assertIn("API connection failed", str(context.exception)) if __name__ == '__main__': unittest.main()