e3d7794ae7
- 创建 tenant_manager.py 多租户管理模块
- 租户管理（CRUD、slug、状态管理）
- 自定义域名绑定（DNS/文件验证）
- 品牌白标（Logo、主题色、自定义 CSS/JS）
- 成员管理（邀请、角色、权限）
- 资源使用统计和限制检查
- 租户上下文管理器
- 更新 schema.sql 添加租户相关表
- tenants, tenant_domains, tenant_branding
- tenant_members, tenant_permissions, tenant_usage
- 更新 main.py 添加租户 API 端点
- /api/v1/tenants/* 租户管理
- /api/v1/tenants/{id}/domains 域名管理
- /api/v1/tenants/{id}/branding 品牌配置
- /api/v1/tenants/{id}/members 成员管理
- /api/v1/tenants/{id}/usage 使用统计
- /api/v1/resolve-tenant 域名解析
- 创建 test_phase8_task1.py 测试脚本
508 lines
16 KiB
Python
508 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
InsightFlow Phase 8 - Multi-Tenant SaaS Test Script
|
||
多租户 SaaS 架构测试脚本
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from tenant_manager import (
|
||
get_tenant_manager, TenantManager, Tenant, TenantDomain, TenantBranding,
|
||
TenantMember, TenantStatus, TenantTier, TenantRole, DomainStatus,
|
||
TenantContext
|
||
)
|
||
|
||
|
||
def test_tenant_management():
|
||
"""测试租户管理功能"""
|
||
print("=" * 60)
|
||
print("测试租户管理功能")
|
||
print("=" * 60)
|
||
|
||
# 使用测试数据库
|
||
test_db = "test_tenant.db"
|
||
if os.path.exists(test_db):
|
||
os.remove(test_db)
|
||
|
||
manager = get_tenant_manager(test_db)
|
||
|
||
# 1. 创建租户
|
||
print("\n1. 创建租户...")
|
||
try:
|
||
tenant = manager.create_tenant(
|
||
name="Test Company",
|
||
owner_id="user_001",
|
||
tier="pro",
|
||
description="A test tenant for validation",
|
||
settings={"theme": "dark"}
|
||
)
|
||
print(f" ✓ 租户创建成功: {tenant.id}")
|
||
print(f" - 名称: {tenant.name}")
|
||
print(f" - Slug: {tenant.slug}")
|
||
print(f" - 层级: {tenant.tier}")
|
||
print(f" - 状态: {tenant.status}")
|
||
print(f" - 资源限制: {tenant.resource_limits}")
|
||
except Exception as e:
|
||
print(f" ✗ 租户创建失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 2. 获取租户
|
||
print("\n2. 获取租户...")
|
||
try:
|
||
fetched = manager.get_tenant(tenant.id)
|
||
assert fetched is not None
|
||
assert fetched.name == tenant.name
|
||
print(f" ✓ 通过 ID 获取租户成功")
|
||
|
||
fetched_by_slug = manager.get_tenant_by_slug(tenant.slug)
|
||
assert fetched_by_slug is not None
|
||
assert fetched_by_slug.id == tenant.id
|
||
print(f" ✓ 通过 Slug 获取租户成功")
|
||
except Exception as e:
|
||
print(f" ✗ 获取租户失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 3. 更新租户
|
||
print("\n3. 更新租户...")
|
||
try:
|
||
updated = manager.update_tenant(
|
||
tenant.id,
|
||
name="Test Company Updated",
|
||
tier="enterprise"
|
||
)
|
||
assert updated is not None
|
||
assert updated.name == "Test Company Updated"
|
||
assert updated.tier == "enterprise"
|
||
print(f" ✓ 租户更新成功")
|
||
print(f" - 新名称: {updated.name}")
|
||
print(f" - 新层级: {updated.tier}")
|
||
except Exception as e:
|
||
print(f" ✗ 租户更新失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 4. 列出租户
|
||
print("\n4. 列出租户...")
|
||
try:
|
||
tenants = manager.list_tenants()
|
||
assert len(tenants) >= 1
|
||
print(f" ✓ 列出租户成功，共 {len(tenants)} 个租户")
|
||
except Exception as e:
|
||
print(f" ✗ 列出租户失败: {e}")
|
||
return False
|
||
|
||
return tenant.id
|
||
|
||
|
||
def test_domain_management(tenant_id: str):
|
||
"""测试域名管理功能"""
|
||
print("\n" + "=" * 60)
|
||
print("测试域名管理功能")
|
||
print("=" * 60)
|
||
|
||
manager = get_tenant_manager("test_tenant.db")
|
||
|
||
# 1. 添加域名
|
||
print("\n1. 添加自定义域名...")
|
||
try:
|
||
domain = manager.add_domain(tenant_id, "app.example.com", is_primary=True)
|
||
print(f" ✓ 域名添加成功: {domain.id}")
|
||
print(f" - 域名: {domain.domain}")
|
||
print(f" - 状态: {domain.status}")
|
||
print(f" - 验证令牌: {domain.verification_token}")
|
||
print(f" - 是否主域名: {domain.is_primary}")
|
||
except Exception as e:
|
||
print(f" ✗ 域名添加失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 2. 获取域名验证指导
|
||
print("\n2. 获取域名验证指导...")
|
||
try:
|
||
instructions = manager.get_domain_verification_instructions(domain.id)
|
||
assert instructions is not None
|
||
print(f" ✓ 获取验证指导成功")
|
||
print(f" - DNS 记录: {instructions['dns_record']}")
|
||
except Exception as e:
|
||
print(f" ✗ 获取验证指导失败: {e}")
|
||
return False
|
||
|
||
# 3. 验证域名
|
||
print("\n3. 验证域名...")
|
||
try:
|
||
success = manager.verify_domain(tenant_id, domain.id)
|
||
if success:
|
||
print(f" ✓ 域名验证成功")
|
||
else:
|
||
print(f" ! 域名验证返回 False（可能是模拟验证）")
|
||
except Exception as e:
|
||
print(f" ✗ 域名验证失败: {e}")
|
||
return False
|
||
|
||
# 4. 获取域名列表
|
||
print("\n4. 获取域名列表...")
|
||
try:
|
||
domains = manager.list_domains(tenant_id)
|
||
assert len(domains) >= 1
|
||
print(f" ✓ 获取域名列表成功，共 {len(domains)} 个域名")
|
||
for d in domains:
|
||
print(f" - {d.domain} ({d.status})")
|
||
except Exception as e:
|
||
print(f" ✗ 获取域名列表失败: {e}")
|
||
return False
|
||
|
||
# 5. 通过域名获取租户
|
||
print("\n5. 通过域名解析租户...")
|
||
try:
|
||
resolved = manager.get_tenant_by_domain("app.example.com")
|
||
if resolved:
|
||
assert resolved.id == tenant_id
|
||
print(f" ✓ 域名解析租户成功")
|
||
else:
|
||
print(f" ! 域名解析租户返回 None（可能域名未激活）")
|
||
except Exception as e:
|
||
print(f" ✗ 域名解析失败: {e}")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def test_branding_management(tenant_id: str):
|
||
"""测试品牌配置管理功能"""
|
||
print("\n" + "=" * 60)
|
||
print("测试品牌配置管理功能")
|
||
print("=" * 60)
|
||
|
||
manager = get_tenant_manager("test_tenant.db")
|
||
|
||
# 1. 更新品牌配置
|
||
print("\n1. 更新品牌配置...")
|
||
try:
|
||
branding = manager.update_branding(
|
||
tenant_id,
|
||
logo_url="https://example.com/logo.png",
|
||
favicon_url="https://example.com/favicon.ico",
|
||
primary_color="#FF5733",
|
||
secondary_color="#33FF57",
|
||
custom_css="body { font-size: 14px; }",
|
||
custom_js="console.log('Custom JS loaded');"
|
||
)
|
||
assert branding is not None
|
||
print(f" ✓ 品牌配置更新成功")
|
||
print(f" - Logo: {branding.logo_url}")
|
||
print(f" - 主色调: {branding.primary_color}")
|
||
print(f" - 次色调: {branding.secondary_color}")
|
||
except Exception as e:
|
||
print(f" ✗ 品牌配置更新失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 2. 获取品牌配置
|
||
print("\n2. 获取品牌配置...")
|
||
try:
|
||
fetched = manager.get_branding(tenant_id)
|
||
assert fetched is not None
|
||
assert fetched.primary_color == "#FF5733"
|
||
print(f" ✓ 获取品牌配置成功")
|
||
except Exception as e:
|
||
print(f" ✗ 获取品牌配置失败: {e}")
|
||
return False
|
||
|
||
# 3. 生成品牌 CSS
|
||
print("\n3. 生成品牌 CSS...")
|
||
try:
|
||
css = manager.get_branding_css(tenant_id)
|
||
assert "--tenant-primary" in css
|
||
assert "#FF5733" in css
|
||
print(f" ✓ 品牌 CSS 生成成功")
|
||
print(f" - CSS 长度: {len(css)} 字符")
|
||
except Exception as e:
|
||
print(f" ✗ 品牌 CSS 生成失败: {e}")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def test_member_management(tenant_id: str):
|
||
"""测试成员管理功能"""
|
||
print("\n" + "=" * 60)
|
||
print("测试成员管理功能")
|
||
print("=" * 60)
|
||
|
||
manager = get_tenant_manager("test_tenant.db")
|
||
|
||
# 1. 邀请成员
|
||
print("\n1. 邀请成员...")
|
||
try:
|
||
member = manager.invite_member(
|
||
tenant_id=tenant_id,
|
||
email="user@example.com",
|
||
role="admin",
|
||
invited_by="user_001"
|
||
)
|
||
print(f" ✓ 成员邀请成功: {member.id}")
|
||
print(f" - 邮箱: {member.email}")
|
||
print(f" - 角色: {member.role}")
|
||
print(f" - 状态: {member.status}")
|
||
print(f" - 权限: {member.permissions}")
|
||
except Exception as e:
|
||
print(f" ✗ 成员邀请失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 2. 获取成员列表
|
||
print("\n2. 获取成员列表...")
|
||
try:
|
||
members = manager.list_members(tenant_id)
|
||
assert len(members) >= 2 # owner + invited member
|
||
print(f" ✓ 获取成员列表成功，共 {len(members)} 个成员")
|
||
for m in members:
|
||
print(f" - {m.email} ({m.role}, {m.status})")
|
||
except Exception as e:
|
||
print(f" ✗ 获取成员列表失败: {e}")
|
||
return False
|
||
|
||
# 3. 接受邀请
|
||
print("\n3. 接受邀请...")
|
||
try:
|
||
# 注意：accept_invitation 使用的是 member id 而不是 token
|
||
# 修正：查看源码后发现它接受的是 invitation_id（即 member id）
|
||
accepted = manager.accept_invitation(member.id, "user_002")
|
||
if accepted:
|
||
print(f" ✓ 邀请接受成功")
|
||
else:
|
||
print(f" ! 邀请接受返回 False（可能是状态不对）")
|
||
except Exception as e:
|
||
print(f" ✗ 邀请接受失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 4. 更新成员角色
|
||
print("\n4. 更新成员角色...")
|
||
try:
|
||
success = manager.update_member_role(
|
||
tenant_id=tenant_id,
|
||
member_id=member.id,
|
||
role="member"
|
||
)
|
||
if success:
|
||
print(f" ✓ 成员角色更新成功")
|
||
else:
|
||
print(f" ! 成员角色更新返回 False")
|
||
except Exception as e:
|
||
print(f" ✗ 成员角色更新失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 5. 检查权限
|
||
print("\n5. 检查用户权限...")
|
||
try:
|
||
# 检查 owner 权限
|
||
has_permission = manager.check_permission(
|
||
tenant_id=tenant_id,
|
||
user_id="user_001",
|
||
resource="project",
|
||
action="create"
|
||
)
|
||
print(f" ✓ 权限检查成功")
|
||
print(f" - Owner 是否有 project:create 权限: {has_permission}")
|
||
except Exception as e:
|
||
print(f" ✗ 权限检查失败: {e}")
|
||
return False
|
||
|
||
# 6. 获取用户租户列表
|
||
print("\n6. 获取用户租户列表...")
|
||
try:
|
||
user_tenants = manager.get_user_tenants("user_001")
|
||
assert len(user_tenants) >= 1
|
||
print(f" ✓ 获取用户租户列表成功，共 {len(user_tenants)} 个租户")
|
||
except Exception as e:
|
||
print(f" ✗ 获取用户租户列表失败: {e}")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def test_usage_stats(tenant_id: str):
|
||
"""测试使用统计功能"""
|
||
print("\n" + "=" * 60)
|
||
print("测试使用统计功能")
|
||
print("=" * 60)
|
||
|
||
manager = get_tenant_manager("test_tenant.db")
|
||
|
||
# 1. 记录使用
|
||
print("\n1. 记录资源使用...")
|
||
try:
|
||
manager.record_usage(
|
||
tenant_id=tenant_id,
|
||
storage_bytes=1024 * 1024 * 50, # 50MB
|
||
transcription_seconds=600, # 10分钟
|
||
api_calls=100,
|
||
projects_count=5,
|
||
entities_count=50,
|
||
members_count=3
|
||
)
|
||
print(f" ✓ 资源使用记录成功")
|
||
except Exception as e:
|
||
print(f" ✗ 资源使用记录失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 2. 获取使用统计
|
||
print("\n2. 获取使用统计...")
|
||
try:
|
||
stats = manager.get_usage_stats(tenant_id)
|
||
print(f" ✓ 使用统计获取成功")
|
||
print(f" - 存储: {stats['storage_mb']:.2f} MB")
|
||
print(f" - 转录: {stats['transcription_minutes']:.2f} 分钟")
|
||
print(f" - API 调用: {stats['api_calls']}")
|
||
print(f" - 项目数: {stats['projects_count']}")
|
||
print(f" - 实体数: {stats['entities_count']}")
|
||
print(f" - 成员数: {stats['members_count']}")
|
||
print(f" - 配额: {stats['limits']}")
|
||
except Exception as e:
|
||
print(f" ✗ 使用统计获取失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
# 3. 检查资源限制
|
||
print("\n3. 检查资源限制...")
|
||
try:
|
||
allowed, current, limit = manager.check_resource_limit(tenant_id, "storage")
|
||
print(f" ✓ 资源限制检查成功")
|
||
print(f" - 存储: {allowed}, 当前: {current}, 限制: {limit}")
|
||
except Exception as e:
|
||
print(f" ✗ 资源限制检查失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def test_tenant_context():
|
||
"""测试租户上下文管理"""
|
||
print("\n" + "=" * 60)
|
||
print("测试租户上下文管理")
|
||
print("=" * 60)
|
||
|
||
# 1. 设置和获取租户上下文
|
||
print("\n1. 设置和获取租户上下文...")
|
||
try:
|
||
TenantContext.set_current_tenant("tenant_123")
|
||
tenant_id = TenantContext.get_current_tenant()
|
||
assert tenant_id == "tenant_123"
|
||
print(f" ✓ 租户上下文设置成功: {tenant_id}")
|
||
except Exception as e:
|
||
print(f" ✗ 租户上下文设置失败: {e}")
|
||
return False
|
||
|
||
# 2. 设置和获取用户上下文
|
||
print("\n2. 设置和获取用户上下文...")
|
||
try:
|
||
TenantContext.set_current_user("user_456")
|
||
user_id = TenantContext.get_current_user()
|
||
assert user_id == "user_456"
|
||
print(f" ✓ 用户上下文设置成功: {user_id}")
|
||
except Exception as e:
|
||
print(f" ✗ 用户上下文设置失败: {e}")
|
||
return False
|
||
|
||
# 3. 清除上下文
|
||
print("\n3. 清除上下文...")
|
||
try:
|
||
TenantContext.clear()
|
||
assert TenantContext.get_current_tenant() is None
|
||
assert TenantContext.get_current_user() is None
|
||
print(f" ✓ 上下文清除成功")
|
||
except Exception as e:
|
||
print(f" ✗ 上下文清除失败: {e}")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def cleanup():
|
||
"""清理测试数据"""
|
||
print("\n" + "=" * 60)
|
||
print("清理测试数据")
|
||
print("=" * 60)
|
||
|
||
test_db = "test_tenant.db"
|
||
if os.path.exists(test_db):
|
||
os.remove(test_db)
|
||
print(f"✓ 删除测试数据库: {test_db}")
|
||
|
||
|
||
def main():
|
||
"""主测试函数"""
|
||
print("\n" + "=" * 60)
|
||
print("InsightFlow Phase 8 - Multi-Tenant SaaS 测试")
|
||
print("=" * 60)
|
||
|
||
all_passed = True
|
||
tenant_id = None
|
||
|
||
try:
|
||
# 测试租户上下文
|
||
if not test_tenant_context():
|
||
all_passed = False
|
||
|
||
# 测试租户管理
|
||
tenant_id = test_tenant_management()
|
||
if not tenant_id:
|
||
all_passed = False
|
||
|
||
# 测试域名管理
|
||
if not test_domain_management(tenant_id):
|
||
all_passed = False
|
||
|
||
# 测试品牌配置
|
||
if not test_branding_management(tenant_id):
|
||
all_passed = False
|
||
|
||
# 测试成员管理
|
||
if not test_member_management(tenant_id):
|
||
all_passed = False
|
||
|
||
# 测试使用统计
|
||
if not test_usage_stats(tenant_id):
|
||
all_passed = False
|
||
|
||
except Exception as e:
|
||
print(f"\n测试过程中发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
all_passed = False
|
||
|
||
finally:
|
||
cleanup()
|
||
|
||
print("\n" + "=" * 60)
|
||
if all_passed:
|
||
print("✓ 所有测试通过！")
|
||
else:
|
||
print("✗ 部分测试失败")
|
||
print("=" * 60)
|
||
|
||
return 0 if all_passed else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|