mcp/tests/test_export_pending_outbound_tools.py

import unittest

from app import GatewayApp
from public_gateway import PublicGatewayApp
from tools.export_pending_outbound_orders import ExportPendingOutboundOrdersTool
from tools.list_pending_outbound_export_filter_options import (
    ListPendingOutboundExportFilterOptionsTool,
)


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

    def call_tool(self, tool_code, route_path, payload, request_id):
        self.calls.append((tool_code, route_path, payload, request_id))
        return {'code': 'MCP_0000', 'data': {'file_url': 'https://files.test/order.xlsx'}}

    def list_enabled_tools(self):
        return {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': [
                    'export_pending_outbound_orders',
                    'list_pending_outbound_export_filter_options',
                ],
            },
        }


class PublicSessionStore:
    def get(self, gateway_session_id):
        return {'mcp_token': 'MT_test'} if gateway_session_id == 'GWS_test' else None


class PublicApiClient:
    def list_enabled_tools(self, token):
        return {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': [
                    'export_pending_outbound_orders',
                    'list_pending_outbound_export_filter_options',
                ],
            },
        }


class ExportPendingOutboundToolsTest(unittest.TestCase):
    def test_metadata_uses_add_page_filter_names_and_native_types(self):
        schema = ExportPendingOutboundOrdersTool().metadata()['input_schema']
        properties = schema['properties']

        for field in (
            'number', 'product_type_id', 'product_id', 'order_warehouse_id',
            'receiver_country', 'is_remove', 'address', 'inbound_date', 'status',
            'is_battery', 'is_magnetic', 'is_wood', 'is_other', 'is_fda',
            'is_toy', 'is_ultra_limit', 'is_sensitive', 'is_food', 'no_property',
            'merge_declare_number', 'importer_id', 'packing_type',
        ):
            self.assertIn(field, properties)

        self.assertEqual('array', properties['product_id']['type'])
        self.assertEqual('array', properties['status']['type'])
        self.assertEqual('string', properties['packing_type']['type'])
        self.assertNotIn('property_2', properties)

    def test_export_call_preserves_three_page_multi_select_shapes(self):
        client = RecordingApiClient()
        tool = ExportPendingOutboundOrdersTool(api_client=client)

        result = tool.call(
            product_id=[12, 13],
            status=[30, 40],
            is_battery='Y',
            packing_type='散货,整柜',
            is_remove=0,
            request_id='rq_export',
        )

        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual(
            {
                'product_id': [12, 13],
                'status': [30, 40],
                'is_battery': 'Y',
                'packing_type': '散货,整柜',
                'is_remove': 0,
            },
            client.calls[0][2],
        )
        self.assertEqual('/mcp/tools/exportPendingOutboundOrders', client.calls[0][1])

    def test_filter_options_forwards_product_type_dependency_and_bounds_page(self):
        client = RecordingApiClient()
        tool = ListPendingOutboundExportFilterOptionsTool(api_client=client)

        tool.call(
            filter_type=' product ',
            product_type_id=8,
            keyword=' 海运 ',
            page=0,
            limit=200,
            request_id='rq_filters',
        )

        self.assertEqual(
            {
                'filter_type': 'product',
                'product_type_id': 8,
                'keyword': '海运',
                'page': 1,
                'limit': 100,
            },
            client.calls[0][2],
        )
        self.assertEqual('/mcp/tools/listPendingOutboundExportFilterOptions', client.calls[0][1])

    def test_filter_options_rejects_missing_client_unknown_type_and_invalid_dependency(self):
        with self.assertRaisesRegex(RuntimeError, 'api client is required'):
            ListPendingOutboundExportFilterOptionsTool().call('country')

        tool = ListPendingOutboundExportFilterOptionsTool(api_client=RecordingApiClient())
        with self.assertRaisesRegex(ValueError, 'unsupported filter_type'):
            tool.call('unknown')
        with self.assertRaisesRegex(ValueError, 'product_type_id must be greater than 0'):
            tool.call('product', product_type_id=0)
        tool.call('country')

    def test_export_requires_api_client(self):
        with self.assertRaisesRegex(RuntimeError, 'api client is required'):
            ExportPendingOutboundOrdersTool().call()

    def test_gateways_register_both_tools_when_backend_enables_them(self):
        local_names = {
            item['name']
            for item in GatewayApp(api_client=RecordingApiClient()).list_tools()
        }
        public_names = {
            item['name']
            for item in PublicGatewayApp(
                PublicSessionStore(), PublicApiClient()
            ).list_tools('GWS_test')
        }

        self.assertIn('export_pending_outbound_orders', local_names)
        self.assertIn('list_pending_outbound_export_filter_options', local_names)
        self.assertIn('export_pending_outbound_orders', public_names)
        self.assertIn('list_pending_outbound_export_filter_options', public_names)


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