mcp/tests/test_remote_address_tools.py

import unittest

from tools.maintain_remote_address import MaintainRemoteAddressTool
from tools.query_remote_address_list import QueryRemoteAddressListTool
from tools.save_remote_address import SaveRemoteAddressTool
from app import GatewayApp
from public_gateway import PublicGatewayApp
from services.output_presenter import OutputPresenter


class RemoteAddressToolTest(unittest.TestCase):
    def test_tools_are_registered_on_base_backend(self):
        local = GatewayApp().registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()
        self.assertEqual(local, public)
        self.assertIn('query_remote_address_list', local)
        self.assertIn('maintain_remote_address', local)
        self.assertIn('save_remote_address', local)
        self.assertEqual('base', QueryRemoteAddressListTool.backend)
        self.assertEqual('base', MaintainRemoteAddressTool.backend)
        self.assertEqual('base', SaveRemoteAddressTool.backend)

    def test_query_posts_filters_to_base_route(self):
        client = FakeClient()
        result = QueryRemoteAddressListTool(api_client=client).call(
            address_no=' PY12 ',
            remote_name='美国',
            status=1,
            postcode='10001',
            page=2,
            limit=20,
            request_id='rq_remote_list',
        )
        self.assertEqual({'ok': True}, result)
        self.assertEqual('query_remote_address_list', client.calls[0][0])
        self.assertEqual('/mcp/tools/queryRemoteAddressList', client.calls[0][1])
        self.assertEqual({
            'page': 2,
            'limit': 20,
            'address_no': 'PY12',
            'remote_name': '美国',
            'postcode': '10001',
            'status': 1,
        }, client.calls[0][2])

    def test_maintain_without_number_prepares_create(self):
        client = FakeClient()
        MaintainRemoteAddressTool(api_client=client).call(request_id='rq_maintain')
        self.assertEqual('/mcp/tools/maintainRemoteAddress', client.calls[0][1])
        self.assertEqual({'page': 1, 'limit': 20}, client.calls[0][2])

    def test_presenter_hides_temp_key(self):
        columns = [
            {'key': key, 'name': name}
            for key, name in OutputPresenter.REMOTE_ADDRESS_DETAIL_COLUMNS
        ]
        record = {key: '' for key, _name in OutputPresenter.REMOTE_ADDRESS_DETAIL_COLUMNS}
        result = OutputPresenter().present('maintain_remote_address', {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'summary': '准备新增地址库',
                'mode_text': '新增',
                'address_no': '',
                'remote_name': '',
                'status_text': '',
                'temp_key': 'secret-temp-key',
                'columns': columns,
                'records': [record],
                'tips': [],
            },
            'meta': {'request_id': 'rq_maintain', 'page': 1, 'limit': 20, 'total': 0},
        })
        rendered = str(result)
        self.assertNotIn('secret-temp-key', rendered)
        self.assertNotIn('temp_key', rendered)

    def test_save_create_posts_details(self):
        client = FakeClient()
        SaveRemoteAddressTool(api_client=client).call(
            remote_name=' 美西偏远 ',
            status=1,
            details=[{
                'country': 'US',
                'province': 'CA',
                'start_postcode': '90001',
                'end_postcode': '90099',
            }],
            request_id='rq_save_remote',
        )
        self.assertEqual('save_remote_address', client.calls[0][0])
        self.assertEqual('/mcp/tools/saveRemoteAddress', client.calls[0][1])
        self.assertEqual({
            'remote_name': '美西偏远',
            'status': 1,
            'details': [{
                'country': 'US',
                'province': 'CA',
                'start_postcode': '90001',
                'end_postcode': '90099',
            }],
        }, client.calls[0][2])

    def test_save_rejects_end_postcode_without_start(self):
        tool = SaveRemoteAddressTool(api_client=FakeClient())
        with self.assertRaises(ValueError):
            tool.call(
                remote_name='美西偏远',
                status=1,
                details=[{'country': 'US', 'end_postcode': '90099'}],
            )

    def test_save_rejects_non_ascii_postcode(self):
        tool = SaveRemoteAddressTool(api_client=FakeClient())
        with self.assertRaises(ValueError):
            tool.call(
                remote_name='美西偏远',
                status=1,
                details=[{'country': 'US', 'start_postcode': '九万'}],
            )

    def test_query_rejects_overlong_text(self):
        tool = QueryRemoteAddressListTool(api_client=FakeClient())
        with self.assertRaises(ValueError):
            tool.call(address_no='P' * 21)
        with self.assertRaises(ValueError):
            tool.call(country='USUSUSUSUSU')

    def test_save_edit_rejects_details(self):
        tool = SaveRemoteAddressTool(api_client=FakeClient())
        with self.assertRaises(ValueError):
            tool.call(
                address_no='PY12',
                remote_name='美西偏远',
                status=0,
                details=[{'country': 'US'}],
            )

    def test_presenter_shows_saved_address(self):
        result = OutputPresenter().present('save_remote_address', {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': {
                'summary': '已新增地址库 PY12 美西偏远，状态启用',
                'mode_text': '新增',
                'address_no': 'PY12',
                'remote_name': '美西偏远',
                'status_text': '启用',
            },
            'meta': {'request_id': 'rq_save_remote'},
        })
        rendered = str(result)
        self.assertIn('PY12', rendered)
        self.assertIn('美西偏远', rendered)
        self.assertNotIn('temp_key', rendered)

    def test_presenter_keeps_missing_detail_message(self):
        result = OutputPresenter().present('save_remote_address', {
            'code': 'MCP_1401',
            'msg': '地址库没有详情，请导入',
            'data': {},
            'meta': {'request_id': 'rq_save_remote'},
        })
        self.assertIn('地址库没有详情，请导入', str(result))

    def test_tools_expose_metadata_and_reject_bad_input(self):
        self.assertIn('address_no', QueryRemoteAddressListTool().metadata()['input_schema']['properties'])
        self.assertIn('address_no', MaintainRemoteAddressTool().metadata()['input_schema']['properties'])
        self.assertIn('details', SaveRemoteAddressTool().metadata()['input_schema']['properties'])
        with self.assertRaises(RuntimeError):
            QueryRemoteAddressListTool().call()
        with self.assertRaises(RuntimeError):
            MaintainRemoteAddressTool().call()
        with self.assertRaises(RuntimeError):
            SaveRemoteAddressTool().call(remote_name='美西', status=1, details=[{'country': 'US'}])

        query = QueryRemoteAddressListTool(api_client=FakeClient())
        for kwargs in (
            {'country': 1},
            {'status': True},
            {'status': 2},
            {'page': True},
            {'page': 0},
            {'limit': 101},
        ):
            with self.assertRaises(ValueError):
                query.call(**kwargs)
        query.call(province='CA', city='LA', remote_name='   ')

        maintain = MaintainRemoteAddressTool(api_client=FakeClient())
        with self.assertRaises(ValueError):
            maintain.call(address_no=12)
        with self.assertRaises(ValueError):
            maintain.call(page=True)
        with self.assertRaises(ValueError):
            maintain.call(limit=0)
        maintain.call(address_no='  ')
        maintain.call(address_no=' PY12 ')

        save = SaveRemoteAddressTool(api_client=FakeClient())
        detail = {'country': 'US', 'province': '', 'start_postcode': 'A 1-b'}
        for kwargs in (
            {'remote_name': 1, 'status': 1, 'details': [detail]},
            {'remote_name': ' ', 'status': 1, 'details': [detail]},
            {'remote_name': '名' * 101, 'status': 1, 'details': [detail]},
            {'remote_name': '美西', 'status': True, 'details': [detail]},
            {'remote_name': '美西', 'status': 2, 'details': [detail]},
            {'remote_name': '美西', 'status': 1, 'address_no': 1, 'details': [detail]},
            {'remote_name': '美西', 'status': 1, 'address_no': 'P' * 21, 'details': [detail]},
            {'remote_name': '美西', 'status': 1},
            {'remote_name': '美西', 'status': 1, 'details': 'US'},
            {'remote_name': '美西', 'status': 1, 'details': []},
            {'remote_name': '美西', 'status': 1, 'details': [{'country': 'US'}] * 201},
            {'remote_name': '美西', 'status': 1, 'details': ['US']},
            {'remote_name': '美西', 'status': 1, 'details': [{'country': 'US', 'extra': '1'}]},
            {'remote_name': '美西', 'status': 1, 'details': [{'country': 1}]},
            {'remote_name': '美西', 'status': 1, 'details': [{'country': 'U' * 51}]},
            {'remote_name': '美西', 'status': 1, 'details': [{'province': 'CA'}]},
            {'remote_name': '美西', 'status': 1, 'details': [{'shipper_country': 'CN', 'shipper_end_postcode': '518099'}]},
        ):
            with self.assertRaises(ValueError):
                save.call(**kwargs)
        client = FakeClient()
        SaveRemoteAddressTool(api_client=client).call(
            remote_name='美西',
            status=1,
            address_no='   ',
            details=[detail],
        )
        self.assertNotIn('address_no', client.calls[0][2])
        SaveRemoteAddressTool(api_client=client).call(
            address_no='PY12',
            remote_name='美西',
            status=0,
        )

    def test_presenter_rejects_bad_maintain_and_save_payloads(self):
        presenter = OutputPresenter()
        columns = [
            {'key': key, 'name': name}
            for key, name in OutputPresenter.REMOTE_ADDRESS_DETAIL_COLUMNS
        ]
        record = {key: '' for key, _name in OutputPresenter.REMOTE_ADDRESS_DETAIL_COLUMNS}
        base = {
            'summary': '准备新增地址库',
            'mode_text': '新增',
            'address_no': '',
            'remote_name': '',
            'status_text': '',
            'temp_key': 'temp',
            'columns': columns,
            'records': [record],
        }
        bad_payloads = [None, dict(base, extra=1), dict(base, mode_text='查看')]
        missing = dict(base)
        missing.pop('summary')
        bad_payloads.append(missing)
        bad_payloads.append(dict(base, summary=1))
        bad_payloads.append(dict(base, columns=[]))
        bad_payloads.append(dict(base, records={}))
        bad_columns = [dict(column) for column in columns]
        bad_columns[0] = {'key': 'other', 'name': columns[0]['name']}
        bad_payloads.append(dict(base, columns=bad_columns))
        bad_payloads.append(dict(base, records=[{}]))
        for data in bad_payloads:
            result = presenter.present('maintain_remote_address', {
                'code': 'MCP_0000',
                'msg': 'success',
                'data': data,
                'meta': {'request_id': 'rq_maintain'},
            })
            self.assertIn('格式异常', str(result))

        tipped = dict(base, tips=['该地址库还没有地址明细'])
        tipped_result = presenter.present('maintain_remote_address', {
            'code': 'MCP_0000',
            'msg': 'success',
            'data': tipped,
            'meta': {'request_id': 'rq_maintain'},
        })
        self.assertIn('该地址库还没有地址明细', str(tipped_result))

        saved = {
            'summary': '已新增地址库 PY12 美西，状态启用',
            'mode_text': '新增',
            'address_no': 'PY12',
            'remote_name': '美西',
            'status_text': '启用',
        }
        for data in (
            dict(saved, extra=1),
            dict(saved, mode_text='查看'),
            dict(saved, status_text='草稿'),
            dict(saved, address_no='12'),
            dict(saved, remote_name=' '),
            dict(saved, summary=1),
        ):
            result = presenter.present('save_remote_address', {
                'code': 'MCP_0000',
                'msg': 'success',
                'data': data,
                'meta': {'request_id': 'rq_save_remote'},
            })
            self.assertIn('格式异常', str(result))

        for message in (
            '一次最多保存200条地址明细',
            '第1条收件人邮编不正确',
            '不能保存',
        ):
            result = presenter.present('save_remote_address', {
                'code': 'MCP_1401',
                'msg': message,
                'data': {},
                'meta': {'request_id': 'rq_save_remote'},
            })
            rendered = str(result)
            if message == '不能保存':
                self.assertIn('参数不正确', rendered)
            else:
                self.assertIn(message, rendered)
        for tool_name in ('maintain_remote_address', 'save_remote_address'):
            result = presenter.present(tool_name, {
                'code': 'MCP_1501',
                'msg': '地址库不存在',
                'data': {},
                'meta': {'request_id': 'rq_missing'},
            })
            self.assertIn('地址库不存在', str(result))


class FakeClient:
    def __init__(self):
        self.calls = []

    def call_tool(self, name, route, payload, request_id):
        self.calls.append((name, route, payload, request_id))
        return {'ok': True}


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