""" 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 unittest.mock import MagicMock, patch from public_server import ( PublicMcpHttpHandler, create_http_handler, extract_client_ip, serve_public, ) # --------------------------------------------------------------------------- # 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 registered_tool_names(self): return ('query_order',) def list_tools(self, gateway_session_id): 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']) def test_tools_list_preserves_metadata_without_input_schema(self): gateway = FakeGateway() gateway.list_tools = MagicMock(return_value=[{'name': 'query_order'}]) handler = PublicMcpHttpHandler(gateway) response = handler.handle_json_rpc( headers={'X-Gateway-Session': 'GWS_A'}, message={'jsonrpc': '2.0', 'id': 1, 'method': 'tools/list'}, ) tool = response['result']['tools'][0] self.assertEqual({'name': 'query_order'}, tool) self.assertNotIn('inputSchema', tool) class ServePublicLifecycleTest(unittest.TestCase): def test_rate_limiter_cleanup_and_keyboard_interrupt_shutdown(self): class StopCleanupLoop(Exception): pass server = MagicMock() server.serve_forever.side_effect = KeyboardInterrupt limiter = MagicMock() with patch('public_server.SimpleRateLimiter', return_value=limiter) as limiter_class, \ patch('public_server.ThreadingHTTPServer', return_value=server) as server_class, \ patch('public_server.threading.Thread') as thread_class: serve_public( FakeGateway(), host='127.0.0.1', port='9876', rate_limit_max_requests=12, rate_limit_window_seconds=34, ) limiter_class.assert_called_once_with( max_requests=12, window_seconds=34, max_in_flight=2, ) self.assertEqual(('127.0.0.1', 9876), server_class.call_args.args[0]) thread_class.assert_called_once_with( target=thread_class.call_args.kwargs['target'], daemon=True, ) thread_class.return_value.start.assert_called_once_with() server.serve_forever.assert_called_once_with() server.shutdown.assert_called_once_with() cleanup_target = thread_class.call_args.kwargs['target'] with patch('public_server.time.sleep', side_effect=[None, StopCleanupLoop]): with self.assertRaises(StopCleanupLoop): cleanup_target() limiter.cleanup.assert_called_once_with(max_age_seconds=34) # --------------------------------------------------------------------------- # 集成测试：启动真实 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()