| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116 |
- import unittest
- from app import GatewayApp
- from mcp_protocol import McpProtocolHandler
- from services.token_store import InMemoryTokenStore
- class DummyApiClient:
- def list_enabled_tools(self, request_id=''):
- return {
- 'code': 'MCP_0000',
- 'data': {
- 'tool_codes': [
- 'query_order',
- 'query_track',
- 'query_order_exact',
- 'list_order_filter_options',
- ],
- },
- }
- 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 NoBindAuthCodeToolTest(unittest.TestCase):
- def build_handler(self, with_token=True):
- store = InMemoryTokenStore(refresh_skew_seconds=60)
- if with_token:
- store.save('MT_ready', '2099-01-01T00:00:00')
- app = GatewayApp(
- auth_client=None,
- api_client=DummyApiClient(),
- token_store=store,
- )
- return McpProtocolHandler(app), store
- def test_tools_list_does_not_include_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('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)
- def test_bind_auth_code_call_is_not_registered(self):
- handler, _ = 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.assertTrue(response['result']['isError'])
- self.assertNotIn('bind_auth_code', response['result']['content'][0]['text'])
- self.assertEqual(
- 'MCP_9001',
- response['result']['structuredContent']['code'],
- )
- def test_query_order_without_token_returns_device_invalid_message(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.assertNotIn('授权码', response['result']['content'][0]['text'])
- self.assertIn('Workbuddy', response['result']['content'][0]['text'])
- if __name__ == '__main__':
- unittest.main()
|