mcp/tests/test_public_gateway.py

import unittest

from public_gateway import PublicGatewayApp


class FakeSessionStore:
    def __init__(self):
        self.sessions = {}

    def get(self, gateway_session_id):
        return self.sessions.get(gateway_session_id)


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

    def call_tool(self, token, tool_code, route_path, payload, request_id):
        self.calls.append((token, tool_code, route_path, payload, request_id))
        return {'code': 'MCP_0000', 'data': {'token_used': token}}


class PublicGatewayAppTest(unittest.TestCase):
    def test_two_employees_use_isolated_tokens(self):
        store = FakeSessionStore()
        store.sessions['GWS_A'] = {'mcp_token': 'MT_A'}
        store.sessions['GWS_B'] = {'mcp_token': 'MT_B'}
        api_client = FakeApiClient()
        app = PublicGatewayApp(session_store=store, api_client=api_client, auth_client=None)

        result_a = app.call_tool('GWS_A', 'query_order', {'keyword': 'A'}, request_id='rq_a')
        result_b = app.call_tool('GWS_B', 'query_order', {'keyword': 'B'}, request_id='rq_b')

        self.assertEqual('MT_A', result_a['data']['token_used'])
        self.assertEqual('MT_B', result_b['data']['token_used'])
        self.assertEqual('MT_A', api_client.calls[0][0])
        self.assertEqual('MT_B', api_client.calls[1][0])

    def test_missing_session_returns_bind_hint(self):
        app = PublicGatewayApp(session_store=FakeSessionStore(), api_client=FakeApiClient(), auth_client=None)

        with self.assertRaises(RuntimeError) as error:
            app.call_tool('GWS_missing', 'query_order', {'keyword': 'A'}, request_id='rq_missing')

        self.assertIn('绑定授权码', str(error.exception))


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