ruoyunbai/compliance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dms deepseek

Browse Source
This commit is contained in:
gongwenxin 2025-08-13 10:08:11 +08:00
parent 9bd3cb63f3
commit 2aa94b6e83
3 changed files with 30 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
api_server.py
View File

@ -728,9 +728,12 @@ def run_api_tests_endpoint():
'strictness-level': 'CRITICAL', 'strictness-level': 'CRITICAL',
'ignore-ssl': True, # 默认忽略SSL证书验证 'ignore-ssl': True, # 默认忽略SSL证书验证
# Default LLM options # Default LLM options
'llm-api-key': os.environ.get("OPENAI_API_KEY"), # 'llm-api-key': os.environ.get("OPENAI_API_KEY"),
'llm-base-url': "https://dashscope.aliyuncs.com/compatible-mode/v1", 'llm-api-key': "sk-lbGrsUPL1iby86h554FaE536C343435dAa9bA65967A840B2",
'llm-model-name': "qwen-plus", # '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-request-body': False,
'use-llm-for-path-params': False, 'use-llm-for-path-params': False,
'use-llm-for-query-params': False, 'use-llm-for-query-params': False,

9
run_api_tests.py
View File

@ -87,13 +87,16 @@ def parse_args():
llm_group = parser.add_argument_group('LLM 配置选项 (可选)') llm_group = parser.add_argument_group('LLM 配置选项 (可选)')
llm_group.add_argument('--llm-api-key', llm_group.add_argument('--llm-api-key',
# default=os.environ.get("OPENAI_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 读取。') help='LLM服务的API密钥 (例如 OpenAI API Key)。默认从环境变量 OPENAI_API_KEY 读取。')
llm_group.add_argument('--llm-base-url', 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代理)。') help='LLM服务的自定义基础URL (例如 OpenAI API代理)。')
llm_group.add_argument('--llm-model-name', llm_group.add_argument('--llm-model-name',
default="qwen-plus", # 设置一个常用的默认模型 # default="qwen-plus", # 设置一个常用的默认模型
default="deepseek-v3", # 设置一个常用的默认模型
help='要使用的LLM模型名称 (例如 "gpt-3.5-turbo", "gpt-4")。') help='要使用的LLM模型名称 (例如 "gpt-3.5-turbo", "gpt-4")。')
llm_group.add_argument('--use-llm-for-request-body', llm_group.add_argument('--use-llm-for-request-body',
action='store_true', action='store_true',

163
test_with_mock.py Executable file → Normal file
View File

@ -72,10 +72,10 @@ class MockServerManager:
except Exception as e: except Exception as e:
print(f"[错误] 停止mock服务器失败: {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服务器""" """测试API服务器"""
print(f"\n[测试] API服务器功能测试...") print(f"\n[测试] API服务器功能测试...")
# 测试配置（对应原始命令行参数） # 测试配置（对应原始命令行参数）
test_config = { test_config = {
"base-url": target_api, "base-url": target_api,
@ -89,31 +89,31 @@ def test_api_server(api_server, target_api, output_log="log_dms1.txt"):
"strictness-level": "CRITICAL", "strictness-level": "CRITICAL",
"ignore-ssl": True "ignore-ssl": True
} }
print(f"[信息] 测试配置:") print(f"[信息] 测试配置:")
print(json.dumps(test_config, indent=2, ensure_ascii=False)) print(json.dumps(test_config, indent=2, ensure_ascii=False))
try: try:
print(f"[信息] 发送测试请求到API服务器...") print(f"[信息] 发送测试请求到API服务器...")
start_time = time.time() start_time = time.time()
response = requests.post( response = requests.post(
f"{api_server}/run", f"{api_server}/run",
json=test_config, json=test_config,
headers={"Content-Type": "application/json"}, headers={"Content-Type": "application/json"},
timeout=120 # 2分钟超时 timeout=120 # 2分钟超时
) )
end_time = time.time() end_time = time.time()
duration = end_time - start_time duration = end_time - start_time
# 保存结果到日志文件 # 保存结果到日志文件
with open(output_log, 'w', encoding='utf-8') as f: with open(output_log, 'w', encoding='utf-8') as f:
f.write(f"=== DMS合规性测试结果 ===\n") f.write(f"=== DMS合规性测试结果 ===\n")
f.write(f"执行时间: {duration:.2f}秒\n") f.write(f"执行时间: {duration:.2f}秒\n")
f.write(f"HTTP状态码: {response.status_code}\n") f.write(f"HTTP状态码: {response.status_code}\n")
f.write(f"响应头: {dict(response.headers)}\n\n") f.write(f"响应头: {dict(response.headers)}\n\n")
if response.status_code == 200: if response.status_code == 200:
f.write("=== 测试成功 ===\n") f.write("=== 测试成功 ===\n")
try: try:
@ -125,11 +125,11 @@ def test_api_server(api_server, target_api, output_log="log_dms1.txt"):
else: else:
f.write("=== 测试失败 ===\n") f.write("=== 测试失败 ===\n")
f.write(f"错误信息: {response.text}\n") f.write(f"错误信息: {response.text}\n")
# 显示结果 # 显示结果
print(f"[信息] 请求完成，耗时: {duration:.2f}秒") print(f"[信息] 请求完成，耗时: {duration:.2f}秒")
print(f"[信息] HTTP状态码: {response.status_code}") print(f"[信息] HTTP状态码: {response.status_code}")
if response.status_code == 200: if response.status_code == 200:
print(f"[成功] 测试执行成功") print(f"[成功] 测试执行成功")
try: 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('passed_tests', 0)}")
print(f" - 失败数: {summary.get('failed_tests', 0)}") print(f" - 失败数: {summary.get('failed_tests', 0)}")
print(f" - 跳过数: {summary.get('skipped_tests', 0)}") print(f" - 跳过数: {summary.get('skipped_tests', 0)}")
if 'output_files' in result: if 'output_files' in result:
print(f"[信息] 生成的文件:") print(f"[信息] 生成的文件:")
for file_info in result['output_files']: for file_info in result['output_files']:
print(f" - {file_info}") print(f" - {file_info}")
return True return True
except: except:
print(f"[信息] 响应内容已保存到: {output_log}") 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"[失败] 测试执行失败")
print(f"[错误] 错误信息: {response.text[:200]}...") print(f"[错误] 错误信息: {response.text[:200]}...")
return False return False
except requests.exceptions.Timeout: except requests.exceptions.Timeout:
print(f"[错误] 请求超时（2分钟）") print(f"[错误] 请求超时（2分钟）")
return False return False
@ -166,20 +166,20 @@ def test_api_server(api_server, target_api, output_log="log_dms1.txt"):
def main(): def main():
print("=== DMS合规性测试工具 - 完整功能测试 ===") print("=== DMS合规性测试工具 - 完整功能测试 ===")
api_server = "http://localhost:5050" api_server = "http://localhost:5050"
history_server = "http://localhost:5051" history_server = "http://localhost:5051"
target_api = "http://host.docker.internal:5001/" # 使用Docker内部主机访问 target_api = "http://host.docker.internal:5001/" # 使用Docker内部主机访问
# 检查mock脚本是否存在 # 检查mock脚本是否存在
mock_script = "../mock_dms_server.py" mock_script = "mock_dms_server.py"
if not Path(mock_script).exists(): if not Path(mock_script).exists():
print(f"[警告] Mock脚本不存在: {mock_script}") print(f"[警告] Mock脚本不存在: {mock_script}")
print(f"[信息] 请手动启动mock服务器: python mock_dms_server.py") print(f"[信息] 请手动启动mock服务器: python mock_dms_server.py")
mock_manager = None mock_manager = None
else: else:
mock_manager = MockServerManager(mock_script, 5001) mock_manager = MockServerManager(mock_script, 5001)
try: try:
# 1. 启动mock服务器（如果存在） # 1. 启动mock服务器（如果存在）
if mock_manager: if mock_manager:
@ -187,133 +187,6 @@ def main():
print(f"[错误] Mock服务器启动失败") print(f"[错误] Mock服务器启动失败")
print(f"[建议] 请手动启动: python {mock_script}") print(f"[建议] 请手动启动: python {mock_script}")
return 1 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服务器健康检查 # 2. 测试API服务器健康检查
print(f"\n[测试] API服务器健康检查...") print(f"\n[测试] API服务器健康检查...")
@ -337,7 +210,7 @@ def main():
if success: if success:
print(f"\n[成功] 所有测试完成！") print(f"\n[成功] 所有测试完成！")
print(f"[信息] 查看详细日志: cat log_dms1.txt") print(f"[信息] 查看详细日志: cat log_dms_docker.txt")
print(f"[信息] 查看测试报告: ls -la test_reports/") print(f"[信息] 查看测试报告: ls -la test_reports/")
return 0 return 0
else: else: