mcp/tests/test_inbound_receipt_tools.py

import unittest

from app import GatewayApp
from public_gateway import PublicGatewayApp
from services.output_presenter import OutputPresenter
from tools.export_inbound_receipt import ExportInboundReceiptTool
from tools.list_inbound_receipt_templates import ListInboundReceiptTemplatesTool


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

    def call_tool(self, tool_name, route_path, payload, request_id):
        self.calls.append((tool_name, route_path, payload, request_id))
        return {'code': 'MCP_0000', 'data': {}}


class InboundReceiptToolTest(unittest.TestCase):
    def test_local_and_public_registries_match(self):
        local = set(GatewayApp(api_client=object()).registered_tool_names())
        public = set(PublicGatewayApp(object(), object()).registered_tool_names())
        self.assertEqual(local, public)
        self.assertIn('list_inbound_receipt_templates', local)
        self.assertIn('export_inbound_receipt', local)

    def test_list_tool_strips_and_dedupes_container_codes(self):
        api = _Api()
        tool = ListInboundReceiptTemplatesTool(api)
        tool.call([' CONT-1 ', 'CONT-1'], request_id='rq_list')
        self.assertEqual(
            ('list_inbound_receipt_templates', '/mcp/tools/listInboundReceiptTemplates',
             {'container_codes': ['CONT-1']}, 'rq_list'),
            api.calls[0]
        )
        self.assertIn('container_codes', tool.metadata()['input_schema']['required'])

    def test_export_requires_selected_template_and_does_not_wait(self):
        api = _Api()
        tool = ExportInboundReceiptTool(api)
        description = tool.metadata()['description']
        self.assertIn('query_export_task', description)
        self.assertIn('不会在本次调用中等待文件生成', description)
        tool.call(['CONT-1'], ' FMS-INBOUND-LAX ', request_id='rq_export')
        self.assertEqual('FMS-INBOUND-LAX', api.calls[0][2]['template_code'])
        with self.assertRaises(ValueError):
            tool.call(['CONT-1'], '   ')

    def test_presenter_shows_kind_and_passes_container_rejection(self):
        presenter = OutputPresenter()
        shown = presenter.present('list_inbound_receipt_templates', {
            'code': 'MCP_0000',
            'data': {
                'receipt_kind': '直送入仓单',
                'templates': [
                    {'name': '直送柜', 'template_code': 'FMS-INBOUND-ZSG'},
                ],
            },
            'meta': {'request_id': 'rq_ok'},
        })
        self.assertFalse(shown['is_error'])
        self.assertEqual('直送入仓单', shown['structured_content']['receipt_kind'])
        self.assertNotIn('id', shown['structured_content']['rows'][0])

        rejected = presenter.present('list_inbound_receipt_templates', {
            'code': 'MCP_1601',
            'msg': '这些柜号没有对应的排舱单：NONE-1',
            'data': {},
            'meta': {'request_id': 'rq_missing'},
        })
        self.assertTrue(rejected['is_error'])
        self.assertIn('NONE-1', rejected['text'])

        hidden = presenter.present('export_inbound_receipt', {
            'code': 'MCP_1601',
            'msg': 'SQLSTATE select * from fms_outbound',
            'data': {},
            'meta': {'request_id': 'rq_sql'},
        })
        self.assertNotIn('SQLSTATE', hidden['text'])
        self.assertIn('没有可导出的数据', hidden['text'])


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