mcp/tests/test_gateway_runtime.py

import unittest
from datetime import datetime, timedelta

from app import GatewayApp
from services.token_store import InMemoryTokenStore


class DummyAuthClient:
    def __init__(self, token_store):
        self.token_store = token_store
        self.refresh_calls = []

    def refresh(self, mcp_token):
        self.refresh_calls.append(mcp_token)
        self.token_store.save('MT_refresh', '2099-01-02T00:00:00')
        return {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'mcp_token': 'MT_refresh',
                'expire_time': '2099-01-02T00:00:00',
            },
        }


class DummyApiClient:
    def __init__(self):
        self.calls = []
        self.enabled_calls = 0
        self.enabled_response = {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': [
                    'query_order',
                    'query_track',
                    'query_order_exact',
                    'list_order_filter_options',
                ],
            },
        }

    def list_enabled_tools(self):
        self.enabled_calls += 1
        return self.enabled_response

    def call_tool(self, tool_code, route_path, payload, request_id):
        self.calls.append(
            {
                'tool_code': tool_code,
                'route_path': route_path,
                'payload': payload,
                'request_id': request_id,
            }
        )
        return {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'summary': 'ok',
                'records': [],
                'tips': [],
            },
            'meta': {
                'request_id': request_id,
            },
        }


class GatewayRuntimeTest(unittest.TestCase):
    def test_list_tools_does_not_include_bind_auth_code(self):
        store = InMemoryTokenStore(refresh_skew_seconds=60)
        app = GatewayApp(
            auth_client=DummyAuthClient(store),
            api_client=DummyApiClient(),
            token_store=store,
        )

        names = [tool['name'] for tool in app.list_tools()]

        self.assertIn('query_order', names)
        self.assertIn('query_track', names)
        self.assertIn('query_order_exact', names)
        self.assertIn('list_order_filter_options', names)
        self.assertNotIn('bind_auth_code', names)
        self.assertFalse(hasattr(app, 'bind'))

    def test_list_tools_intersects_registry_codes_with_local_tools(self):
        api_client = DummyApiClient()
        api_client.enabled_response = {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': ['query_order_exact', 'unknown_tool'],
            },
        }
        app = GatewayApp(api_client=api_client)

        names = [tool['name'] for tool in app.list_tools()]

        self.assertEqual(['query_order_exact'], names)
        self.assertEqual(1, api_client.enabled_calls)

    def test_list_tools_fails_closed_on_registry_error(self):
        api_client = DummyApiClient()
        api_client.enabled_response = {
            'code': 'MCP_9001',
            'msg': 'registry unavailable',
            'data': {},
        }
        app = GatewayApp(api_client=api_client)

        with self.assertRaisesRegex(RuntimeError, 'registry unavailable'):
            app.list_tools()

    def test_call_tool_rejects_dynamically_disabled_tool_before_forwarding(self):
        store = InMemoryTokenStore(refresh_skew_seconds=60)
        store.save('MT_valid', '2099-01-01T00:00:00')
        api_client = DummyApiClient()
        api_client.enabled_response = {
            'code': 'MCP_0000',
            'data': {'tool_codes': ['query_track']},
        }
        app = GatewayApp(api_client=api_client, token_store=store)

        with self.assertRaisesRegex(RuntimeError, 'tool disabled: query_order'):
            app.call_tool('query_order', {'keyword': 'ORDER-1'})

        self.assertEqual([], api_client.calls)

    def test_call_tool_refreshes_expiring_token_and_generates_request_id(self):
        store = InMemoryTokenStore(refresh_skew_seconds=60)
        expiring_time = (datetime.now() + timedelta(seconds=10)).isoformat(timespec='seconds')
        store.save('MT_old', expiring_time)
        auth_client = DummyAuthClient(store)
        api_client = DummyApiClient()
        app = GatewayApp(
            auth_client=auth_client,
            api_client=api_client,
            token_store=store,
        )

        response = app.call_tool('query_order', {'keyword': 'SO20260706001'})

        self.assertEqual(['MT_old'], auth_client.refresh_calls)
        self.assertEqual('MT_refresh', store.get()['token'])
        self.assertEqual('query_order', api_client.calls[0]['tool_code'])
        self.assertTrue(api_client.calls[0]['request_id'].startswith('rq_'))
        self.assertEqual(api_client.calls[0]['request_id'], response['meta']['request_id'])


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