mcp/tests/test_scoped_api_client.py

import unittest
from unittest.mock import MagicMock

from services.scoped_api_client import ScopedApiClient


class TestScopedApiClient(unittest.TestCase):
    def setUp(self):
        self.mock_transport = MagicMock()
        self.client = ScopedApiClient(
            base_url='http://api.example.com',
            transport=self.mock_transport,
            timeout=15
        )

    def test_call_tool_success(self):
        token = 'MT_valid_token'
        tool_code = 'query_order'
        route_path = '/mcp/tools/query_order'
        payload = {'order_no': 'ABC123'}
        request_id = 'rq_test123'

        self.mock_transport.post_json.return_value = {
            'code': '0',
            'msg': 'success',
            'data': {'order': 'details'}
        }

        result = self.client.call_tool(token, tool_code, route_path, payload, request_id)

        # Verify transport call
        self.mock_transport.post_json.assert_called_once_with(
            'http://api.example.com/mcp/tools/query_order',
            payload,
            {
                'Authorization': 'Bearer MT_valid_token',
                'X-MCP-Tool-Code': 'query_order',
                'X-Request-Id': 'rq_test123',
            },
            15
        )

        # Verify result
        self.assertEqual(result['code'], '0')
        self.assertEqual(result['data']['order'], 'details')

    def test_call_tool_includes_client_ip_header_when_present(self):
        self.mock_transport.post_json.return_value = {'code': '0'}

        self.client.call_tool(
            'MT_valid_token',
            'query_order',
            '/mcp/tools/query_order',
            {'order_no': 'ABC123'},
            'rq_ip_header',
            client_ip='203.0.113.9'
        )

        call_headers = self.mock_transport.post_json.call_args[0][2]
        self.assertEqual(call_headers['X-MCP-Client-IP'], '203.0.113.9')

    def test_call_tool_raises_on_empty_token(self):
        with self.assertRaises(RuntimeError) as context:
            self.client.call_tool('', 'query_order', '/path', {}, 'rq_1')

        self.assertIn('mcp token missing', str(context.exception))

    def test_call_tool_raises_on_none_token(self):
        with self.assertRaises(RuntimeError) as context:
            self.client.call_tool(None, 'query_order', '/path', {}, 'rq_1')

        self.assertIn('mcp token missing', str(context.exception))

    def test_call_tool_strips_leading_slash_from_route(self):
        self.mock_transport.post_json.return_value = {'code': '0'}

        self.client.call_tool('MT_token', 'tool', '/api/endpoint', {}, 'rq_1')

        call_url = self.mock_transport.post_json.call_args[0][0]
        self.assertEqual(call_url, 'http://api.example.com/api/endpoint')

    def test_call_tool_handles_route_without_leading_slash(self):
        self.mock_transport.post_json.return_value = {'code': '0'}

        self.client.call_tool('MT_token', 'tool', 'api/endpoint', {}, 'rq_1')

        call_url = self.mock_transport.post_json.call_args[0][0]
        self.assertEqual(call_url, 'http://api.example.com/api/endpoint')

    def test_call_tool_strips_trailing_slash_from_base_url(self):
        client_with_slash = ScopedApiClient(
            base_url='http://api.example.com/',
            transport=self.mock_transport,
            timeout=10
        )

        self.mock_transport.post_json.return_value = {'code': '0'}

        client_with_slash.call_tool('MT_token', 'tool', '/endpoint', {}, 'rq_1')

        call_url = self.mock_transport.post_json.call_args[0][0]
        # Should not have double slash
        self.assertEqual(call_url, 'http://api.example.com/endpoint')

    def test_call_tool_timeout_parameter(self):
        client_short_timeout = ScopedApiClient(
            base_url='http://api.example.com',
            transport=self.mock_transport,
            timeout=5
        )

        self.mock_transport.post_json.return_value = {'code': '0'}

        client_short_timeout.call_tool('MT_token', 'tool', '/path', {}, 'rq_1')

        # Verify timeout is passed correctly
        call_timeout = self.mock_transport.post_json.call_args[0][3]
        self.assertEqual(call_timeout, 5)

    def test_call_tool_with_empty_payload(self):
        self.mock_transport.post_json.return_value = {'code': '0'}

        self.client.call_tool('MT_token', 'tool', '/path', {}, 'rq_1')

        call_payload = self.mock_transport.post_json.call_args[0][1]
        self.assertEqual(call_payload, {})

    def test_call_tool_headers_format(self):
        self.mock_transport.post_json.return_value = {'code': '0'}

        self.client.call_tool('MT_abc123', 'query_track', '/track', {'no': '123'}, 'rq_xyz')

        call_headers = self.mock_transport.post_json.call_args[0][2]
        self.assertEqual(call_headers['Authorization'], 'Bearer MT_abc123')
        self.assertEqual(call_headers['X-MCP-Tool-Code'], 'query_track')
        self.assertEqual(call_headers['X-Request-Id'], 'rq_xyz')

    def test_list_enabled_tools_uses_only_explicit_token(self):
        expected = {
            'code': 'MCP_0000',
            'data': {'tool_codes': ['query_order_exact']},
        }
        self.mock_transport.post_json.return_value = expected

        result = self.client.list_enabled_tools(' MT_scoped ')

        self.assertIs(expected, result)
        self.mock_transport.post_json.assert_called_once_with(
            'http://api.example.com/mcp/tools/listEnabledTools',
            {},
            {'Authorization': 'Bearer MT_scoped'},
            15,
        )

    def test_list_enabled_tools_rejects_empty_token(self):
        with self.assertRaisesRegex(RuntimeError, 'mcp token missing'):
            self.client.list_enabled_tools('')

        self.mock_transport.post_json.assert_not_called()


if __name__ == '__main__':
    unittest.main()