test_bind_auth_code_tool.py 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116
  1. import unittest
  2. from app import GatewayApp
  3. from mcp_protocol import McpProtocolHandler
  4. from services.token_store import InMemoryTokenStore
  5. class DummyApiClient:
  6. def list_enabled_tools(self, request_id=''):
  7. return {
  8. 'code': 'MCP_0000',
  9. 'data': {
  10. 'tool_codes': [
  11. 'query_order',
  12. 'query_track',
  13. 'query_order_exact',
  14. 'list_order_filter_options',
  15. ],
  16. },
  17. }
  18. def call_tool(self, tool_code, route_path, payload, request_id):
  19. return {
  20. 'code': 'MCP_0000',
  21. 'msg': 'success',
  22. 'data': {
  23. 'summary': 'ok',
  24. 'records': [],
  25. 'tips': [],
  26. },
  27. 'meta': {
  28. 'request_id': request_id,
  29. },
  30. }
  31. class NoBindAuthCodeToolTest(unittest.TestCase):
  32. def build_handler(self, with_token=True):
  33. store = InMemoryTokenStore(refresh_skew_seconds=60)
  34. if with_token:
  35. store.save('MT_ready', '2099-01-01T00:00:00')
  36. app = GatewayApp(
  37. auth_client=None,
  38. api_client=DummyApiClient(),
  39. token_store=store,
  40. )
  41. return McpProtocolHandler(app), store
  42. def test_tools_list_does_not_include_bind_auth_code(self):
  43. handler, _ = self.build_handler()
  44. response = handler.handle_request(
  45. {
  46. 'jsonrpc': '2.0',
  47. 'id': 1,
  48. 'method': 'tools/list',
  49. 'params': {},
  50. }
  51. )
  52. names = [tool['name'] for tool in response['result']['tools']]
  53. self.assertIn('query_order', names)
  54. self.assertIn('query_track', names)
  55. self.assertIn('query_order_exact', names)
  56. self.assertIn('list_order_filter_options', names)
  57. self.assertNotIn('bind_auth_code', names)
  58. def test_bind_auth_code_call_is_not_registered(self):
  59. handler, _ = self.build_handler()
  60. response = handler.handle_request(
  61. {
  62. 'jsonrpc': '2.0',
  63. 'id': 2,
  64. 'method': 'tools/call',
  65. 'params': {
  66. 'name': 'bind_auth_code',
  67. 'arguments': {
  68. 'auth_code': 'AUTH123',
  69. },
  70. },
  71. }
  72. )
  73. self.assertTrue(response['result']['isError'])
  74. self.assertNotIn('bind_auth_code', response['result']['content'][0]['text'])
  75. self.assertEqual(
  76. 'MCP_9001',
  77. response['result']['structuredContent']['code'],
  78. )
  79. def test_query_order_without_token_returns_device_invalid_message(self):
  80. handler, _ = self.build_handler(with_token=False)
  81. response = handler.handle_request(
  82. {
  83. 'jsonrpc': '2.0',
  84. 'id': 3,
  85. 'method': 'tools/call',
  86. 'params': {
  87. 'name': 'query_order',
  88. 'arguments': {
  89. 'keyword': 'SO20260706001',
  90. },
  91. },
  92. }
  93. )
  94. self.assertTrue(response['result']['isError'])
  95. self.assertNotIn('授权码', response['result']['content'][0]['text'])
  96. self.assertIn('Workbuddy', response['result']['content'][0]['text'])
  97. if __name__ == '__main__':
  98. unittest.main()