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'])

    def test_container_codes_and_template_code_are_validated(self):
        listing = ListInboundReceiptTemplatesTool(api_client=_Api())
        exporting = ExportInboundReceiptTool(api_client=_Api())
        with self.assertRaises(RuntimeError):
            ListInboundReceiptTemplatesTool().call(['CONT-1'])
        with self.assertRaises(RuntimeError):
            ExportInboundReceiptTool().call(['CONT-1'], 'CODE')
        for value in (None, [], ['CONT-1'] * 201, [1], [' '], ['C' * 101]):
            with self.assertRaises(ValueError):
                listing.clean_container_codes(value)
        for value in (None, 1, ' ', 'C' * 101):
            with self.assertRaises(ValueError):
                exporting.call(['CONT-1'], value)

        presenter = OutputPresenter()
        ok = {
            'receipt_kind': '非直送入仓单',
            'templates': [{'name': ' 非直送柜 ', 'template_code': 'CODE'}],
        }
        shown = presenter.present('list_inbound_receipt_templates', {
            'code': 'MCP_0000',
            'data': ok,
            'meta': {'request_id': 'rq_ok'},
        })
        self.assertFalse(shown['is_error'])
        for data in (
            dict(ok, extra=1),
            dict(ok, receipt_kind='其他'),
            dict(ok, templates=[]),
            dict(ok, templates=[{'name': '模板'}]),
            dict(ok, templates=[{'name': ' ', 'template_code': 'CODE'}]),
            dict(ok, templates=[{'name': '模板', 'template_code': 1}]),
        ):
            result = presenter.present('list_inbound_receipt_templates', {
                'code': 'MCP_0000',
                'data': data,
                'meta': {'request_id': 'rq_bad'},
            })
            self.assertTrue(result['is_error'])
        for message in (
            '当前没有可用的入仓单模板',
            '所选模板不属于这次柜号对应的入仓单',
            '这些柜号没有对应的排舱单：',
            '这些柜号没有对应的排舱单：' + ('C' * 101),
            '这些柜号没有对应的排舱单：CONT-1、 CONT-2',
            '这些柜号没有对应的排舱单：CONT-1、\n',
            '这些柜号是直送柜：CONT-1。这些柜号是非直送柜：CONT-2。',
            '这些柜号是直送柜： CONT-1。这些柜号是非直送柜：CONT-2。',
        ):
            result = presenter.present('list_inbound_receipt_templates', {
                'code': 'MCP_1601',
                'msg': message,
                'data': {},
                'meta': {'request_id': 'rq_msg'},
            })
            if message.startswith('这些柜号是直送柜：CONT-1。') or message in (
                '当前没有可用的入仓单模板',
                '所选模板不属于这次柜号对应的入仓单',
            ):
                self.assertIn(message, result['text'])
            else:
                self.assertNotIn('CONT-1、', result['text'])


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