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.exchange_calls = []
        self.refresh_calls = []

    def exchange(self, auth_code):
        self.exchange_calls.append(auth_code)
        self.token_store.save('MT_exchange', '2099-01-01T00:00:00')
        return {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'mcp_token': 'MT_exchange',
                'expire_time': '2099-01-01T00:00:00',
            },
        }

    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 = []

    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_bind_exchanges_auth_code_and_persists_token(self):
        store = InMemoryTokenStore(refresh_skew_seconds=60)
        app = GatewayApp(
            auth_client=DummyAuthClient(store),
            api_client=DummyApiClient(),
            token_store=store,
        )

        response = app.bind('AUTH123')

        self.assertEqual('MCP_0000', response['code'])
        self.assertEqual('MT_exchange', store.get()['token'])

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