mcp/tests/test_headhaul_document_tools.py

import importlib
import json
import os
import unittest
from io import BytesIO
from unittest.mock import patch

import socket
import threading
from http.server import HTTPServer

from app import GatewayApp
from constants import DEVICE_INVALID_MESSAGE
from public_gateway import PublicGatewayApp
from public_server import (
    PublicMcpHttpHandler,
    create_http_handler,
    discard_request_body,
    UPLOAD_BODY_LIMIT,
)
from utils.rate_limiter import SimpleRateLimiter
from services.api_client import ApiClient, JsonTransport
from services.headhaul_document_app import (
    HeadhaulSlotMemory,
    public_upload_url,
)
from services.output_presenter import OutputPresenter
from services.scoped_api_client import ScopedApiClient
from tools.headhaul_document_common import bounded_integer, number_list
from tools.delete_headhaul_document import DeleteHeadhaulDocumentTool
from tools.list_headhaul_document_filter_options import (
    ListHeadhaulDocumentFilterOptionsTool,
)
from tools.prepare_headhaul_document_upload import PrepareHeadhaulDocumentUploadTool
from tools.query_headhaul_document_list import QueryHeadhaulDocumentListTool
from tools.save_headhaul_document import SaveHeadhaulDocumentTool


COLUMNS = [
    ('document_type_name', '文档类型'),
    ('entry_no_text', '录入单号快照'),
    ('original_name', '文件名'),
    ('safe_url', '安全链接'),
    ('operator_name', '操作人'),
    ('operate_time', '操作时间'),
]


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

    def list_enabled_tools(self, request_id=''):
        return {
            'code': 'MCP_0000',
            'data': {
                'tool_codes': [
                    'query_headhaul_document_list',
                    'list_headhaul_document_filter_options',
                    'prepare_headhaul_document_upload',
                    'save_headhaul_document',
                    'delete_headhaul_document',
                ],
            },
        }

    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': {}, 'meta': {}}


class HeadhaulDocumentToolContractTest(unittest.TestCase):
    def test_query_schema_forbids_identity_and_mixed_locator(self):
        metadata = QueryHeadhaulDocumentListTool().metadata()
        schema = metadata['input_schema']
        self.assertEqual('query_headhaul_document_list', metadata['name'])
        self.assertEqual(
            '/mcp/tools/queryHeadhaulDocumentList',
            QueryHeadhaulDocumentListTool.route_path,
        )
        self.assertFalse(schema['additionalProperties'])
        self.assertNotIn('required', schema)
        for forbidden in ('company_id', 'admin_id', 'is_super', 'target_no'):
            self.assertNotIn(forbidden, schema['properties'])
        self.assertIn('使用场景：', metadata['description'])
        self.assertIn('禁止使用：', metadata['description'])
        self.assertIn('admin/document/index', metadata['description'])

    def test_query_requires_locator_and_forwards_typed_numbers(self):
        client = RecordingApiClient()
        with self.assertRaises(ValueError):
            QueryHeadhaulDocumentListTool(client).call()
        QueryHeadhaulDocumentListTool(client).call(order_numbers=['FMS-1'])
        self.assertEqual(
            (
                'query_headhaul_document_list',
                '/mcp/tools/queryHeadhaulDocumentList',
                {
                    'order_numbers': ['FMS-1'],
                    'page': 1,
                    'limit': 20,
                },
                'rq_query_headhaul_document_list',
            ),
            client.calls[0],
        )

    def test_filter_schema_has_three_types(self):
        metadata = ListHeadhaulDocumentFilterOptionsTool().metadata()
        self.assertEqual(
            ['查询文档类型', '添加文档类型', '录入单号种类'],
            metadata['input_schema']['properties']['filter_type']['enum'],
        )
        self.assertIn('使用场景：', metadata['description'])
        self.assertIn('禁止使用：', metadata['description'])
        self.assertIn('文档类型名称 - 文件维度', metadata['description'])
        self.assertIn('完整显示名', metadata['description'])

    def test_prepare_does_not_declare_upload_widget(self):
        metadata = PrepareHeadhaulDocumentUploadTool().metadata()
        self.assertNotIn('_meta', metadata)
        self.assertEqual(['document_type_id'], metadata['input_schema']['required'])
        self.assertIn('对话窗口', metadata['description'])
        self.assertIn('选文件框', metadata['description'])
        client = RecordingApiClient()
        PrepareHeadhaulDocumentUploadTool(client).call(3)
        self.assertEqual(
            {'document_type_id': 3},
            client.calls[0][2],
        )
        with self.assertRaises(ValueError):
            PrepareHeadhaulDocumentUploadTool(client).call(True)

    def test_save_rejects_ticket_and_link_together(self):
        client = RecordingApiClient()
        with self.assertRaises(ValueError):
            SaveHeadhaulDocumentTool(client).call(
                3,
                [{'entry_no_type': 7, 'numbers': ['FMS-1']}],
                ticket_ref='mhdt_a',
                link_url='https://example.com/a',
            )
        SaveHeadhaulDocumentTool(client).call(
            3,
            [{'entry_no_type': 7, 'numbers': ['FMS-1']}],
            link_url='https://example.com/a',
        )
        self.assertEqual(
            'https://example.com/a',
            client.calls[0][2]['link_url'],
        )

    def test_delete_requires_document_ref(self):
        client = RecordingApiClient()
        with self.assertRaises(ValueError):
            DeleteHeadhaulDocumentTool(client).call(' ')
        DeleteHeadhaulDocumentTool(client).call('mhdd_opaque')
        self.assertEqual({'document_ref': 'mhdd_opaque'}, client.calls[0][2])

    def test_local_and_public_registries_include_five_tools(self):
        local = GatewayApp().registered_tool_names()
        public = PublicGatewayApp(None, None).registered_tool_names()
        self.assertEqual(local, public)
        self.assertEqual(39, len(local))
        for name in (
            'query_headhaul_document_list',
            'list_headhaul_document_filter_options',
            'prepare_headhaul_document_upload',
            'save_headhaul_document',
            'delete_headhaul_document',
        ):
            self.assertIn(name, local)
        self.assertEqual(38, len(OutputPresenter.SAFE_TOOLS))

    def test_presenter_list_keeps_six_columns_and_document_ref(self):
        presenter = OutputPresenter()
        columns = [{'key': key, 'name': name} for key, name in COLUMNS]
        result = presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {
                'columns': columns,
                'records': [{
                    'document_type_name': '报关资料',
                    'entry_no_text': 'FMS-1',
                    'original_name': 'a.pdf',
                    'safe_url': 'https://files.example/a.pdf',
                    'operator_name': '测试员',
                    'operate_time': '2026-09-18 01:00:00',
                    'document_ref': 'mhdd_abc',
                }],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False,
                'request_id': 'rq_doc',
            },
        })
        self.assertFalse(result['is_error'])
        self.assertEqual('文档身份', result['structured_content']['headers'][-1]['label'])
        self.assertEqual('mhdd_abc', result['structured_content']['rows'][0][-1])
        bad = dict(result)
        extra = {
            'code': 'MCP_0000',
            'data': {
                'columns': columns,
                'records': [{
                    'document_type_name': '报关资料',
                    'entry_no_text': 'FMS-1',
                    'original_name': 'a.pdf',
                    'safe_url': 'javascript:alert(1)',
                    'operator_name': '测试员',
                    'operate_time': '2026-09-18 01:00:00',
                    'document_ref': 'mhdd_abc',
                }],
            },
            'meta': {
                'page': 1, 'limit': 20, 'has_more': False,
                'request_id': 'rq_doc',
            },
        }
        self.assertTrue(presenter.present('query_headhaul_document_list', extra)['is_error'])

    def test_presenter_prepare_save_delete(self):
        presenter = OutputPresenter()
        prepare = presenter.present('prepare_headhaul_document_upload', {
            'code': 'MCP_0000',
            'data': {
                'upload_token': 'tok',
                'upload_url': '/mcp/upload-headhaul-document',
                'expires_in': 1800,
            },
            'meta': {'request_id': 'rq_p'},
        })
        self.assertFalse(prepare['is_error'])
        self.assertIn('对话窗口', prepare['text'])
        self.assertIn('选文件框', prepare['text'])
        self.assertNotIn('app_resource', prepare)
        self.assertNotIn('ui', prepare['meta'])
        from mcp_protocol import McpProtocolHandler
        embedded = McpProtocolHandler._presented_response('1', prepare)
        self.assertEqual(1, len(embedded['result']['content']))
        self.assertEqual('text', embedded['result']['content'][0]['type'])
        self.assertEqual('tok', prepare['structured_content']['upload_token'])
        self.assertEqual(
            public_upload_url(),
            prepare['structured_content']['upload_url'],
        )
        self.assertEqual('tok', prepare['meta']['upload_token'])
        self.assertEqual('rq_p', prepare['meta']['request_id'])
        with patch.dict(os.environ, {'FMS_GATEWAY_PUBLIC_BASE': 'http://gateway.test'}):
            linked = presenter.present('prepare_headhaul_document_upload', {
                'code': 'MCP_0000',
                'data': {
                    'upload_token': 'tok',
                    'upload_url': '/mcp/upload-headhaul-document',
                    'expires_in': 1800,
                },
                'meta': {'request_id': 'rq_p'},
            })
        self.assertEqual(
            'http://gateway.test/mcp/upload-headhaul-document',
            linked['structured_content']['upload_url'],
        )
        self.assertIn(
            'http://gateway.test/mcp/upload-headhaul-document',
            linked['text'],
        )
        save = presenter.present('save_headhaul_document', {
            'code': 'MCP_0000',
            'data': {'document_ref': 'mhdd_x'},
            'meta': {'request_id': 'rq_s'},
        })
        self.assertEqual('已保存', save['structured_content']['结果'])
        delete = presenter.present('delete_headhaul_document', {
            'code': 'MCP_0000',
            'data': {'result': '已删除'},
            'meta': {'request_id': 'rq_d'},
        })
        self.assertFalse(delete['is_error'])
        self.assertTrue(presenter.present('save_headhaul_document', {
            'code': 'MCP_0000',
            'data': {'document_ref': 'id-1'},
            'meta': {'request_id': 'rq_bad'},
        })['is_error'])

    def test_slot_memory_and_public_upload_url(self):
        memory = HeadhaulSlotMemory()
        memory.remember('', 'x')
        memory.remember('s', '')
        memory.remember('s', 'a')
        memory.remember('s', 'b')
        self.assertEqual('', memory.session_for_token('a'))
        self.assertEqual('s', memory.session_for_token('b'))
        self.assertEqual('b', memory.token_for_session('s'))
        with patch.dict(os.environ, {'FMS_GATEWAY_PUBLIC_BASE': 'https://mcp.example.com/'}):
            self.assertEqual(
                'https://mcp.example.com/mcp/upload-headhaul-document',
                public_upload_url(),
            )
        with patch.dict(os.environ, {'FMS_GATEWAY_PUBLIC_BASE': ''}):
            self.assertEqual('/mcp/upload-headhaul-document', public_upload_url())

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

            def registered_tool_names(self):
                return ('prepare_headhaul_document_upload',)

            def session_for_headhaul_token(self, token):
                return 'GWS_test' if token == 'slot-token' else ''

            def upload_headhaul_document(self, *args, **kwargs):
                self.calls.append((args, kwargs))
                return {
                    'code': 'MCP_0000',
                    'data': {'ticket_ref': 'mhdt_x'},
                    'meta': {},
                }

        class Parser:
            def parse(self, headers):
                return type('C', (), {
                    'has_session': lambda self: False,
                    'gateway_session_id': '',
                })()

        gateway = Gateway()
        handler = PublicMcpHttpHandler(gateway, context_parser=Parser())
        body = (
            b'------Bound\r\n'
            b'Content-Disposition: form-data; name="upload_token"\r\n\r\n'
            b'slot-token\r\n'
            b'------Bound\r\n'
            b'Content-Disposition: form-data; name="file"; filename="a.pdf"\r\n'
            b'Content-Type: application/pdf\r\n\r\n'
            b'%PDF\r\n'
            b'------Bound--\r\n'
        )
        status, payload = handler.handle_upload(
            {'Content-Type': 'multipart/form-data; boundary=----Bound'},
            body,
            trace_request_id='rq_slot',
        )
        self.assertEqual(200, status)
        self.assertEqual('MCP_0000', payload['code'])
        self.assertEqual('GWS_test', gateway.calls[0][0][0])

    def test_public_gateway_prepare_remembers_upload_token(self):
        class Store:
            def get(self, gateway_session_id):
                return {
                    'mcp_token': 'MT_test',
                    'admin_id': 1,
                    'company_id': 1,
                } if gateway_session_id == 'GWS_test' else None

            def touch_session(self, gateway_session_id):
                return self.get(gateway_session_id)

        class Client:
            def list_enabled_tools(self, token='', request_id=''):
                return {
                    'code': 'MCP_0000',
                    'data': {
                        'tool_codes': ['prepare_headhaul_document_upload'],
                    },
                }

            def call_tool(self, **kwargs):
                return {
                    'code': 'MCP_0000',
                    'data': {
                        'upload_token': 'prepared-token',
                        'upload_url': '/mcp/upload-headhaul-document',
                        'expires_in': 1800,
                    },
                    'meta': {'request_id': 'rq_p'},
                }

        app = PublicGatewayApp(Store(), Client())
        app.call_tool('GWS_test', 'prepare_headhaul_document_upload', {'document_type_id': 3})
        self.assertEqual('GWS_test', app.session_for_headhaul_token('prepared-token'))

        class EmptyDataClient:
            def list_enabled_tools(self, token='', request_id=''):
                return {
                    'code': 'MCP_0000',
                    'data': {
                        'tool_codes': ['prepare_headhaul_document_upload'],
                    },
                }

            def call_tool(self, **kwargs):
                return {'code': 'MCP_0000', 'data': []}

        empty_app = PublicGatewayApp(Store(), EmptyDataClient())
        empty_app.call_tool(
            'GWS_test',
            'prepare_headhaul_document_upload',
            {'document_type_id': 3},
        )
        self.assertEqual('', empty_app.session_for_headhaul_token('prepared-token'))

        class BlankTokenClient(EmptyDataClient):
            def call_tool(self, **kwargs):
                return {
                    'code': 'MCP_0000',
                    'data': {'upload_token': '  '},
                }

        blank_app = PublicGatewayApp(Store(), BlankTokenClient())
        blank_app.call_tool(
            'GWS_test',
            'prepare_headhaul_document_upload',
            {'document_type_id': 3},
        )
        self.assertEqual('', blank_app.session_for_headhaul_token('  '))

    def test_local_gateway_remembers_prepare_slot(self):
        class Client(RecordingApiClient):
            def list_enabled_tools(self, request_id=''):
                return {
                    'code': 'MCP_0000',
                    'data': {
                        'tool_codes': ['prepare_headhaul_document_upload'],
                    },
                }

            def call_tool(self, tool_code, route_path, payload, request_id):
                super().call_tool(tool_code, route_path, payload, request_id)
                return {
                    'code': 'MCP_0000',
                    'data': {'upload_token': 'local-tok'},
                }

        app = GatewayApp(api_client=Client(), token_store=None)
        app.call_tool('prepare_headhaul_document_upload', {'document_type_id': 3})
        self.assertEqual('local', app.session_for_headhaul_token('local-tok'))

        class EmptyClient(RecordingApiClient):
            def list_enabled_tools(self, request_id=''):
                return {
                    'code': 'MCP_0000',
                    'data': {
                        'tool_codes': ['prepare_headhaul_document_upload'],
                    },
                }

            def call_tool(self, tool_code, route_path, payload, request_id):
                return {'code': 'MCP_0000', 'data': None}

        empty = GatewayApp(api_client=EmptyClient(), token_store=None)
        empty.call_tool('prepare_headhaul_document_upload', {'document_type_id': 3})
        self.assertEqual('', empty.session_for_headhaul_token('local-tok'))

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

            def registered_tool_names(self):
                return ('prepare_headhaul_document_upload',)

            def upload_headhaul_document(self, *args, **kwargs):
                self.calls.append((args, kwargs))
                return {
                    'code': 'MCP_0000',
                    'data': {'ticket_ref': 'mhdt_x'},
                    'meta': {},
                }

        class Parser:
            def parse(self, headers):
                return type('C', (), {
                    'has_session': lambda self: True,
                    'gateway_session_id': 'GWS_test',
                })()

        gateway = Gateway()
        handler = PublicMcpHttpHandler(gateway, context_parser=Parser())
        body = (
            b'------Bound\r\n'
            b'Content-Disposition: form-data; name="upload_token"\r\n\r\n'
            b'tok123\r\n'
            b'------Bound\r\n'
            b'Content-Disposition: form-data; name="file"; filename="a.pdf"\r\n'
            b'Content-Type: application/pdf\r\n\r\n'
            b'%PDF\r\n'
            b'------Bound--\r\n'
        )
        status, payload = handler.handle_upload(
            {'Content-Type': 'multipart/form-data; boundary=----Bound'},
            body,
            client_ip='127.0.0.1',
            trace_request_id='rq_up',
        )
        self.assertEqual(200, status)
        self.assertEqual('MCP_0000', payload['code'])
        self.assertEqual(1, len(gateway.calls))

    def test_json_transport_builds_multipart(self):
        captured = {}

        class Response:
            def read(self):
                return b'{"code":"MCP_0000"}'

            def __enter__(self):
                return self

            def __exit__(self, exc_type, exc, tb):
                return False

        def fake_urlopen(request, timeout=0):
            captured['url'] = request.full_url
            captured['timeout'] = timeout
            captured['content_type'] = request.headers.get('Content-type')
            captured['data'] = request.data
            return Response()

        with patch('services.api_client.urllib.request.urlopen', fake_urlopen):
            result = JsonTransport().post_multipart(
                'http://tools.example/mcp/tools/uploadHeadhaulDocument',
                {'upload_token': 'tok'},
                {'file': ('a.pdf', b'%PDF', 'application/pdf')},
                {'Authorization': 'Bearer t'},
                120,
            )
        self.assertEqual({'code': 'MCP_0000'}, result)
        self.assertIn('multipart/form-data', captured['content_type'])
        self.assertIn(b'tok', captured['data'])
        self.assertIn(b'a.pdf', captured['data'])

    def test_protocol_resources_and_filter_call(self):
        from mcp_protocol import McpProtocolHandler

        class App:
            def list_resources(self):
                from services.headhaul_document_app import resource_list
                return resource_list()

            def read_resource(self, uri, gateway_session_id=''):
                from services.headhaul_document_app import read_resource
                return read_resource(uri)

        handler = McpProtocolHandler(App())
        listed = handler.handle_request({
            'jsonrpc': '2.0',
            'id': 8,
            'method': 'resources/list',
            'params': {},
        })
        self.assertEqual([], listed['result']['resources'])
        missing = handler.handle_request({
            'jsonrpc': '2.0',
            'id': 10,
            'method': 'resources/read',
            'params': {'uri': 'ui://missing'},
        })
        self.assertEqual(-32602, missing['error']['code'])

        class FoundApp:
            def read_resource(self, uri, gateway_session_id=''):
                return {'uri': uri, 'mimeType': 'text/plain', 'text': 'ok'}

        found = McpProtocolHandler(FoundApp()).handle_request({
            'jsonrpc': '2.0',
            'id': 13,
            'method': 'resources/read',
            'params': {'uri': 'ui://example'},
        })
        self.assertEqual(
            [{'uri': 'ui://example', 'mimeType': 'text/plain', 'text': 'ok'}],
            found['result']['contents'],
        )

        client = RecordingApiClient()
        ListHeadhaulDocumentFilterOptionsTool(client).call('查询文档类型')
        self.assertEqual('查询文档类型', client.calls[0][2]['filter_type'])
        with self.assertRaises(RuntimeError):
            QueryHeadhaulDocumentListTool().call(order_numbers=['X'])
        with self.assertRaises(ValueError):
            ListHeadhaulDocumentFilterOptionsTool(client).call('客户名称')

    def test_upload_http_rejects_missing_session_and_file(self):
        class Parser:
            def parse(self, headers):
                return type('C', (), {
                    'has_session': lambda self: False,
                    'gateway_session_id': '',
                })()

        handler = PublicMcpHttpHandler(
            type('G', (), {'registered_tool_names': lambda self: ()})(),
            context_parser=Parser(),
        )
        status, payload = handler.handle_upload({}, b'', trace_request_id='rq_no')
        self.assertEqual(401, status)
        self.assertEqual('MCP_1101', payload['code'])

        class OkParser:
            def parse(self, headers):
                return type('C', (), {
                    'has_session': lambda self: True,
                    'gateway_session_id': 'GWS_test',
                })()

        ok_handler = PublicMcpHttpHandler(
            type('G', (), {'registered_tool_names': lambda self: ()})(),
            context_parser=OkParser(),
        )
        status, payload = ok_handler.handle_upload(
            {'Content-Type': 'text/plain'},
            b'nope',
            trace_request_id='rq_bad',
        )
        self.assertEqual(400, status)

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

        class Client:
            def __init__(self):
                self.uploads = []

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

            def upload_headhaul_document(self, **kwargs):
                self.uploads.append(kwargs)
                return {'code': 'MCP_0000', 'data': {'ticket_ref': 'mhdt_z'}}

        client = Client()
        app = PublicGatewayApp(Store(), client)
        result = app.upload_headhaul_document(
            'GWS_test', 'tok', 'a.pdf', b'%PDF', 'application/pdf', 'rq_u', '1.1.1.1'
        )
        self.assertEqual('MCP_0000', result['code'])
        self.assertEqual('tok', client.uploads[0]['upload_token'])
        self.assertEqual([], GatewayApp().list_resources())
        self.assertIsNone(GatewayApp().read_resource('ui://widget/headhaul-document-upload.html'))
        self.assertEqual([], PublicGatewayApp(Store(), client).list_resources())
        self.assertIsNone(PublicGatewayApp(Store(), client).read_resource('ui://missing'))

        with self.assertRaisesRegex(RuntimeError, 'tool disabled'):
            PublicGatewayApp(
                Store(),
                type('C', (), {
                    'list_enabled_tools': lambda self, token, request_id='': {
                        'code': 'MCP_0000',
                        'data': {'tool_codes': []},
                    },
                })(),
            ).upload_headhaul_document(
                'GWS_test', 'tok', 'a.pdf', b'%PDF', 'application/pdf', 'rq_u',
            )
        with self.assertRaisesRegex(RuntimeError, 'upload client unavailable'):
            PublicGatewayApp(
                Store(),
                type('C', (), {
                    'list_enabled_tools': lambda self, token, request_id='': {
                        'code': 'MCP_0000',
                        'data': {
                            'tool_codes': ['prepare_headhaul_document_upload'],
                        },
                    },
                })(),
            ).upload_headhaul_document(
                'GWS_test', 'tok', 'a.pdf', b'%PDF', 'application/pdf', 'rq_u',
            )

    def test_query_and_common_validators_reject_invalid_values(self):
        client = RecordingApiClient()
        query = QueryHeadhaulDocumentListTool(client)
        with self.assertRaises(ValueError):
            query.call(document_type_id=True)
        with self.assertRaises(ValueError):
            query.call(document_type_id=0)
        with self.assertRaises(ValueError):
            query.call(original_name='   ')
        with self.assertRaises(ValueError):
            query.call(original_name='x' * 256)
        query.call(original_name='a.pdf', document_type_id=3, page=2, limit=10)
        with self.assertRaises(ValueError):
            query.call(order_numbers=['ok', 1])
        with self.assertRaises(ValueError):
            query.call(order_numbers=[''])
        with self.assertRaises(ValueError):
            query.call(order_numbers=['x' * 101])
        with self.assertRaises(ValueError):
            number_list([], 'order_numbers')
        self.assertEqual(['A'], number_list(['A', 'A'], 'order_numbers'))
        with self.assertRaises(ValueError):
            number_list(['N{0}'.format(i) for i in range(201)], 'order_numbers')
        with self.assertRaises(ValueError):
            bounded_integer(True, 'page')
        with self.assertRaises(ValueError):
            bounded_integer(0, 'page')
        with self.assertRaises(ValueError):
            bounded_integer(101, 'limit')

    def test_save_entries_and_payload_branches(self):
        with self.assertRaises(RuntimeError):
            SaveHeadhaulDocumentTool().call(
                3, [{'entry_no_type': 7, 'numbers': ['FMS-1']}],
                link_url='https://example.com/a',
            )
        client = RecordingApiClient()
        tool = SaveHeadhaulDocumentTool(client)
        with self.assertRaises(ValueError):
            tool.call(0, [{'entry_no_type': 7, 'numbers': ['FMS-1']}],
                      link_url='https://example.com/a')
        with self.assertRaises(ValueError):
            tool.call(3, [{'entry_no_type': 7, 'numbers': ['FMS-1']}],
                      link_url='ftp://example.com/a')
        with self.assertRaises(ValueError):
            tool.call(3, [], link_url='https://example.com/a')
        with self.assertRaises(ValueError):
            tool.call(3, ['bad'], link_url='https://example.com/a')
        with self.assertRaises(ValueError):
            tool.call(3, [{'entry_no_type': True, 'numbers': ['A']}],
                      link_url='https://example.com/a')
        with self.assertRaises(ValueError):
            tool.call(3, [{'entry_no_type': 7, 'numbers': []}],
                      link_url='https://example.com/a')
        with self.assertRaises(ValueError):
            tool.call(3, [{'entry_no_type': 7, 'numbers': [1]}],
                      link_url='https://example.com/a')
        with self.assertRaises(ValueError):
            tool.call(3, [{'entry_no_type': 7, 'numbers': ['  ']}],
                      link_url='https://example.com/a')
        tool.call(
            3,
            [{'entry_no_type': 7, 'numbers': ['FMS-1', 'FMS-1']}],
            ticket_ref=' mhdt_a ',
        )
        self.assertEqual('mhdt_a', client.calls[-1][2]['ticket_ref'])
        self.assertEqual(['FMS-1'], client.calls[-1][2]['entries'][0]['numbers'])
        with self.assertRaises(RuntimeError):
            PrepareHeadhaulDocumentUploadTool().call(3)
        with self.assertRaises(RuntimeError):
            DeleteHeadhaulDocumentTool().call('mhdd_x')
        with self.assertRaises(RuntimeError):
            ListHeadhaulDocumentFilterOptionsTool().call('查询文档类型')
        ListHeadhaulDocumentFilterOptionsTool(client).call(
            '添加文档类型', keyword='报关',
        )
        with self.assertRaises(ValueError):
            ListHeadhaulDocumentFilterOptionsTool(client).call(
                '查询文档类型', keyword='k' * 101,
            )
        with self.assertRaises(ValueError):
            DeleteHeadhaulDocumentTool(client).call('x' * 501)

    def test_presenter_fail_closed_on_malformed_headhaul_payloads(self):
        presenter = OutputPresenter()
        columns = [{'key': key, 'name': name} for key, name in COLUMNS]
        meta = {'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq'}
        record = {
            'document_type_name': '报关资料',
            'entry_no_text': 'FMS-1',
            'original_name': 'a.pdf',
            'safe_url': '',
            'operator_name': '测试员',
            'operate_time': '2026-09-18 01:00:00',
            'document_ref': 'mhdd_abc',
        }
        self.assertFalse(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [record]},
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000', 'data': {'columns': columns}, 'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns[:1], 'records': []},
            'meta': meta,
        })['is_error'])
        bad_columns = list(columns)
        bad_columns[0] = {'key': 'x', 'name': '文档类型'}
        self.assertTrue(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {'columns': bad_columns, 'records': []},
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': ['nope']},
            'meta': meta,
        })['is_error'])
        missing_ref = dict(record)
        missing_ref.pop('document_ref')
        self.assertTrue(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [missing_ref]},
            'meta': meta,
        })['is_error'])
        blank_ref = dict(record)
        blank_ref['document_ref'] = '  '
        self.assertTrue(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [blank_ref]},
            'meta': meta,
        })['is_error'])
        not_prefixed = dict(record)
        not_prefixed['document_ref'] = 'other'
        self.assertTrue(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [not_prefixed]},
            'meta': meta,
        })['is_error'])
        not_string = dict(record)
        not_string['original_name'] = 1
        self.assertTrue(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': [not_string]},
            'meta': meta,
        })['is_error'])
        self.assertTrue(presenter.present('query_headhaul_document_list', {
            'code': 'MCP_0000',
            'data': {'columns': columns, 'records': []},
            'meta': {'request_id': 'rq'},
        })['is_error'])
        self.assertTrue(presenter.present('prepare_headhaul_document_upload', {
            'code': 'MCP_0000',
            'data': {'upload_token': 'tok'},
            'meta': {'request_id': 'rq'},
        })['is_error'])
        self.assertTrue(presenter.present('prepare_headhaul_document_upload', {
            'code': 'MCP_0000',
            'data': {
                'upload_token': 'tok',
                'upload_url': '/other',
                'expires_in': 1800,
            },
            'meta': {'request_id': 'rq'},
        })['is_error'])
        self.assertTrue(presenter.present('prepare_headhaul_document_upload', {
            'code': 'MCP_0000',
            'data': {
                'upload_token': 'tok',
                'upload_url': '/mcp/upload-headhaul-document',
                'expires_in': 1800,
                'resource_uri': 'ui://widget/headhaul-document-upload.html',
            },
            'meta': {'request_id': 'rq'},
        })['is_error'])
        self.assertTrue(presenter.present('save_headhaul_document', {
            'code': 'MCP_0000',
            'data': {'document_ref': 'mhdd_x', 'extra': 1},
            'meta': {'request_id': 'rq'},
        })['is_error'])
        self.assertTrue(presenter.present('delete_headhaul_document', {
            'code': 'MCP_0000',
            'data': {'result': '已删除', 'extra': 1},
            'meta': {'request_id': 'rq'},
        })['is_error'])
        self.assertTrue(presenter.present('delete_headhaul_document', {
            'code': 'MCP_0000',
            'data': {'result': 'ok'},
            'meta': {'request_id': 'rq'},
        })['is_error'])

    def test_upload_clients_and_http_error_paths(self):
        transport = type('T', (), {
            'calls': [],
            'post_multipart': lambda self, *args: self.calls.append(args) or {'code': 'MCP_0000'},
        })()
        api = ApiClient(
            'http://tools.example',
            type('S', (), {'require_token': lambda self: 'MT_x'})(),
            transport=transport,
            timeout=5,
        )
        api.upload_headhaul_document('tok', 'a.pdf', b'%PDF', '', 'rq_a')
        self.assertEqual(
            'http://tools.example/mcp/tools/uploadHeadhaulDocument',
            transport.calls[0][0],
        )
        scoped = ScopedApiClient('http://tools.example/', transport=transport, timeout=3)
        with self.assertRaises(RuntimeError):
            scoped.upload_headhaul_document('', 'tok', 'a.pdf', b'%PDF')
        scoped.upload_headhaul_document(
            'MT_y', 'tok', 'a.pdf', b'%PDF', '', 'rq_b', '1.2.3.4',
        )
        self.assertEqual('1.2.3.4', transport.calls[-1][3]['X-MCP-Client-IP'])
        scoped.upload_headhaul_document('MT_y', 'tok', 'a.pdf', b'%PDF')
        self.assertNotIn('X-MCP-Client-IP', transport.calls[-1][3])

        class OkParser:
            def parse(self, headers):
                return type('C', (), {
                    'has_session': lambda self: True,
                    'gateway_session_id': 'GWS_test',
                })()

        class RaisingGateway:
            def __init__(self, exc):
                self.exc = exc

            def registered_tool_names(self):
                return ('prepare_headhaul_document_upload',)

            def upload_headhaul_document(self, *args, **kwargs):
                raise self.exc

        body = (
            b'------Bound\r\n'
            b'Content-Disposition: form-data; name="upload_token"\r\n\r\n'
            b'tok123\r\n'
            b'------Bound\r\n'
            b'Content-Disposition: form-data; name="file"; filename="a.pdf"\r\n'
            b'Content-Type: application/pdf\r\n\r\n'
            b'%PDF\r\n'
            b'------Bound--\r\n'
        )
        headers = {'Content-Type': 'multipart/form-data; boundary="----Bound"'}
        too_big = PublicMcpHttpHandler(RaisingGateway(RuntimeError('x')), OkParser())
        status, payload = too_big.handle_upload(
            headers, b'x' * (102 * 1024 * 1024 + 1), trace_request_id='rq_big',
        )
        self.assertEqual(413, status)
        for exc, expected in (
            (RuntimeError(DEVICE_INVALID_MESSAGE), 401),
            (RuntimeError('tool disabled: prepare_headhaul_document_upload'), 403),
            (RuntimeError('boom'), 500),
        ):
            handler = PublicMcpHttpHandler(RaisingGateway(exc), OkParser())
            status, payload = handler.handle_upload(headers, body, trace_request_id='rq_e')
            self.assertEqual(expected, status)
        class BadPayloadGateway:
            def registered_tool_names(self):
                return ('prepare_headhaul_document_upload',)

            def upload_headhaul_document(self, *args, **kwargs):
                return 'nope'
        status, payload = PublicMcpHttpHandler(BadPayloadGateway(), OkParser()).handle_upload(
            headers, body, trace_request_id='rq_p',
        )
        self.assertEqual(500, status)
        fields, files = PublicMcpHttpHandler._parse_multipart(
            {'CONTENT-TYPE': 'multipart/form-data; boundary=----Bound'},
            b'------Bound\r\nno-separator\r\n------Bound\r\n'
            b'Content-Disposition: form-data; filename="x.pdf"\r\n\r\n'
            b'data\r\n------Bound--\r\n',
        )
        self.assertEqual({}, fields)
        self.assertEqual({}, files)

        class NoResourceApp:
            def registered_tool_names(self):
                return ()

        handler = PublicMcpHttpHandler(NoResourceApp(), OkParser())
        listed = handler.handle_json_rpc(
            headers={'X-Gateway-Session': 'GWS_test'},
            message={'jsonrpc': '2.0', 'id': 1, 'method': 'resources/list'},
            trace_request_id='rq_rl',
        )
        self.assertEqual([], listed['result']['resources'])
        missing = handler.handle_json_rpc(
            headers={'X-Gateway-Session': 'GWS_test'},
            message={'jsonrpc': '2.0', 'id': 2, 'method': 'resources/read', 'params': {}},
            trace_request_id='rq_rr',
        )
        self.assertEqual(-32602, missing['error']['code'])
        from services.request_context import RequestContextParser
        no_session = PublicMcpHttpHandler(
            NoResourceApp(), RequestContextParser(),
        ).handle_json_rpc(
            headers={},
            message={'jsonrpc': '2.0', 'id': 3, 'method': 'resources/list'},
            trace_request_id='rq_ns',
        )
        self.assertEqual(-32001, no_session['error']['code'])

        class ResourceApp:
            def registered_tool_names(self):
                return ()

            def list_resources(self):
                from services.headhaul_document_app import resource_list
                return resource_list()

            def read_resource(self, uri, gateway_session_id=''):
                from services.headhaul_document_app import read_resource
                return read_resource(uri)

        ok_handler = PublicMcpHttpHandler(ResourceApp(), OkParser())
        listed = ok_handler.handle_json_rpc(
            headers={'X-Gateway-Session': 'GWS_test'},
            message={'jsonrpc': '2.0', 'id': 4, 'method': 'resources/list'},
        )
        self.assertEqual([], listed['result']['resources'])
        read = ok_handler.handle_json_rpc(
            headers={'X-Gateway-Session': 'GWS_test'},
            message={
                'jsonrpc': '2.0', 'id': 5,
                'method': 'resources/read',
                'params': {'uri': 'ui://widget/headhaul-document-upload.html'},
            },
        )
        self.assertEqual(-32602, read['error']['code'])

        from mcp_protocol import McpProtocolHandler
        empty = McpProtocolHandler(type('A', (), {})())
        listed = empty.handle_request({
            'jsonrpc': '2.0', 'id': 11, 'method': 'resources/list', 'params': {},
        })
        self.assertEqual([], listed['result']['resources'])
        missing = empty.handle_request({
            'jsonrpc': '2.0', 'id': 12,
            'method': 'resources/read',
            'params': 'ui://widget/headhaul-document-upload.html',
        })
        self.assertEqual(-32602, missing['error']['code'])
        PublicMcpHttpHandler._parse_multipart({'Accept': 'text/plain'}, b'')
        PublicMcpHttpHandler._parse_multipart(
            {'Content-Type': 'multipart/form-data; boundary=----Bound'},
            b'------Bound\r\nContent-Disposition: form-data; name="x"\r\n\r\nvalue------Bound--\r\n',
        )
        not_dict = PublicMcpHttpHandler(
            ResourceApp(), OkParser(),
        ).handle_json_rpc(
            headers={'X-Gateway-Session': 'GWS_test'},
            message={
                'jsonrpc': '2.0', 'id': 6,
                'method': 'resources/read',
                'params': 'ui://widget/headhaul-document-upload.html',
            },
        )
        self.assertEqual(-32602, not_dict['error']['code'])

        class FoundPublicApp:
            def registered_tool_names(self):
                return ()

            def read_resource(self, uri, gateway_session_id=''):
                return {'uri': uri, 'mimeType': 'text/plain', 'text': 'ok'}

        found = PublicMcpHttpHandler(FoundPublicApp(), OkParser()).handle_json_rpc(
            headers={'X-Gateway-Session': 'GWS_test'},
            message={
                'jsonrpc': '2.0', 'id': 7,
                'method': 'resources/read',
                'params': {'uri': 'ui://example'},
            },
        )
        self.assertEqual(
            [{'uri': 'ui://example', 'mimeType': 'text/plain', 'text': 'ok'}],
            found['result']['contents'],
        )

        class UploadGateway:
            def registered_tool_names(self):
                return ('prepare_headhaul_document_upload',)

            def upload_headhaul_document(self, *args, **kwargs):
                return {'code': 'MCP_0000', 'data': {'ticket_ref': 'mhdt_http'}}

        handler_class = create_http_handler(UploadGateway())
        sock = socket.socket()
        sock.bind(('127.0.0.1', 0))
        port = sock.getsockname()[1]
        sock.close()
        server = HTTPServer(('127.0.0.1', port), handler_class)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            import urllib.request
            req = urllib.request.Request(
                'http://127.0.0.1:{0}/mcp/upload-headhaul-document'.format(port),
                data=body,
                headers={
                    'Content-Type': 'multipart/form-data; boundary=----Bound',
                    'X-Gateway-Session': 'GWS_test',
                },
                method='POST',
            )
            with urllib.request.urlopen(req, timeout=5) as resp:
                payload = json.loads(resp.read())
            self.assertEqual('MCP_0000', payload['code'])
            empty_req = urllib.request.Request(
                'http://127.0.0.1:{0}/mcp/upload-headhaul-document'.format(port),
                data=b'',
                headers={
                    'Content-Type': 'multipart/form-data; boundary=----Bound',
                    'X-Gateway-Session': 'GWS_test',
                },
                method='POST',
            )
            try:
                urllib.request.urlopen(empty_req, timeout=5)
                self.fail('expected 400')
            except Exception as exc:
                self.assertEqual(400, exc.code)
        finally:
            server.shutdown()
            server.server_close()

    def test_upload_gate_rejects_oversize_rate_and_concurrency(self):
        from http.client import HTTPConnection

        discard_request_body(BytesIO(b'abcd'), None, 0)
        discard_request_body(None, None, 10)
        buf = BytesIO(b'x' * 200)
        discard_request_body(buf, None, 200, chunk=32)
        self.assertEqual(200, buf.tell())
        short = BytesIO(b'xy')
        discard_request_body(short, None, 50, chunk=32)
        self.assertEqual(2, short.tell())
        class BoomConn:
            def gettimeout(self):
                raise OSError('no timeout')
        discard_request_body(BytesIO(b'ab'), BoomConn(), 2)
        class RestoreBoom:
            def __init__(self):
                self.calls = 0
            def gettimeout(self):
                return 9
            def settimeout(self, value):
                self.calls += 1
                if self.calls > 1:
                    raise OSError('restore failed')
        discard_request_body(BytesIO(b'ab'), RestoreBoom(), 2)
        class ReadBoom:
            def read(self, size):
                raise OSError('read failed')
        class OkConn:
            def gettimeout(self):
                return 1
            def settimeout(self, value):
                return None
        discard_request_body(ReadBoom(), OkConn(), 8)

        handler = PublicMcpHttpHandler(
            type('G', (), {'registered_tool_names': lambda self: ()})(),
        )
        bad, payload, key = handler.gate_upload({}, -1, '1.1.1.1', 'rq')
        self.assertEqual(400, bad)
        self.assertEqual('MCP_1401', payload['code'])
        self.assertIsNone(key)
        too_big, payload, key = handler.gate_upload(
            {}, UPLOAD_BODY_LIMIT + 1, '1.1.1.1', 'rq',
        )
        self.assertEqual(413, too_big)
        self.assertIsNone(key)
        self.assertEqual(400, handler.gate_upload({}, True, '1.1.1.1', 'rq')[0])

        limiter = SimpleRateLimiter(max_requests=1, window_seconds=60, max_in_flight=1)
        limited = PublicMcpHttpHandler(
            type('G', (), {'registered_tool_names': lambda self: ()})(),
            rate_limiter=limiter,
        )
        first = limited.gate_upload(
            {'X-Gateway-Session': 'GWS_test'}, 10, '1.1.1.1', 'rq',
        )
        self.assertIsNone(first[0])
        second = limited.gate_upload(
            {'X-Gateway-Session': 'GWS_test'}, 10, '1.1.1.1', 'rq',
        )
        self.assertEqual(429, second[0])
        self.assertEqual('MCP_9001', second[1]['code'])
        limiter.release(first[2])

        busy = SimpleRateLimiter(max_requests=10, window_seconds=60, max_in_flight=1)
        concurrent = PublicMcpHttpHandler(
            type('G', (), {'registered_tool_names': lambda self: ()})(),
            rate_limiter=busy,
        )
        held = concurrent.gate_upload(
            {'X-Gateway-Session': 'GWS_a'}, 10, '1.1.1.1', 'rq',
        )
        self.assertIsNone(held[0])
        blocked = concurrent.gate_upload(
            {'X-Gateway-Session': 'GWS_a'}, 10, '1.1.1.1', 'rq',
        )
        self.assertEqual(429, blocked[0])
        busy.release(held[2])
        ip_only = concurrent.gate_upload({}, 10, '8.8.8.8', 'rq')
        self.assertIsNone(ip_only[0])
        busy.release(ip_only[2])

        class RejectGateway:
            def registered_tool_names(self):
                return ()

            def upload_headhaul_document(self, *args, **kwargs):
                raise AssertionError('must not read oversize body')

        handler_class = create_http_handler(RejectGateway())
        sock = socket.socket()
        sock.bind(('127.0.0.1', 0))
        port = sock.getsockname()[1]
        sock.close()
        server = HTTPServer(('127.0.0.1', port), handler_class)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            conn = HTTPConnection('127.0.0.1', port, timeout=5)
            conn.putrequest('POST', '/mcp/upload-headhaul-document')
            conn.putheader('Content-Type', 'multipart/form-data; boundary=----Bound')
            conn.putheader('Content-Length', str(UPLOAD_BODY_LIMIT + 1))
            conn.putheader('X-Gateway-Session', 'GWS_test')
            conn.endheaders()
            resp = conn.getresponse()
            self.assertEqual(413, resp.status)
            oversize = json.loads(resp.read())
            self.assertEqual('MCP_1401', oversize['code'])
            self.assertIn('100M', oversize['msg'])
            conn.close()

            conn = HTTPConnection('127.0.0.1', port, timeout=5)
            conn.putrequest('POST', '/mcp/upload-headhaul-document')
            conn.putheader('Content-Type', 'multipart/form-data; boundary=----Bound')
            conn.putheader('Content-Length', 'nope')
            conn.putheader('X-Gateway-Session', 'GWS_test')
            conn.endheaders()
            resp = conn.getresponse()
            self.assertEqual(400, resp.status)
            conn.close()
        finally:
            server.shutdown()
            server.server_close()

        class OkGateway:
            def registered_tool_names(self):
                return ('prepare_headhaul_document_upload',)

            def upload_headhaul_document(self, *args, **kwargs):
                return {'code': 'MCP_0000', 'data': {'ticket_ref': 'mhdt_ok'}}

        ok_limiter = SimpleRateLimiter(
            max_requests=10, window_seconds=60, max_in_flight=2,
        )
        ok_handler = create_http_handler(OkGateway(), ok_limiter)
        sock = socket.socket()
        sock.bind(('127.0.0.1', 0))
        port = sock.getsockname()[1]
        sock.close()
        server = HTTPServer(('127.0.0.1', port), ok_handler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            upload_body = (
                b'------Bound\r\n'
                b'Content-Disposition: form-data; name="upload_token"\r\n\r\n'
                b'tok123\r\n'
                b'------Bound\r\n'
                b'Content-Disposition: form-data; name="file"; filename="a.pdf"\r\n'
                b'Content-Type: application/pdf\r\n\r\n'
                b'%PDF\r\n'
                b'------Bound--\r\n'
            )
            conn = HTTPConnection('127.0.0.1', port, timeout=5)
            conn.putrequest('POST', '/mcp/upload-headhaul-document')
            conn.putheader('Content-Type', 'multipart/form-data; boundary=----Bound')
            conn.putheader('Content-Length', str(len(upload_body)))
            conn.putheader('X-Gateway-Session', 'GWS_test')
            conn.endheaders()
            conn.send(upload_body)
            resp = conn.getresponse()
            self.assertEqual(200, resp.status)
            payload = json.loads(resp.read())
            self.assertEqual('MCP_0000', payload['code'])
            conn.close()
        finally:
            server.shutdown()
            server.server_close()