mcp/tests/test_mcp_protocol_coverage.py

"""
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')

        response = handler.handle_request({'id': 5, 'method': 'tools/list'})
        self.assertIn('error', response)
        self.assertIn('backend error', response['error']['message'])

    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.assertTrue(response['result']['isError'])
                self.assertEqual(
                    '工具返回格式异常',
                    response['result']['structuredContent']['message'],
                )

        handler.gateway_app.call_tool.assert_not_called()


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