mcp/tests/test_scoped_clients.py

import unittest

from services.scoped_api_client import ScopedApiClient
from services.scoped_auth_client import ScopedAuthClient


class FakeTransport:
    def __init__(self):
        self.calls = []

    def post_json(self, url, payload, headers, timeout):
        self.calls.append((url, payload, headers, timeout))
        return {'code': 'MCP_0000', 'data': {'ok': True}}


class ScopedApiClientTest(unittest.TestCase):
    def test_call_tool_uses_explicit_token_not_global_store(self):
        transport = FakeTransport()
        client = ScopedApiClient('https://tools.example.com', transport=transport, timeout=7)

        response = client.call_tool(
            token='MT_employee_a',
            tool_code='query_order',
            route_path='/mcp/tools/queryOrder',
            payload={'keyword': 'USC'},
            request_id='rq_public_1',
        )

        self.assertEqual('MCP_0000', response['code'])
        url, payload, headers, timeout = transport.calls[0]
        self.assertEqual('https://tools.example.com/mcp/tools/queryOrder', url)
        self.assertEqual('Bearer MT_employee_a', headers['Authorization'])
        self.assertEqual('query_order', headers['X-MCP-Tool-Code'])
        self.assertEqual('rq_public_1', headers['X-Request-Id'])
        self.assertEqual(7, timeout)


class ScopedAuthClientTest(unittest.TestCase):
    def test_exchange_sends_gateway_session_key_and_stores_token(self):
        transport = FakeTransport()
        store = {}

        class Store:
            def save(self, gateway_session_id, session):
                store[gateway_session_id] = session
                return session

        def post_json(url, payload, headers, timeout):
            transport.calls.append((url, payload, headers, timeout))
            return {
                'code': 'MCP_0000',
                'data': {
                    'mcp_token': 'MT_employee_a',
                    'expire_time': '2099-12-31 23:59:59',
                    'admin_id': 1,
                    'company_id': 10,
                },
            }

        transport.post_json = post_json
        client = ScopedAuthClient(
            base_url='https://base.example.com',
            client_type='workbuddy',
            session_store=Store(),
            transport=transport,
            timeout=5,
        )

        response = client.exchange('GWS_employee_a', 'AUTH_CODE_A')

        self.assertEqual('MCP_0000', response['code'])
        url, payload, headers, timeout = transport.calls[0]
        self.assertEqual('https://base.example.com/mcp/auth/exchange', url)
        self.assertEqual('AUTH_CODE_A', payload['auth_code'])
        self.assertEqual('workbuddy', payload['client_type'])
        self.assertEqual('gws_' + 'GWS_employee_a', payload['session_key'])
        self.assertEqual('MT_employee_a', store['GWS_employee_a']['mcp_token'])


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