153 lines
4.7 KiB
Python
Executable File
153 lines
4.7 KiB
Python
Executable File
#!/usr/bin/env python
|
|
"""
|
|
测试新增 API 集成
|
|
|
|
验证新的认证、用户管理和审计日志接口是否正确集成
|
|
"""
|
|
import sys
|
|
import subprocess
|
|
import time
|
|
|
|
def check_imports():
|
|
"""检查关键模块是否可以导入"""
|
|
print("=" * 60)
|
|
print("步骤 1: 检查模块导入")
|
|
print("=" * 60)
|
|
|
|
modules = [
|
|
("app.core.encryption", "加密模块"),
|
|
("app.core.security", "安全模块"),
|
|
("app.core.audit", "审计模块"),
|
|
("app.domain.models.role", "角色模型"),
|
|
("app.domain.schemas.user", "用户Schema"),
|
|
("app.domain.schemas.audit", "审计Schema"),
|
|
("app.auth.permissions", "权限控制"),
|
|
("app.api.v1.endpoints.auth", "认证接口"),
|
|
("app.api.v1.endpoints.user_management", "用户管理接口"),
|
|
("app.api.v1.endpoints.audit", "审计日志接口"),
|
|
("app.infra.repositories.user_repository", "用户仓储"),
|
|
("app.infra.repositories.audit_repository", "审计仓储"),
|
|
("app.infra.audit.middleware", "审计中间件"),
|
|
]
|
|
|
|
success = 0
|
|
failed = 0
|
|
|
|
for module_name, desc in modules:
|
|
try:
|
|
__import__(module_name)
|
|
print(f"✓ {desc:20s} ({module_name})")
|
|
success += 1
|
|
except Exception as e:
|
|
print(f"✗ {desc:20s} ({module_name})")
|
|
print(f" 错误: {e}")
|
|
failed += 1
|
|
|
|
print(f"\n结果: {success} 成功, {failed} 失败")
|
|
print()
|
|
return failed == 0
|
|
|
|
def check_router():
|
|
"""检查路由配置"""
|
|
print("=" * 60)
|
|
print("步骤 2: 检查路由配置")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
from app.api.v1 import router
|
|
from app.api.v1.endpoints import auth, user_management, audit
|
|
|
|
print("✓ router 模块已导入")
|
|
print("✓ auth 端点已导入")
|
|
print("✓ user_management 端点已导入")
|
|
print("✓ audit 端点已导入")
|
|
|
|
# 检查 router 中是否包含新增的路由
|
|
api_router = router.api_router
|
|
print(f"\n已注册的路由数量: {len(api_router.routes)}")
|
|
|
|
# 查找新增的路由
|
|
auth_routes = [r for r in api_router.routes if hasattr(r, 'path') and '/auth' in r.path]
|
|
user_routes = [r for r in api_router.routes if hasattr(r, 'path') and '/users' in r.path]
|
|
audit_routes = [r for r in api_router.routes if hasattr(r, 'path') and '/audit' in r.path]
|
|
|
|
print(f"认证相关路由: {len(auth_routes)} 个")
|
|
print(f"用户管理路由: {len(user_routes)} 个")
|
|
print(f"审计日志路由: {len(audit_routes)} 个")
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"✗ 路由配置检查失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
def check_main_app():
|
|
"""检查 main.py 配置"""
|
|
print("\n" + "=" * 60)
|
|
print("步骤 3: 检查 main.py 配置")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
from app.main import app
|
|
|
|
print("✓ FastAPI app 已创建")
|
|
print(f" 标题: {app.title}")
|
|
print(f" 版本: {app.version}")
|
|
|
|
# 检查中间件
|
|
middleware_names = [m.__class__.__name__ for m in app.user_middleware]
|
|
print(f"\n已注册的中间件: {len(middleware_names)} 个")
|
|
for name in middleware_names:
|
|
print(f" - {name}")
|
|
|
|
# 检查路由
|
|
print(f"\n已注册的路由: {len(app.routes)} 个")
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"✗ main.py 配置检查失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
def main():
|
|
print("\n🔍 TJWater Server API 集成测试\n")
|
|
|
|
results = []
|
|
|
|
# 测试 1: 模块导入
|
|
results.append(("模块导入", check_imports()))
|
|
|
|
# 测试 2: 路由配置
|
|
results.append(("路由配置", check_router()))
|
|
|
|
# 测试 3: main.py
|
|
results.append(("main.py配置", check_main_app()))
|
|
|
|
# 总结
|
|
print("\n" + "=" * 60)
|
|
print("测试总结")
|
|
print("=" * 60)
|
|
|
|
for name, success in results:
|
|
status = "✓ 通过" if success else "✗ 失败"
|
|
print(f"{status:8s} - {name}")
|
|
|
|
all_passed = all(success for _, success in results)
|
|
|
|
if all_passed:
|
|
print("\n✅ 所有测试通过！")
|
|
print("\n下一步:")
|
|
print(" 1. 确保数据库迁移已执行")
|
|
print(" 2. 配置 .env 文件")
|
|
print(" 3. 启动服务: uvicorn app.main:app --reload")
|
|
print(" 4. 访问文档: http://localhost:8000/docs")
|
|
return 0
|
|
else:
|
|
print("\n❌ 部分测试失败，请检查错误信息")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|