mcp/tests/test_export_receivable_cost_list_tool.py

import io
import unittest

from app import GatewayApp
from public_gateway import PublicGatewayApp
from services.output_presenter import OutputPresenter
from tools.export_receivable_cost_list import ExportReceivableCostListTool


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

    def list_enabled_tools(self, request_id=''):
        return {
            'code': 'MCP_0000',
            'data': {'tool_codes': ['export_receivable_cost_list']},
        }

    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': {
                'task_ref': 'mexp_test',
                'status': 'queued',
                'retry_after_seconds': 10,
            },
        }


class ExportReceivableCostListToolTest(unittest.TestCase):
    def test_schema_is_closed_and_matches_list_filters_without_pagination(self):
        metadata = ExportReceivableCostListTool().metadata()
        schema = metadata['input_schema']
        self.assertEqual('export_receivable_cost_list', metadata['name'])
        self.assertFalse(schema['additionalProperties'])
        self.assertEqual({
            'reference_numbers', 'tracking_numbers', 'order_numbers',
            'bill_numbers', 'business_date_start', 'business_date_end',
            'customer_id', 'sub_customer_id', 'billing_status',
            'verification_status', 'document_type', 'cost_type_id',
        }, set(schema['properties']))
        self.assertNotIn('page', schema['properties'])
        self.assertNotIn('limit', schema['properties'])
        self.assertNotIn('ids', schema['properties'])
        self.assertIn('明确要求导出应收费用单', metadata['description'])
        self.assertIn('query_receivable_cost_list', metadata['description'])
        self.assertIn('query_export_task', metadata['description'])
        self.assertIn('禁止在一次调用内轮询', metadata['description'])

    def test_call_forwards_exact_filters(self):
        client = RecordingApiClient()
        result = ExportReceivableCostListTool(client).call(
            order_numbers=[' ORDER-1 '],
            billing_status=0,
        )
        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual(
            (
                'export_receivable_cost_list',
                '/mcp/tools/exportReceivableCostList',
                {
                    'order_numbers': ['ORDER-1'],
                    'billing_status': 0,
                },
                'rq_export_receivable_cost_list',
            ),
            client.calls[0],
        )

    def test_call_validation_boundaries(self):
        tool = ExportReceivableCostListTool()
        with self.assertRaisesRegex(RuntimeError, 'api client is required'):
            tool.call(order_numbers=['O1'])

        client = RecordingApiClient()
        tool = ExportReceivableCostListTool(client)
        with self.assertRaisesRegex(ValueError, 'non-empty list'):
            tool.call(order_numbers='O1')
        with self.assertRaisesRegex(ValueError, 'must be strings'):
            tool.call(order_numbers=[1])
        with self.assertRaisesRegex(ValueError, '1 to 100 chars'):
            tool.call(order_numbers=[''])
        with self.assertRaisesRegex(ValueError, '1 to 100 chars'):
            tool.call(order_numbers=['x' * 101])
        with self.assertRaisesRegex(ValueError, 'at most 200'):
            tool.call(order_numbers=['n{0}'.format(i) for i in range(201)])
        with self.assertRaisesRegex(ValueError, 'both start and end'):
            tool.call(business_date_start='2026-07-01')
        with self.assertRaisesRegex(ValueError, 'within 31 days'):
            tool.call(
                business_date_start='2026-07-01',
                business_date_end='2026-08-01',
            )
        with self.assertRaisesRegex(ValueError, 'number filters or a business'):
            tool.call()
        with self.assertRaisesRegex(ValueError, 'requires customer_id'):
            tool.call(order_numbers=['O1'], sub_customer_id=9)

        ExportReceivableCostListTool(client).call(
            order_numbers=['ORDER-1', ' ORDER-1 '],
        )
        self.assertEqual({'order_numbers': ['ORDER-1']}, client.calls[-1][2])

        ExportReceivableCostListTool(client).call(
            business_date_start='2026-07-01',
            business_date_end='2026-07-31',
            customer_id=7,
            sub_customer_id=9,
            verification_status=-1,
            document_type=8,
            cost_type_id=12,
        )
        self.assertEqual(
            {
                'business_date_start': '2026-07-01',
                'business_date_end': '2026-07-31',
                'customer_id': 7,
                'sub_customer_id': 9,
                'verification_status': -1,
                'document_type': 8,
                'cost_type_id': 12,
            },
            client.calls[-1][2],
        )

    def test_cli_forwards_export_filters(self):
        client = RecordingApiClient()
        app = GatewayApp(api_client=client)
        code = app.run_cli([
            'call', '--tool', 'export_receivable_cost_list',
            '--order-numbers', 'ORDER-1',
        ], stdout=io.StringIO())
        self.assertEqual(0, code)
        self.assertEqual({'order_numbers': ['ORDER-1']}, client.calls[-1][2])

        code = app.run_cli([
            'call', '--tool', 'export_receivable_cost_list',
            '--order-numbers', 'ORDER-1',
            '--business-date-start', '2026-07-01',
            '--business-date-end', '2026-07-31',
            '--customer-id', '7',
            '--sub-customer-id', '9',
            '--billing-status', '0',
            '--verification-status', '-1',
            '--document-type', '8',
            '--cost-type-id', '12',
        ], stdout=io.StringIO())
        self.assertEqual(0, code)
        self.assertEqual('export_receivable_cost_list', client.calls[-1][0])
        self.assertEqual(
            {
                'order_numbers': ['ORDER-1'],
                'business_date_start': '2026-07-01',
                'business_date_end': '2026-07-31',
                'customer_id': 7,
                'sub_customer_id': 9,
                'billing_status': 0,
                'verification_status': -1,
                'document_type': 8,
                'cost_type_id': 12,
            },
            client.calls[-1][2],
        )

    def test_local_and_public_registries_include_export_tool(self):
        local = GatewayApp().registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()
        self.assertEqual(local, public)
        self.assertEqual(22, len(local))
        self.assertIn('export_receivable_cost_list', local)
        self.assertEqual(21, len(OutputPresenter.SAFE_TOOLS))

    def test_presenter_reuses_queued_export_contract(self):
        presented = OutputPresenter().present(
            'export_receivable_cost_list',
            {
                'code': 'MCP_0000',
                'data': {
                    'task_ref': 'mexp_test',
                    'status': 'queued',
                    'retry_after_seconds': 10,
                },
            },
        )
        self.assertFalse(presented['is_error'])
        task = presented['structured_content']['task']
        self.assertEqual('queued', task['status'])
        self.assertEqual('mexp_test', task['task_ref'])
        self.assertEqual(10, task['retry_after_seconds'])

    def test_wrong_route_path_fails_and_restore_passes(self):
        tool = ExportReceivableCostListTool()
        original = tool.route_path
        tool.route_path = '/mcp/tools/exportReceivableCostListWrong'
        self.assertNotEqual(
            '/mcp/tools/exportReceivableCostList',
            tool.route_path,
        )
        tool.route_path = original
        self.assertEqual(
            '/mcp/tools/exportReceivableCostList',
            tool.route_path,
        )


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