Simon
144 lines
3.7 KiB
Python
144 lines
3.7 KiB
Python
#!/usr/bin/env python
|
||
# =============================================================================
|
||
# 企微IT智能服务台 — 服务路由验证脚本
|
||
# =============================================================================
|
||
# 用法：
|
||
# python test_service_routes.py # 测试所有服务
|
||
# python test_service_routes.py core # 测试单个服务
|
||
# =============================================================================
|
||
|
||
import os
|
||
import sys
|
||
|
||
# 设置 Python 路径
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
|
||
def test_service(service_name: str) -> bool:
|
||
"""测试单个服务的路由加载
|
||
|
||
Args:
|
||
service_name: 服务名 (core/conversation/agent/ai/admin)
|
||
|
||
Returns:
|
||
True 如果测试通过
|
||
"""
|
||
# 设置环境变量
|
||
os.environ['SERVICE_NAME'] = service_name
|
||
|
||
# 重新导入模块
|
||
if 'app.api.service_routes' in sys.modules:
|
||
del sys.modules['app.api.service_routes']
|
||
|
||
from app.api.service_routes import (
|
||
get_routes_for_service,
|
||
is_service_mode,
|
||
get_current_service_name,
|
||
)
|
||
|
||
# 验证服务名
|
||
actual_name = get_current_service_name()
|
||
if actual_name != service_name:
|
||
print(f" ❌ 服务名不匹配: 期望 {service_name}, 实际 {actual_name}")
|
||
return False
|
||
|
||
# 验证服务模式
|
||
if not is_service_mode():
|
||
print(f" ❌ 未进入服务化模式")
|
||
return False
|
||
|
||
# 获取路由
|
||
routes = get_routes_for_service(service_name)
|
||
|
||
print(f" 🏷️ 服务: {service_name}")
|
||
print(f" 📋 路由数量: {len(routes)}")
|
||
|
||
for router, tags, prefix in routes:
|
||
print(f" ✅ {tags[0]}")
|
||
|
||
return True
|
||
|
||
|
||
def test_monolith_mode() -> bool:
|
||
"""测试单体模式（不设置 SERVICE_NAME）
|
||
|
||
Returns:
|
||
True 如果测试通过
|
||
"""
|
||
# 清除环境变量
|
||
if 'SERVICE_NAME' in os.environ:
|
||
del os.environ['SERVICE_NAME']
|
||
|
||
# 重新导入模块
|
||
if 'app.api.service_routes' in sys.modules:
|
||
del sys.modules['app.api.service_routes']
|
||
|
||
from app.api.service_routes import (
|
||
get_routes_for_service,
|
||
is_service_mode,
|
||
get_current_service_name,
|
||
)
|
||
|
||
# 验证服务名
|
||
actual_name = get_current_service_name()
|
||
if actual_name != "":
|
||
print(f" ❌ 服务名应该为空: 实际 {actual_name}")
|
||
return False
|
||
|
||
# 验证不是服务模式
|
||
if is_service_mode():
|
||
print(f" ❌ 不应该进入服务化模式")
|
||
return False
|
||
|
||
# 获取路由（应该获取全部路由）
|
||
routes = get_routes_for_service("")
|
||
|
||
print(f" 🏢 单体模式")
|
||
print(f" 📋 路由数量: {len(routes)}")
|
||
|
||
return True
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("=" * 60)
|
||
print("🧪 服务路由验证测试")
|
||
print("=" * 60)
|
||
|
||
# 测试单体模式
|
||
print("\n📦 测试单体模式...")
|
||
if not test_monolith_mode():
|
||
print("❌ 单体模式测试失败")
|
||
return 1
|
||
print("✅ 单体模式测试通过")
|
||
|
||
# 测试所有服务
|
||
services = ['core', 'conversation', 'agent', 'ai', 'admin']
|
||
|
||
# 如果提供了命令行参数，只测试指定服务
|
||
if len(sys.argv) > 1:
|
||
services = [sys.argv[1]]
|
||
|
||
print("\n📦 测试各服务...")
|
||
all_passed = True
|
||
|
||
for service in services:
|
||
print(f"\n{'─' * 40}")
|
||
if not test_service(service):
|
||
all_passed = False
|
||
print(f"❌ 服务 {service} 测试失败")
|
||
else:
|
||
print(f"✅ 服务 {service} 测试通过")
|
||
|
||
print("\n" + "=" * 60)
|
||
if all_passed:
|
||
print("✅ 全部测试通过！")
|
||
return 0
|
||
else:
|
||
print("❌ 部分测试失败")
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|