mcp/tests/test_public_server_coverage.py

"""
Coverage补充测试：public_server.py

目标路径：
- extract_client_ip — 空 client_address
- do_GET — 非 /health 路径 → 404
- do_POST — Content-Type 正确但 body 为无效 JSON → -32700 parse error
- PublicMcpHttpHandler.handle_json_rpc — 未知 method → -32601 error
"""
import json
import socket
import threading
import unittest
from http.server import HTTPServer

from public_server import PublicMcpHttpHandler, create_http_handler, extract_client_ip


# ---------------------------------------------------------------------------
# extract_client_ip 边界情况
# ---------------------------------------------------------------------------

class ExtractClientIpTest(unittest.TestCase):
    def test_returns_empty_string_when_client_address_is_none(self):
        result = extract_client_ip({}, None)
        self.assertEqual('', result)

    def test_returns_empty_string_when_client_address_is_empty_tuple(self):
        result = extract_client_ip({}, ())
        self.assertEqual('', result)

    def test_returns_ip_from_client_address(self):
        result = extract_client_ip({'X-Forwarded-For': '1.2.3.4'}, ('10.0.0.1', 12345))
        self.assertEqual('10.0.0.1', result)


# ---------------------------------------------------------------------------
# PublicMcpHttpHandler.handle_json_rpc — 未知 method
# ---------------------------------------------------------------------------

class FakeGateway:
    def list_tools(self):
        return [{'name': 'query_order', 'description': '', 'input_schema': {'type': 'object'}}]

    def call_tool(self, *args, **kwargs):
        return {'data': {}}


class HandleJsonRpcUnknownMethodTest(unittest.TestCase):
    def test_unknown_method_returns_method_not_found(self):
        handler = PublicMcpHttpHandler(FakeGateway())
        response = handler.handle_json_rpc(
            headers={},
            message={'jsonrpc': '2.0', 'id': 1, 'method': 'unknown/method'},
        )
        self.assertIn('error', response)
        self.assertEqual(-32601, response['error']['code'])


# ---------------------------------------------------------------------------
# 集成测试：启动真实 HTTP 服务器，覆盖 do_GET 404 和 do_POST JSON 错误
# ---------------------------------------------------------------------------

def _find_free_port():
    with socket.socket() as s:
        s.bind(('127.0.0.1', 0))
        return s.getsockname()[1]


class HttpHandlerIntegrationTest(unittest.TestCase):
    """
    启动一个真实的 ThreadingHTTPServer（随机端口），测试
    HTTP 层面的边界路径。
    """

    @classmethod
    def setUpClass(cls):
        port = _find_free_port()
        gateway = FakeGateway()
        handler_class = create_http_handler(gateway, rate_limiter=None)
        cls.server = HTTPServer(('127.0.0.1', port), handler_class)
        cls.server_thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.server_thread.start()
        cls.base_url = 'http://127.0.0.1:{0}'.format(port)

    @classmethod
    def tearDownClass(cls):
        cls.server.shutdown()

    def _get(self, path):
        import urllib.request
        req = urllib.request.Request(cls_url := self.base_url + path, method='GET')
        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                return resp.status, resp.read()
        except Exception as e:
            # urllib raises HTTPError for non-2xx
            if hasattr(e, 'code'):
                return e.code, b''
            raise

    def _post(self, path, body, content_type='application/json'):
        import urllib.request
        data = body.encode('utf-8') if isinstance(body, str) else body
        req = urllib.request.Request(
            self.base_url + path,
            data=data,
            headers={'Content-Type': content_type, 'Content-Length': str(len(data))},
            method='POST',
        )
        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                return resp.status, json.loads(resp.read())
        except Exception as e:
            if hasattr(e, 'code') and hasattr(e, 'read'):
                body_bytes = e.read()
                return e.code, json.loads(body_bytes) if body_bytes else {}
            raise

    # do_GET /health → 200
    def test_get_health_returns_200(self):
        status, body = self._get('/health')
        self.assertEqual(200, status)

    # do_GET 非 /health → 404（覆盖 lines 113-114）
    def test_get_unknown_path_returns_404(self):
        import urllib.request
        req = urllib.request.Request(self.base_url + '/not-a-real-path', method='GET')
        try:
            urllib.request.urlopen(req, timeout=5)
            self.fail('Expected 404')
        except Exception as e:
            self.assertEqual(404, e.code)

    # do_POST 无效 JSON → -32700 parse error（覆盖 lines 127-130）
    def test_post_invalid_json_returns_parse_error(self):
        status, body = self._post('/mcp', 'this is not json at all')
        self.assertEqual(200, status)
        self.assertIn('error', body)
        self.assertEqual(-32700, body['error']['code'])
        self.assertEqual('Parse error', body['error']['message'])

    # do_POST 到非 /mcp 路径 → 404
    def test_post_wrong_path_returns_404(self):
        import urllib.request
        data = json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'initialize'}).encode()
        req = urllib.request.Request(
            self.base_url + '/wrong',
            data=data,
            headers={'Content-Type': 'application/json', 'Content-Length': str(len(data))},
            method='POST',
        )
        try:
            urllib.request.urlopen(req, timeout=5)
            self.fail('Expected 404')
        except Exception as e:
            self.assertEqual(404, e.code)

    # do_POST 正常 initialize → 200
    def test_post_initialize_returns_200(self):
        status, body = self._post('/mcp', json.dumps({
            'jsonrpc': '2.0',
            'id': 1,
            'method': 'initialize',
            'params': {},
        }))
        self.assertEqual(200, status)
        self.assertIn('result', body)
        self.assertIn('protocolVersion', body['result'])


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