211 lines
6.3 KiB
Python
Executable File
211 lines
6.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
测试多服务Docker配置
|
||
验证API服务器和历史查看器是否都正常运行
|
||
"""
|
||
|
||
import requests
|
||
import time
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
def test_service(name, url, expected_content=None):
|
||
"""测试单个服务"""
|
||
print(f"🔍 测试{name}服务: {url}")
|
||
|
||
try:
|
||
response = requests.get(url, timeout=10)
|
||
|
||
if response.status_code == 200:
|
||
print(f"✅ {name}服务响应正常 (状态码: {response.status_code})")
|
||
|
||
if expected_content and expected_content in response.text:
|
||
print(f"✅ {name}服务内容验证通过")
|
||
return True
|
||
elif expected_content:
|
||
print(f"⚠️ {name}服务内容验证失败，但服务可访问")
|
||
return True
|
||
else:
|
||
return True
|
||
else:
|
||
print(f"❌ {name}服务响应异常 (状态码: {response.status_code})")
|
||
return False
|
||
|
||
except requests.exceptions.ConnectionError:
|
||
print(f"❌ {name}服务连接失败 - 服务可能未启动")
|
||
return False
|
||
except requests.exceptions.Timeout:
|
||
print(f"❌ {name}服务响应超时")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ {name}服务测试出错: {e}")
|
||
return False
|
||
|
||
def wait_for_services(max_wait=60):
|
||
"""等待服务启动"""
|
||
print(f"⏳ 等待服务启动（最多等待{max_wait}秒）...")
|
||
|
||
for i in range(max_wait):
|
||
try:
|
||
# 测试API服务
|
||
api_response = requests.get("http://localhost:5050", timeout=5)
|
||
if api_response.status_code == 200:
|
||
print("✅ API服务已启动")
|
||
break
|
||
except:
|
||
pass
|
||
|
||
print(f"⏳ 等待中... ({i+1}/{max_wait})")
|
||
time.sleep(1)
|
||
else:
|
||
print("❌ 服务启动超时")
|
||
return False
|
||
|
||
# 额外等待历史查看器服务
|
||
time.sleep(5)
|
||
return True
|
||
|
||
def main():
|
||
"""主测试函数"""
|
||
print("=" * 60)
|
||
print("DMS合规性测试工具 - 多服务Docker测试")
|
||
print("=" * 60)
|
||
|
||
# 等待服务启动
|
||
if not wait_for_services():
|
||
sys.exit(1)
|
||
|
||
# 测试结果
|
||
results = []
|
||
|
||
# 测试API服务器
|
||
api_result = test_service(
|
||
"API服务器",
|
||
"http://localhost:5050",
|
||
"DMS" # 期望在页面中找到DMS字样
|
||
)
|
||
results.append(("API服务器", api_result))
|
||
|
||
# 测试历史查看器
|
||
history_result = test_service(
|
||
"历史查看器",
|
||
"http://localhost:5051"
|
||
)
|
||
results.append(("历史查看器", history_result))
|
||
|
||
# 测试API端点（如果存在）
|
||
print("\n🔍 测试API端点...")
|
||
try:
|
||
# 测试一些可能存在的API端点
|
||
test_endpoints = [
|
||
"/api/health",
|
||
"/health",
|
||
"/api/status",
|
||
"/status"
|
||
]
|
||
|
||
for endpoint in test_endpoints:
|
||
try:
|
||
response = requests.get(f"http://localhost:5050{endpoint}", timeout=5)
|
||
if response.status_code == 200:
|
||
print(f"✅ API端点 {endpoint} 可访问")
|
||
break
|
||
except:
|
||
continue
|
||
else:
|
||
print("⚠️ 未找到标准API健康检查端点")
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ API端点测试跳过: {e}")
|
||
|
||
# 显示测试结果
|
||
print("\n" + "=" * 60)
|
||
print("测试结果汇总:")
|
||
print("=" * 60)
|
||
|
||
passed = 0
|
||
total = len(results)
|
||
|
||
for service_name, result in results:
|
||
status = "✅ 通过" if result else "❌ 失败"
|
||
print(f"{service_name:15} : {status}")
|
||
if result:
|
||
passed += 1
|
||
|
||
print(f"\n总计: {passed}/{total} 个服务通过测试")
|
||
|
||
if passed == total:
|
||
print("\n🎉 所有服务测试通过！")
|
||
print("\n📋 服务访问地址:")
|
||
print("- API服务器: http://localhost:5050")
|
||
print("- 历史查看器: http://localhost:5051")
|
||
|
||
print("\n💡 使用建议:")
|
||
print("1. 使用API服务器进行API测试")
|
||
print("2. 使用历史查看器查看测试历史和报告")
|
||
print("3. 两个服务可以独立使用")
|
||
|
||
return True
|
||
else:
|
||
print(f"\n❌ {total - passed} 个服务测试失败")
|
||
print("\n🔧 故障排除建议:")
|
||
print("1. 检查Docker容器是否正常运行: docker ps")
|
||
print("2. 查看容器日志: docker logs dms-compliance-tool")
|
||
print("3. 检查端口是否被占用: lsof -i :5050 -i :5051")
|
||
print("4. 重新启动容器: docker restart dms-compliance-tool")
|
||
|
||
return False
|
||
|
||
def test_docker_services():
|
||
"""测试Docker中的服务"""
|
||
print("🐳 检查Docker容器状态...")
|
||
|
||
import subprocess
|
||
|
||
try:
|
||
# 检查容器是否运行
|
||
result = subprocess.run(
|
||
["docker", "ps", "--filter", "name=dms-compliance-tool", "--format", "{{.Status}}"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=10
|
||
)
|
||
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
status = result.stdout.strip()
|
||
if "Up" in status:
|
||
print(f"✅ Docker容器运行正常: {status}")
|
||
return True
|
||
else:
|
||
print(f"❌ Docker容器状态异常: {status}")
|
||
return False
|
||
else:
|
||
print("❌ 未找到运行中的dms-compliance-tool容器")
|
||
return False
|
||
|
||
except subprocess.TimeoutExpired:
|
||
print("❌ Docker命令执行超时")
|
||
return False
|
||
except FileNotFoundError:
|
||
print("⚠️ Docker命令未找到，跳过容器状态检查")
|
||
return True
|
||
except Exception as e:
|
||
print(f"⚠️ Docker状态检查出错: {e}")
|
||
return True
|
||
|
||
if __name__ == "__main__":
|
||
# 检查Docker容器状态
|
||
docker_ok = test_docker_services()
|
||
|
||
if docker_ok:
|
||
# 运行服务测试
|
||
if main():
|
||
sys.exit(0)
|
||
else:
|
||
sys.exit(1)
|
||
else:
|
||
print("\n❌ Docker容器状态检查失败，请先启动容器")
|
||
print("使用以下命令启动:")
|
||
print("./docker-build.sh")
|
||
sys.exit(1)
|