test_remote_address_tools.py 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178
  1. import unittest
  2. from tools.maintain_remote_address import MaintainRemoteAddressTool
  3. from tools.query_remote_address_list import QueryRemoteAddressListTool
  4. from tools.save_remote_address import SaveRemoteAddressTool
  5. from app import GatewayApp
  6. from public_gateway import PublicGatewayApp
  7. from services.output_presenter import OutputPresenter
  8. class RemoteAddressToolTest(unittest.TestCase):
  9. def test_tools_are_registered_on_base_backend(self):
  10. local = GatewayApp().registered_tool_names()
  11. public = PublicGatewayApp(None, None).registered_tool_names()
  12. self.assertEqual(local, public)
  13. self.assertIn('query_remote_address_list', local)
  14. self.assertIn('maintain_remote_address', local)
  15. self.assertIn('save_remote_address', local)
  16. self.assertEqual('base', QueryRemoteAddressListTool.backend)
  17. self.assertEqual('base', MaintainRemoteAddressTool.backend)
  18. self.assertEqual('base', SaveRemoteAddressTool.backend)
  19. def test_query_posts_filters_to_base_route(self):
  20. client = FakeClient()
  21. result = QueryRemoteAddressListTool(api_client=client).call(
  22. address_no=' PY12 ',
  23. remote_name='美国',
  24. status=1,
  25. postcode='10001',
  26. page=2,
  27. limit=20,
  28. request_id='rq_remote_list',
  29. )
  30. self.assertEqual({'ok': True}, result)
  31. self.assertEqual('query_remote_address_list', client.calls[0][0])
  32. self.assertEqual('/mcp/tools/queryRemoteAddressList', client.calls[0][1])
  33. self.assertEqual({
  34. 'page': 2,
  35. 'limit': 20,
  36. 'address_no': 'PY12',
  37. 'remote_name': '美国',
  38. 'postcode': '10001',
  39. 'status': 1,
  40. }, client.calls[0][2])
  41. def test_maintain_without_number_prepares_create(self):
  42. client = FakeClient()
  43. MaintainRemoteAddressTool(api_client=client).call(request_id='rq_maintain')
  44. self.assertEqual('/mcp/tools/maintainRemoteAddress', client.calls[0][1])
  45. self.assertEqual({'page': 1, 'limit': 20}, client.calls[0][2])
  46. def test_presenter_hides_temp_key(self):
  47. columns = [
  48. {'key': key, 'name': name}
  49. for key, name in OutputPresenter.REMOTE_ADDRESS_DETAIL_COLUMNS
  50. ]
  51. record = {key: '' for key, _name in OutputPresenter.REMOTE_ADDRESS_DETAIL_COLUMNS}
  52. result = OutputPresenter().present('maintain_remote_address', {
  53. 'code': 'MCP_0000',
  54. 'msg': 'success',
  55. 'data': {
  56. 'summary': '准备新增地址库',
  57. 'mode_text': '新增',
  58. 'address_no': '',
  59. 'remote_name': '',
  60. 'status_text': '',
  61. 'temp_key': 'secret-temp-key',
  62. 'columns': columns,
  63. 'records': [record],
  64. 'tips': [],
  65. },
  66. 'meta': {'request_id': 'rq_maintain', 'page': 1, 'limit': 20, 'total': 0},
  67. })
  68. rendered = str(result)
  69. self.assertNotIn('secret-temp-key', rendered)
  70. self.assertNotIn('temp_key', rendered)
  71. def test_save_create_posts_details(self):
  72. client = FakeClient()
  73. SaveRemoteAddressTool(api_client=client).call(
  74. remote_name=' 美西偏远 ',
  75. status=1,
  76. details=[{
  77. 'country': 'US',
  78. 'province': 'CA',
  79. 'start_postcode': '90001',
  80. 'end_postcode': '90099',
  81. }],
  82. request_id='rq_save_remote',
  83. )
  84. self.assertEqual('save_remote_address', client.calls[0][0])
  85. self.assertEqual('/mcp/tools/saveRemoteAddress', client.calls[0][1])
  86. self.assertEqual({
  87. 'remote_name': '美西偏远',
  88. 'status': 1,
  89. 'details': [{
  90. 'country': 'US',
  91. 'province': 'CA',
  92. 'start_postcode': '90001',
  93. 'end_postcode': '90099',
  94. }],
  95. }, client.calls[0][2])
  96. def test_save_rejects_end_postcode_without_start(self):
  97. tool = SaveRemoteAddressTool(api_client=FakeClient())
  98. with self.assertRaises(ValueError):
  99. tool.call(
  100. remote_name='美西偏远',
  101. status=1,
  102. details=[{'country': 'US', 'end_postcode': '90099'}],
  103. )
  104. def test_save_rejects_non_ascii_postcode(self):
  105. tool = SaveRemoteAddressTool(api_client=FakeClient())
  106. with self.assertRaises(ValueError):
  107. tool.call(
  108. remote_name='美西偏远',
  109. status=1,
  110. details=[{'country': 'US', 'start_postcode': '九万'}],
  111. )
  112. def test_query_rejects_overlong_text(self):
  113. tool = QueryRemoteAddressListTool(api_client=FakeClient())
  114. with self.assertRaises(ValueError):
  115. tool.call(address_no='P' * 21)
  116. with self.assertRaises(ValueError):
  117. tool.call(country='USUSUSUSUSU')
  118. def test_save_edit_rejects_details(self):
  119. tool = SaveRemoteAddressTool(api_client=FakeClient())
  120. with self.assertRaises(ValueError):
  121. tool.call(
  122. address_no='PY12',
  123. remote_name='美西偏远',
  124. status=0,
  125. details=[{'country': 'US'}],
  126. )
  127. def test_presenter_shows_saved_address(self):
  128. result = OutputPresenter().present('save_remote_address', {
  129. 'code': 'MCP_0000',
  130. 'msg': 'success',
  131. 'data': {
  132. 'summary': '已新增地址库 PY12 美西偏远,状态启用',
  133. 'mode_text': '新增',
  134. 'address_no': 'PY12',
  135. 'remote_name': '美西偏远',
  136. 'status_text': '启用',
  137. },
  138. 'meta': {'request_id': 'rq_save_remote'},
  139. })
  140. rendered = str(result)
  141. self.assertIn('PY12', rendered)
  142. self.assertIn('美西偏远', rendered)
  143. self.assertNotIn('temp_key', rendered)
  144. def test_presenter_keeps_missing_detail_message(self):
  145. result = OutputPresenter().present('save_remote_address', {
  146. 'code': 'MCP_1401',
  147. 'msg': '地址库没有详情,请导入',
  148. 'data': {},
  149. 'meta': {'request_id': 'rq_save_remote'},
  150. })
  151. self.assertIn('地址库没有详情,请导入', str(result))
  152. class FakeClient:
  153. def __init__(self):
  154. self.calls = []
  155. def call_tool(self, name, route, payload, request_id):
  156. self.calls.append((name, route, payload, request_id))
  157. return {'ok': True}
  158. if __name__ == '__main__':
  159. unittest.main()