""" Coverage补充测试：mcp_protocol.py 目标路径： - run_stdio — ValueError 导致的 parse error 响应、空行跳过、非 dict message - handle_request — 未知 method、非 tools/call 的异常 - _render_text — 无 columns/track 字段时的 json.dumps fallback - _render_text — records 存在但第一条无 status/content/time 时 fallback - _render_track_like_text — location / tracking_number / shipment_id / sub_track 可选字段 - _render_track_like_text — 空 records + tips - _render_table_like_text — 空 records + tips """ import io import json import unittest from mcp_protocol import McpProtocolHandler # --------------------------------------------------------------------------- # _render_text 的 fallback 路径 # --------------------------------------------------------------------------- class RenderTextFallbackTest(unittest.TestCase): def test_empty_dict_returns_ok(self): self.assertEqual('ok', McpProtocolHandler._render_text({})) def test_none_returns_ok(self): self.assertEqual('ok', McpProtocolHandler._render_text(None)) def test_empty_list_returns_ok(self): # non-dict → 走 "not structured_content" 分支 self.assertEqual('ok', McpProtocolHandler._render_text([])) def test_dict_without_columns_or_track_fields_falls_back_to_json_dumps(self): """有 records 但第一条没有 status/content/time → 走 json.dumps fallback。""" content = {'records': [{'order_no': 'SO001', 'qty': 10}]} result = McpProtocolHandler._render_text(content) # 应该是 json.dumps 的输出，包含键名 self.assertIn('order_no', result) self.assertIn('SO001', result) def test_dict_with_no_records_falls_back_to_json_dumps(self): """有 columns 但 records 是 None（不是 list）→ columns 是 list 但 records 不是 → fallback。""" content = {'columns': [{'key': 'no', 'name': '编号'}], 'records': None} result = McpProtocolHandler._render_text(content) self.assertIn('columns', result) def test_plain_dict_no_columns_no_records_falls_back_to_json(self): content = {'foo': 'bar', 'count': 42} result = McpProtocolHandler._render_text(content) self.assertIn('foo', result) self.assertIn('bar', result) def test_records_empty_list_no_columns_falls_back_to_json(self): """records 是空 list（falsy），columns 是 None → 第二个条件 `records` falsy → fallback。""" content = {'records': []} result = McpProtocolHandler._render_text(content) # json.dumps 输出 self.assertIn('records', result) # --------------------------------------------------------------------------- # _render_table_like_text 的 tips / 空 records 路径 # --------------------------------------------------------------------------- class RenderTableEmptyRecordsTest(unittest.TestCase): def test_empty_records_with_tips_shows_tips(self): content = { 'columns': [{'key': 'no', 'name': '编号'}, {'key': 'name', 'name': '名称'}], 'records': [], 'tips': ['暂无数据', '请调整关键词'], } result = McpProtocolHandler._render_text(content) self.assertIn('表头共 2 列：', result) self.assertIn('1. 编号 (no)', result) self.assertIn('提示：暂无数据；请调整关键词', result) def test_records_with_none_value_renders_empty_string(self): content = { 'columns': [{'key': 'k', 'name': '键'}], 'records': [{'k': None}], } result = McpProtocolHandler._render_text(content) self.assertIn('- 键:', result) # --------------------------------------------------------------------------- # _render_track_like_text 的可选字段 # --------------------------------------------------------------------------- class RenderTrackOptionalFieldsTest(unittest.TestCase): """每个可选字段（location / tracking_number / shipment_id / sub_track）单独覆盖。""" def _call(self, **extra): record = {'status': '清关放行', 'content': '启运港放行', 'time': '2026-07-07 10:00:00'} record.update(extra) content = {'records': [record]} return McpProtocolHandler._render_text(content) def test_location_rendered_when_present(self): result = self._call(location='宁波港') self.assertIn('- 地点: 宁波港', result) def test_location_omitted_when_empty(self): result = self._call(location='') self.assertNotIn('地点', result) def test_location_omitted_when_absent(self): result = self._call() self.assertNotIn('地点', result) def test_tracking_number_rendered_when_present(self): result = self._call(tracking_number='TN1234567890') self.assertIn('- 快递单号: TN1234567890', result) def test_tracking_number_omitted_when_empty(self): result = self._call(tracking_number='') self.assertNotIn('快递单号', result) def test_shipment_id_rendered_when_present(self): result = self._call(shipment_id='SHP-9900') self.assertIn('- Shipment ID: SHP-9900', result) def test_shipment_id_omitted_when_empty(self): result = self._call(shipment_id='') self.assertNotIn('Shipment ID', result) def test_sub_track_rendered_when_truthy(self): result = self._call(sub_track=1) self.assertIn('- 子单轨迹: 是', result) def test_sub_track_omitted_when_zero(self): result = self._call(sub_track=0) self.assertNotIn('子单轨迹', result) def test_sub_track_omitted_when_absent(self): result = self._call() self.assertNotIn('子单轨迹', result) def test_all_optional_fields_together(self): result = self._call( location='上海港', tracking_number='YT123', shipment_id='SHP001', sub_track=2, ) self.assertIn('地点: 上海港', result) self.assertIn('快递单号: YT123', result) self.assertIn('Shipment ID: SHP001', result) self.assertIn('子单轨迹: 是', result) def test_track_with_summary_and_tips(self): content = { 'summary': '共1条轨迹', 'records': [ {'status': '已签收', 'content': '已签收', 'time': '2026-07-08 12:00:00'} ], 'tips': ['仅展示最新轨迹'], } result = McpProtocolHandler._render_text(content) self.assertIn('共1条轨迹', result) self.assertIn('已签收', result) self.assertIn('提示：仅展示最新轨迹', result) def test_multiple_track_records(self): content = { 'records': [ {'status': '已发货', 'content': '揽件', 'time': '2026-07-06 09:00:00'}, {'status': '已签收', 'content': '签收', 'time': '2026-07-08 12:00:00'}, ], } result = McpProtocolHandler._render_text(content) self.assertIn('轨迹 1：', result) self.assertIn('轨迹 2：', result) self.assertIn('已发货', result) self.assertIn('已签收', result) # --------------------------------------------------------------------------- # run_stdio — parse error 分支 # --------------------------------------------------------------------------- class RunStdioParseErrorTest(unittest.TestCase): def test_invalid_json_emits_parse_error_response(self): """无效 JSON 行 → 写出 -32700 Parse error 响应。""" handler = McpProtocolHandler(None) stdin = io.StringIO('not-valid-json\n') stdout = io.StringIO() handler.run_stdio(stdin=stdin, stdout=stdout) line = stdout.getvalue().strip() self.assertTrue(line, 'expected a response line on parse error') response = json.loads(line) self.assertIn('error', response) self.assertEqual(-32700, response['error']['code']) self.assertEqual('Parse error', response['error']['message']) self.assertIsNone(response['id']) def test_blank_lines_are_skipped(self): handler = McpProtocolHandler(None) stdin = io.StringIO('\n \n\t\n') stdout = io.StringIO() handler.run_stdio(stdin=stdin, stdout=stdout) self.assertEqual('', stdout.getvalue().strip()) def test_mixed_valid_and_invalid_lines(self): """有效请求和无效行混合；两者都产生输出。""" handler = McpProtocolHandler(None) stdin = io.StringIO( 'bad-json\n' + json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'initialize', 'params': {}}) + '\n' ) stdout = io.StringIO() # initialize 需要 gateway_app，这里简单 patch from unittest.mock import MagicMock handler.gateway_app = MagicMock() handler.gateway_app.list_tools.return_value = [] handler.run_stdio(stdin=stdin, stdout=stdout) lines = [l for l in stdout.getvalue().splitlines() if l.strip()] # parse error + initialize response = 2 lines self.assertEqual(2, len(lines)) parse_err = json.loads(lines[0]) self.assertEqual(-32700, parse_err['error']['code']) def test_stdout_without_flush_writes_all_responses(self): class WriteOnlyOutput: def __init__(self): self.values = [] def write(self, value): self.values.append(value) stdout = WriteOnlyOutput() handler = McpProtocolHandler(None) result = handler.run_stdio( stdin=io.StringIO('bad-json\nstill-bad\n'), stdout=stdout, ) self.assertEqual(0, result) self.assertEqual(2, len(stdout.values)) class ProtocolResponseBoundaryTest(unittest.TestCase): def test_normalize_tool_preserves_metadata_without_input_schema(self): tool = {'name': 'plain_tool', 'description': 'plain'} self.assertEqual(tool, McpProtocolHandler(None)._normalize_tool(tool)) def test_tool_call_response_rejects_non_dict(self): with self.assertRaisesRegex(RuntimeError, 'invalid tool response'): McpProtocolHandler._tool_call_response('1', 'query_order', 'invalid') def test_success_response_wraps_non_dict_data(self): response = McpProtocolHandler._tool_call_response( '1', 'query_order', {'code': 'MCP_0000', 'data': ['A']}, ) self.assertEqual( {'data': ['A']}, response['result']['structuredContent'], ) def test_success_response_uses_empty_structured_content_without_data(self): response = McpProtocolHandler._tool_call_response( '1', 'query_order', {'code': '0', 'data': None}, ) self.assertEqual({}, response['result']['structuredContent']) self.assertEqual('ok', response['result']['content'][0]['text']) def test_error_response_includes_data_without_meta(self): response = McpProtocolHandler._tool_call_response( '1', 'query_order', {'code': 'MCP_9001', 'msg': 'failed', 'data': ['detail']}, ) structured = response['result']['structuredContent'] self.assertEqual(['detail'], structured['data']) self.assertNotIn('meta', structured) self.assertTrue(response['result']['isError']) def test_empty_table_without_tips_returns_headers(self): result = McpProtocolHandler._render_text({ 'columns': [{'key': 'number'}], 'records': [], }) self.assertIn('1. number (number)', result) self.assertNotIn('提示', result) def test_empty_track_with_tips_returns_tip(self): result = McpProtocolHandler._render_track_like_text( {'tips': ['暂无轨迹']}, [], ) self.assertEqual('提示：暂无轨迹', result) def test_empty_track_without_tips_returns_empty_text(self): result = McpProtocolHandler._render_track_like_text({}, []) self.assertEqual('', result) # --------------------------------------------------------------------------- # handle_message / handle_request 边界情况 # --------------------------------------------------------------------------- class HandleMessageEdgeCasesTest(unittest.TestCase): def test_non_dict_message_returns_invalid_request_error(self): handler = McpProtocolHandler(None) response = handler.handle_message('not a dict') self.assertEqual(-32600, response['error']['code']) def test_notification_with_unknown_method_returns_none(self): handler = McpProtocolHandler(None) # 无 id → 通知，非 notifications/initialized → returns None response = handler.handle_message({'jsonrpc': '2.0', 'method': 'some/notification'}) self.assertIsNone(response) def test_unknown_request_method_returns_method_not_found(self): handler = McpProtocolHandler(None) response = handler.handle_request({'id': 1, 'method': 'unknown/method'}) self.assertEqual(-32601, response['error']['code']) self.assertIn('Method not found', response['error']['message']) def test_non_tools_call_exception_returns_error(self): """tools/list 抛异常 → 返回 error 响应（非 isError 内容）。""" from unittest.mock import MagicMock handler = McpProtocolHandler(MagicMock()) handler.gateway_app.list_tools.side_effect = RuntimeError('backend error') with self.assertLogs('mcp_protocol', level='ERROR') as logs: response = handler.handle_request({'id': 5, 'method': 'tools/list'}) self.assertIn('error', response) self.assertEqual(-32000, response['error']['code']) self.assertEqual('Gateway request failed. Please try again later.', response['error']['message']) self.assertNotIn('backend error', json.dumps(response)) record = logs.records[0] self.assertEqual(5, record.jsonrpc_id) self.assertTrue(record.request_id.startswith('rq_stdio_')) self.assertEqual(response['error']['data']['request_id'], record.request_id) self.assertEqual('tools/list', record.protocol_method) self.assertEqual('', record.tool_code) self.assertEqual(-32000, record.protocol_code) self.assertEqual('UNEXPECTED_EXCEPTION', record.diagnostic_reason) def test_tools_call_exception_returns_is_error_content(self): """tools/call 抛异常 → 返回 isError: true 的 content。""" from unittest.mock import MagicMock handler = McpProtocolHandler(MagicMock()) handler.gateway_app.call_tool.side_effect = RuntimeError('tool failed') response = handler.handle_request({ 'id': 6, 'method': 'tools/call', 'params': {'name': 'query_order', 'arguments': {}}, }) self.assertEqual(False, 'error' in response) self.assertTrue(response['result']['isError']) self.assertIn('tool failed', response['result']['content'][0]['text']) def test_tools_call_with_non_object_params_fails_safely(self): handler = McpProtocolHandler(None) response = handler.handle_request({ 'id': 7, 'method': 'tools/call', 'params': 'not-an-object', }) self.assertTrue(response['result']['isError']) self.assertEqual( '工具返回格式异常', response['result']['structuredContent']['message'], ) def test_tools_call_with_non_string_tool_name_fails_safely(self): from unittest.mock import MagicMock handler = McpProtocolHandler(MagicMock()) for tool_name in ([], {}): with self.subTest(tool_name=tool_name): response = handler.handle_request({ 'id': 8, 'method': 'tools/call', 'params': {'name': tool_name, 'arguments': {}}, }) self.assertEqual(-32602, response['error']['code']) self.assertNotIn('result', response) handler.gateway_app.call_tool.assert_not_called() if __name__ == '__main__': unittest.main()