diff --git a/api_server.py b/api_server.py index 98c4b53..ed3a38f 100644 --- a/api_server.py +++ b/api_server.py @@ -728,9 +728,12 @@ def run_api_tests_endpoint(): 'strictness-level': 'CRITICAL', 'ignore-ssl': True, # 默认忽略SSL证书验证 # Default LLM options - 'llm-api-key': os.environ.get("OPENAI_API_KEY"), - 'llm-base-url': "https://dashscope.aliyuncs.com/compatible-mode/v1", - 'llm-model-name': "qwen-plus", + # 'llm-api-key': os.environ.get("OPENAI_API_KEY"), + 'llm-api-key': "sk-lbGrsUPL1iby86h554FaE536C343435dAa9bA65967A840B2", + # 'llm-base-url': "https://dashscope.aliyuncs.com/compatible-mode/v1", + 'llm-base-url': "https://aiproxy.petrotech.cnpc/v1", + # 'llm-model-name': "qwen-plus", + 'llm-model-name': "deepseek-v3", 'use-llm-for-request-body': False, 'use-llm-for-path-params': False, 'use-llm-for-query-params': False, diff --git a/run_api_tests.py b/run_api_tests.py index 8e5ac4d..7c50f41 100644 --- a/run_api_tests.py +++ b/run_api_tests.py @@ -87,13 +87,16 @@ def parse_args(): llm_group = parser.add_argument_group('LLM 配置选项 (可选)') llm_group.add_argument('--llm-api-key', # default=os.environ.get("OPENAI_API_KEY"), # 尝试从环境变量获取 - default='sk-0213c70194624703a1d0d80e0f762b0e', + # default='sk-0213c70194624703a1d0d80e0f762b0e', + default='sk-lbGrsUPL1iby86h554FaE536C343435dAa9bA65967A840B2', help='LLM服务的API密钥 (例如 OpenAI API Key)。默认从环境变量 OPENAI_API_KEY 读取。') llm_group.add_argument('--llm-base-url', - default="https://dashscope.aliyuncs.com/compatible-mode/v1", + # default="https://dashscope.aliyuncs.com/compatible-mode/v1", + default="https://aiproxy.petrotech.cnpc/v1", help='LLM服务的自定义基础URL (例如 OpenAI API代理)。') llm_group.add_argument('--llm-model-name', - default="qwen-plus", # 设置一个常用的默认模型 + # default="qwen-plus", # 设置一个常用的默认模型 + default="deepseek-v3", # 设置一个常用的默认模型 help='要使用的LLM模型名称 (例如 "gpt-3.5-turbo", "gpt-4")。') llm_group.add_argument('--use-llm-for-request-body', action='store_true', diff --git a/test_with_mock.py b/test_with_mock.py old mode 100755 new mode 100644 index 2a96c34..363677e --- a/test_with_mock.py +++ b/test_with_mock.py @@ -72,10 +72,10 @@ class MockServerManager: except Exception as e: print(f"[错误] 停止mock服务器失败: {e}") -def test_api_server(api_server, target_api, output_log="log_dms1.txt"): +def test_api_server(api_server, target_api, output_log="log_dms_docker.txt"): """测试API服务器""" print(f"\n[测试] API服务器功能测试...") - + # 测试配置（对应原始命令行参数） test_config = { "base-url": target_api, @@ -89,31 +89,31 @@ def test_api_server(api_server, target_api, output_log="log_dms1.txt"): "strictness-level": "CRITICAL", "ignore-ssl": True } - + print(f"[信息] 测试配置:") print(json.dumps(test_config, indent=2, ensure_ascii=False)) - + try: print(f"[信息] 发送测试请求到API服务器...") start_time = time.time() - + response = requests.post( f"{api_server}/run", json=test_config, headers={"Content-Type": "application/json"}, timeout=120 # 2分钟超时 ) - + end_time = time.time() duration = end_time - start_time - + # 保存结果到日志文件 with open(output_log, 'w', encoding='utf-8') as f: f.write(f"=== DMS合规性测试结果 ===\n") f.write(f"执行时间: {duration:.2f}秒\n") f.write(f"HTTP状态码: {response.status_code}\n") f.write(f"响应头: {dict(response.headers)}\n\n") - + if response.status_code == 200: f.write("=== 测试成功 ===\n") try: @@ -125,11 +125,11 @@ def test_api_server(api_server, target_api, output_log="log_dms1.txt"): else: f.write("=== 测试失败 ===\n") f.write(f"错误信息: {response.text}\n") - + # 显示结果 print(f"[信息] 请求完成，耗时: {duration:.2f}秒") print(f"[信息] HTTP状态码: {response.status_code}") - + if response.status_code == 200: print(f"[成功] 测试执行成功") try: @@ -142,12 +142,12 @@ def test_api_server(api_server, target_api, output_log="log_dms1.txt"): print(f" - 通过数: {summary.get('passed_tests', 0)}") print(f" - 失败数: {summary.get('failed_tests', 0)}") print(f" - 跳过数: {summary.get('skipped_tests', 0)}") - + if 'output_files' in result: print(f"[信息] 生成的文件:") for file_info in result['output_files']: print(f" - {file_info}") - + return True except: print(f"[信息] 响应内容已保存到: {output_log}") @@ -156,7 +156,7 @@ def test_api_server(api_server, target_api, output_log="log_dms1.txt"): print(f"[失败] 测试执行失败") print(f"[错误] 错误信息: {response.text[:200]}...") return False - + except requests.exceptions.Timeout: print(f"[错误] 请求超时（2分钟）") return False @@ -166,20 +166,20 @@ def test_api_server(api_server, target_api, output_log="log_dms1.txt"): def main(): print("=== DMS合规性测试工具 - 完整功能测试 ===") - + api_server = "http://localhost:5050" history_server = "http://localhost:5051" target_api = "http://host.docker.internal:5001/" # 使用Docker内部主机访问 - + # 检查mock脚本是否存在 - mock_script = "../mock_dms_server.py" + mock_script = "mock_dms_server.py" if not Path(mock_script).exists(): print(f"[警告] Mock脚本不存在: {mock_script}") print(f"[信息] 请手动启动mock服务器: python mock_dms_server.py") mock_manager = None else: mock_manager = MockServerManager(mock_script, 5001) - + try: # 1. 启动mock服务器（如果存在） if mock_manager: @@ -187,133 +187,6 @@ def main(): print(f"[错误] Mock服务器启动失败") print(f"[建议] 请手动启动: python {mock_script}") return 1 - - # 2. 测试API服务器健康检查 - print(f"\n[测试] API服务器健康检查...") - health_response = requests.get(f"{api_server}/", timeout=10) - if health_response.status_code == 200: - print(f"[成功] API服务器运行正常") - else: - print(f"[错误] API服务器不可用: {health_response.status_code}") - return 1 - - # 3. 测试历史查看器 - print(f"\n[测试] 历史查看器...") - history_response = requests.get(f"{history_server}/", timeout=10) - if history_response.status_code == 200: - print(f"[成功] 历史查看器运行正常") - else: - print(f"[警告] 历史查看器异常: {history_response.status_code}") - - # 4. 执行完整的API测试 - success = test_api_server(api_server, target_api) - - if success: - print(f"\n[成功] 所有测试完成！") - print(f"[信息] 查看详细日志: cat log_dms1.txt") - print(f"[信息] 查看测试报告: ls -la test_reports/") - return 0 - else: - print(f"\n[失败] 测试执行失败") - return 1 - - except KeyboardInterrupt: - print(f"\n[信息] 用户中断测试") - return 1 - except Exception as e: - print(f"\n[错误] 测试异常: {e}") - return 1 - finally: - # 清理mock服务器 - if mock_manager: - mock_manager.stop() - -if __name__ == "__main__": - sys.exit(main()) - - try: - print(f"[信息] 发送测试请求到API服务器...") - start_time = time.time() - - response = requests.post( - f"{api_server}/run", - json=test_config, - headers={"Content-Type": "application/json"}, - timeout=120 # 2分钟超时 - ) - - end_time = time.time() - duration = end_time - start_time - - # 保存结果到日志文件 - with open(output_log, 'w', encoding='utf-8') as f: - f.write(f"=== DMS合规性测试结果 ===\n") - f.write(f"执行时间: {duration:.2f}秒\n") - f.write(f"HTTP状态码: {response.status_code}\n") - f.write(f"响应头: {dict(response.headers)}\n\n") - - if response.status_code == 200: - f.write("=== 测试成功 ===\n") - try: - result_json = response.json() - f.write(json.dumps(result_json, indent=2, ensure_ascii=False)) - except: - f.write("响应内容（非JSON格式）:\n") - f.write(response.text) - else: - f.write("=== 测试失败 ===\n") - f.write(f"错误信息: {response.text}\n") - - # 显示结果 - print(f"[信息] 请求完成，耗时: {duration:.2f}秒") - print(f"[信息] HTTP状态码: {response.status_code}") - - if response.status_code == 200: - print(f"[成功] 测试执行成功") - try: - result = response.json() - print(f"[信息] 测试状态: {result.get('status', '未知')}") - if 'summary' in result: - summary = result['summary'] - print(f"[信息] 测试摘要: {summary}") - return True - except: - print(f"[信息] 响应内容已保存到: {output_log}") - return True - else: - print(f"[失败] 测试执行失败") - print(f"[错误] 错误信息: {response.text[:200]}...") - return False - - except requests.exceptions.Timeout: - print(f"[错误] 请求超时（2分钟）") - return False - except Exception as e: - print(f"[错误] 请求异常: {e}") - return False - -def main(): - print("=== DMS合规性测试工具 - 完整功能测试 ===") - - api_server = "http://localhost:5050" - history_server = "http://localhost:5051" - target_api = "http://127.0.0.1:5001/" - - # 检查mock脚本是否存在 - mock_script = "mock_dms_server.py" - if not Path(mock_script).exists(): - print(f"[警告] Mock脚本不存在: {mock_script}") - print(f"[信息] 将直接测试API服务器（可能会因为目标API不可用而失败）") - mock_manager = None - else: - mock_manager = MockServerManager(mock_script, 5001) - - try: - # 1. 启动mock服务器（如果存在） - if mock_manager: - if not mock_manager.start(): - print(f"[错误] Mock服务器启动失败") - return 1 # 2. 测试API服务器健康检查 print(f"\n[测试] API服务器健康检查...") @@ -337,7 +210,7 @@ def main(): if success: print(f"\n[成功] 所有测试完成！") - print(f"[信息] 查看详细日志: cat log_dms1.txt") + print(f"[信息] 查看详细日志: cat log_dms_docker.txt") print(f"[信息] 查看测试报告: ls -la test_reports/") return 0 else: