mcp/tests/test_bind_auth_code_tool.py

import unittest

from app import GatewayApp
from mcp_protocol import McpProtocolHandler
from services.token_store import InMemoryTokenStore


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

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


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


class BindAuthCodeToolTest(unittest.TestCase):
    def build_handler(self, with_token=False):
        store = InMemoryTokenStore(refresh_skew_seconds=60)
        if with_token:
            store.save('MT_ready', '2099-01-01T00:00:00')
        auth_client = DummyAuthClient(store)
        app = GatewayApp(
            auth_client=auth_client,
            api_client=DummyApiClient(),
            token_store=store,
        )
        return McpProtocolHandler(app), auth_client, store

    def test_tools_list_includes_bind_auth_code(self):
        handler, _, _ = self.build_handler()

        response = handler.handle_request(
            {
                'jsonrpc': '2.0',
                'id': 1,
                'method': 'tools/list',
                'params': {},
            }
        )

        names = [tool['name'] for tool in response['result']['tools']]
        self.assertIn('bind_auth_code', names)
        bind_tool = [tool for tool in response['result']['tools'] if tool['name'] == 'bind_auth_code'][0]
        self.assertEqual(['auth_code'], bind_tool['inputSchema']['required'])

    def test_bind_auth_code_exchanges_code_and_hides_token(self):
        handler, auth_client, store = self.build_handler()

        response = handler.handle_request(
            {
                'jsonrpc': '2.0',
                'id': 2,
                'method': 'tools/call',
                'params': {
                    'name': 'bind_auth_code',
                    'arguments': {
                        'auth_code': ' AUTH123 ',
                    },
                },
            }
        )

        self.assertFalse(response['result']['isError'])
        self.assertEqual(['AUTH123'], auth_client.exchange_calls)
        self.assertEqual('MT_bound_secret', store.get()['token'])
        text = response['result']['content'][0]['text']
        structured = response['result']['structuredContent']
        self.assertIn('绑定成功', text)
        self.assertNotIn('MT_bound_secret', text)
        self.assertNotIn('mcp_token', structured)
        self.assertEqual('bound', structured['status'])

    def test_query_order_without_token_returns_employee_binding_hint(self):
        handler, _, _ = self.build_handler(with_token=False)

        response = handler.handle_request(
            {
                'jsonrpc': '2.0',
                'id': 3,
                'method': 'tools/call',
                'params': {
                    'name': 'query_order',
                    'arguments': {
                        'keyword': 'SO20260706001',
                    },
                },
            }
        )

        self.assertTrue(response['result']['isError'])
        self.assertIn('请先登录后台获取 Workbuddy 授权码', response['result']['content'][0]['text'])


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