test_bind_auth_code_tool.py 2.8 KB

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