mcp/tests/test_public_gateway_unit.py

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()