import io import unittest from app import GatewayApp from public_gateway import PublicGatewayApp from services.output_presenter import OutputPresenter from tools.query_container_timeliness_list import ( COLUMNS, QueryContainerTimelinessListTool, ) from tools.export_container_timeliness_report import ( ExportContainerTimelinessReportTool, ) class RecordingApiClient: def __init__(self): self.calls = [] def list_enabled_tools(self, request_id=''): return { 'code': 'MCP_0000', 'data': { 'tool_codes': [ 'query_container_timeliness_list', 'export_container_timeliness_report', ], }, } 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 ContainerTimelinessToolContractTest(unittest.TestCase): def test_query_schema_is_closed_without_identity_or_page_filters(self): metadata = QueryContainerTimelinessListTool().metadata() schema = metadata['input_schema'] self.assertEqual('query_container_timeliness_list', metadata['name']) self.assertEqual( '/mcp/tools/queryContainerTimelinessList', QueryContainerTimelinessListTool.route_path, ) self.assertFalse(schema['additionalProperties']) self.assertEqual([], schema['required']) self.assertEqual( { 'container_codes', 'bl_numbers', 'departure_time_start', 'departure_time_end', 'page', 'limit', }, set(schema['properties']), ) for forbidden in ( 'company_id', 'admin_id', 'is_super', 'outbound_numbers', 'ship_company', 'loading_id', ): self.assertNotIn(forbidden, schema['properties']) description = metadata['description'] self.assertIn('使用场景：', description) self.assertIn('禁止使用：', description) self.assertIn('admin/Report/outboundReport', description) self.assertIn('query_outbound_list', description) self.assertIn('干线实际出发时间', description) self.assertIn('54列', description) self.assertIn('号码类型不明确时必须先询问用户', description) self.assertIn('不得根据号码格式猜测', description) self.assertIn('不得跨字段或跨工具试查', description) def test_query_forwards_numbers_and_optional_departure_window(self): client = RecordingApiClient() QueryContainerTimelinessListTool(client).call( container_codes=['CONT-1', ' CONT-1 '], bl_numbers=['BL-1'], departure_time_start='2026-09-01', departure_time_end='2026-09-30', page=2, limit=10, ) self.assertEqual( ( 'query_container_timeliness_list', '/mcp/tools/queryContainerTimelinessList', { 'container_codes': ['CONT-1'], 'bl_numbers': ['BL-1'], 'departure_time_start': '2026-09-01', 'departure_time_end': '2026-09-30', 'page': 2, 'limit': 10, }, 'rq_query_container_timeliness_list', ), client.calls[-1], ) def test_query_rejects_empty_and_oversize_window(self): tool = QueryContainerTimelinessListTool(RecordingApiClient()) with self.assertRaisesRegex(ValueError, 'provide'): tool.call() with self.assertRaisesRegex(ValueError, 'departure time range requires both'): tool.call(departure_time_start='2026-09-01') with self.assertRaisesRegex(ValueError, 'departure time window'): tool.call( departure_time_start='2026-08-01', departure_time_end='2026-09-01', ) with self.assertRaisesRegex(ValueError, 'is invalid'): tool.call(departure_time_start='09/01/2026', departure_time_end='2026-09-02') with self.assertRaisesRegex(ValueError, 'at most 200'): tool.call( container_codes=['C1'], bl_numbers=['B{0}'.format(i) for i in range(200)], ) with self.assertRaisesRegex(ValueError, 'non-empty'): tool.call(container_codes=[]) with self.assertRaisesRegex(ValueError, 'strings'): tool.call(container_codes=[1]) with self.assertRaisesRegex(ValueError, '1 to 100 chars'): tool.call(container_codes=['']) with self.assertRaisesRegex(ValueError, 'at most 200 container_codes'): tool.call(container_codes=['C{0}'.format(i) for i in range(201)]) with self.assertRaisesRegex(ValueError, 'page is invalid'): tool.call(container_codes=['CONT-1'], page=True) with self.assertRaisesRegex(ValueError, 'page is invalid'): tool.call(container_codes=['CONT-1'], page=0) with self.assertRaisesRegex(ValueError, 'limit is invalid'): tool.call(container_codes=['CONT-1'], limit='20') with self.assertRaisesRegex(ValueError, '1 to 100 chars'): tool.call(container_codes=['x' * 101]) with self.assertRaisesRegex(ValueError, 'departure time window'): tool.call( departure_time_start='2026-09-30', departure_time_end='2026-09-01', ) with self.assertRaisesRegex(RuntimeError, 'api client is required'): QueryContainerTimelinessListTool().call(container_codes=['CONT-1']) with self.assertRaisesRegex(RuntimeError, 'api client is required'): ExportContainerTimelinessReportTool().call(container_codes=['CONT-1']) def test_export_schema_forbids_pagination_and_requires_async_follow_up(self): metadata = ExportContainerTimelinessReportTool().metadata() schema = metadata['input_schema'] self.assertEqual('export_container_timeliness_report', metadata['name']) self.assertFalse(schema['additionalProperties']) self.assertEqual( { 'container_codes', 'bl_numbers', 'departure_time_start', 'departure_time_end', }, set(schema['properties']), ) self.assertNotIn('page', schema['properties']) description = metadata['description'] self.assertIn('明确要求导出', description) self.assertIn('柜子时效', description) self.assertIn('异步导出任务', description) self.assertIn('query_export_task', description) self.assertIn('不会在本次调用中等待文件生成', description) self.assertIn('query_container_timeliness_list', description) def test_export_forwards_locator_without_page(self): client = RecordingApiClient() ExportContainerTimelinessReportTool(client).call( bl_numbers=['BL-1'], ) self.assertEqual( { 'bl_numbers': ['BL-1'], }, client.calls[-1][2], ) self.assertNotIn('page', client.calls[-1][2]) def test_cli_forwards_query_and_export_filters(self): client = RecordingApiClient() app = GatewayApp(api_client=client) code = app.run_cli([ 'call', '--tool', 'query_container_timeliness_list', '--container-codes', 'CONT-1', '--departure-time-start', '2026-09-01', '--departure-time-end', '2026-09-14', '--page', '2', '--limit', '10', ], stdout=io.StringIO()) self.assertEqual(0, code) self.assertEqual( { 'container_codes': ['CONT-1'], 'departure_time_start': '2026-09-01', 'departure_time_end': '2026-09-14', 'page': 2, 'limit': 10, }, client.calls[-1][2], ) code = app.run_cli([ 'call', '--tool', 'export_container_timeliness_report', '--bl-numbers', 'BL-1, BL-2', ], stdout=io.StringIO()) self.assertEqual(0, code) self.assertEqual( {'bl_numbers': ['BL-1', 'BL-2']}, client.calls[-1][2], ) self.assertNotIn('page', client.calls[-1][2]) def test_presenter_accepts_fifty_four_string_columns(self): presenter = OutputPresenter() columns = [{'key': key, 'name': name} for key, name in COLUMNS] record = {key: 'v_{0}'.format(key) for key, _ in COLUMNS} ok = presenter.present('query_container_timeliness_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [record]}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', }, }) self.assertFalse(ok['is_error']) self.assertEqual(54, len(ok['structured_content']['headers'])) self.assertEqual('v_container_code', ok['structured_content']['rows'][0][2]) numeric = dict(record) numeric['total_volume'] = 1.2 self.assertTrue(presenter.present('query_container_timeliness_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [numeric]}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', }, })['is_error']) self.assertTrue(presenter.present('query_container_timeliness_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [], 'extra': 1}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', }, })['is_error']) self.assertTrue(presenter.present('query_container_timeliness_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': []}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', 'total': 1, }, })['is_error']) bad_columns = list(columns) bad_columns[0] = {'key': 'ship_company', 'name': '错'} self.assertTrue(presenter.present('query_container_timeliness_list', { 'code': 'MCP_0000', 'data': {'columns': bad_columns, 'records': []}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', }, })['is_error']) extra = dict(record) extra['seq'] = '1' self.assertTrue(presenter.present('query_container_timeliness_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': [extra]}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', }, })['is_error']) self.assertTrue(presenter.present('query_container_timeliness_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': 'bad'}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', }, })['is_error']) short_columns = columns[:-1] self.assertTrue(presenter.present('query_container_timeliness_list', { 'code': 'MCP_0000', 'data': {'columns': short_columns, 'records': []}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', }, })['is_error']) self.assertTrue(presenter.present('query_container_timeliness_list', { 'code': 'MCP_0000', 'data': {'columns': columns, 'records': ['bad']}, 'meta': { 'page': 1, 'limit': 20, 'has_more': False, 'request_id': 'rq_x', }, })['is_error']) def test_local_and_public_registries_include_both_tools(self): local = GatewayApp().registered_tool_names() public = PublicGatewayApp(None, None).registered_tool_names() self.assertEqual(local, public) self.assertEqual(34, len(local)) self.assertEqual(33, len(OutputPresenter.SAFE_TOOLS)) self.assertIn('query_container_timeliness_list', local) self.assertIn('export_container_timeliness_report', local) if __name__ == '__main__': unittest.main()