#!/usr/bin/env python3 """ InsightFlow Phase 8 Task 1 - 多租户 SaaS 架构测试脚本 测试内容： 1. 租户创建和管理 2. 自定义域名绑定和验证 3. 品牌白标配置 4. 成员邀请和权限管理 5. 资源使用统计 """ import os import sys from tenant_manager import get_tenant_manager sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) def test_tenant_management() -> None: """测试租户管理功能""" print(" = " * 60) print("测试 1: 租户管理") print(" = " * 60) manager = get_tenant_manager() # 1. 创建租户 print("\n1.1 创建租户...") tenant = manager.create_tenant( name = "Test Company", owner_id = "user_001", tier = "pro", description = "A test company tenant" ) 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}") # 2. 获取租户 print("\n1.2 获取租户信息...") fetched = manager.get_tenant(tenant.id) assert fetched is not None, "获取租户失败" print(f"✅ 获取租户成功: {fetched.name}") # 3. 通过 slug 获取 print("\n1.3 通过 slug 获取租户...") by_slug = manager.get_tenant_by_slug(tenant.slug) assert by_slug is not None, "通过 slug 获取失败" print(f"✅ 通过 slug 获取成功: {by_slug.name}") # 4. 更新租户 print("\n1.4 更新租户信息...") updated = manager.update_tenant( tenant_id = tenant.id, name = "Test Company Updated", tier = "enterprise" ) assert updated is not None, "更新租户失败" print(f"✅ 租户更新成功: {updated.name}, 层级: {updated.tier}") # 5. 列出租户 print("\n1.5 列出租户...") tenants = manager.list_tenants(limit = 10) print(f"✅ 找到 {len(tenants)} 个租户") return tenant.id def test_domain_management(tenant_id: str) -> None: """测试域名管理功能""" print("\n" + " = " * 60) print("测试 2: 域名管理") print(" = " * 60) manager = get_tenant_manager() # 1. 添加域名 print("\n2.1 添加自定义域名...") domain = manager.add_domain(tenant_id = tenant_id, domain = "test.example.com", is_primary = True) print(f"✅ 域名添加成功: {domain.domain}") print(f" - ID: {domain.id}") print(f" - 状态: {domain.status}") print(f" - 验证令牌: {domain.verification_token}") # 2. 获取验证指导 print("\n2.2 获取域名验证指导...") instructions = manager.get_domain_verification_instructions(domain.id) print("✅ 验证指导:") print(f" - DNS 记录: {instructions['dns_record']}") print(f" - 文件验证: {instructions['file_verification']}") # 3. 验证域名 print("\n2.3 验证域名...") verified = manager.verify_domain(tenant_id, domain.id) print(f"✅ 域名验证结果: {verified}") # 4. 通过域名获取租户 print("\n2.4 通过域名获取租户...") by_domain = manager.get_tenant_by_domain("test.example.com") if by_domain: print(f"✅ 通过域名获取租户成功: {by_domain.name}") else: print("⚠️ 通过域名获取租户失败（验证可能未通过）") # 5. 列出域名 print("\n2.5 列出所有域名...") domains = manager.list_domains(tenant_id) print(f"✅ 找到 {len(domains)} 个域名") for d in domains: print(f" - {d.domain} ({d.status})") return domain.id def test_branding_management(tenant_id: str) -> None: """测试品牌白标功能""" print("\n" + " = " * 60) print("测试 3: 品牌白标") print(" = " * 60) manager = get_tenant_manager() # 1. 更新品牌配置 print("\n3.1 更新品牌配置...") branding = manager.update_branding( tenant_id = tenant_id, logo_url = "https://example.com/logo.png", favicon_url = "https://example.com/favicon.ico", primary_color = "#1890ff", secondary_color = "#52c41a", custom_css = ".header { background: #1890ff; }", custom_js = "console.log('Custom JS loaded');", login_page_bg = "https://example.com/bg.jpg", ) print("✅ 品牌配置更新成功") print(f" - Logo: {branding.logo_url}") print(f" - 主色: {branding.primary_color}") print(f" - 次色: {branding.secondary_color}") # 2. 获取品牌配置 print("\n3.2 获取品牌配置...") fetched = manager.get_branding(tenant_id) assert fetched is not None, "获取品牌配置失败" print("✅ 获取品牌配置成功") # 3. 生成品牌 CSS print("\n3.3 生成品牌 CSS...") css = manager.get_branding_css(tenant_id) print(f"✅ 生成 CSS 成功 ({len(css)} 字符)") print(f" CSS 预览:\n{css[:200]}...") return branding.id def test_member_management(tenant_id: str) -> None: """测试成员管理功能""" print("\n" + " = " * 60) print("测试 4: 成员管理") print(" = " * 60) manager = get_tenant_manager() # 1. 邀请成员 print("\n4.1 邀请成员...") member1 = manager.invite_member( tenant_id = tenant_id, email = "admin@test.com", role = "admin", invited_by = "user_001" ) print(f"✅ 成员邀请成功: {member1.email}") print(f" - ID: {member1.id}") print(f" - 角色: {member1.role}") print(f" - 权限: {member1.permissions}") member2 = manager.invite_member( tenant_id = tenant_id, email = "member@test.com", role = "member", invited_by = "user_001" ) print(f"✅ 成员邀请成功: {member2.email}") # 2. 接受邀请 print("\n4.2 接受邀请...") accepted = manager.accept_invitation(member1.id, "user_002") print(f"✅ 邀请接受结果: {accepted}") # 3. 列出成员 print("\n4.3 列出所有成员...") members = manager.list_members(tenant_id) print(f"✅ 找到 {len(members)} 个成员") for m in members: print(f" - {m.email} ({m.role}) - {m.status}") # 4. 检查权限 print("\n4.4 检查权限...") can_manage = manager.check_permission(tenant_id, "user_002", "project", "create") print(f"✅ user_002 可以创建项目: {can_manage}") # 5. 更新成员角色 print("\n4.5 更新成员角色...") updated = manager.update_member_role(tenant_id, member2.id, "viewer") print(f"✅ 角色更新结果: {updated}") # 6. 获取用户所属租户 print("\n4.6 获取用户所属租户...") user_tenants = manager.get_user_tenants("user_002") print(f"✅ user_002 属于 {len(user_tenants)} 个租户") for t in user_tenants: print(f" - {t['name']} ({t['member_role']})") return member1.id, member2.id def test_usage_tracking(tenant_id: str) -> None: """测试资源使用统计功能""" print("\n" + " = " * 60) print("测试 5: 资源使用统计") print(" = " * 60) manager = get_tenant_manager() # 1. 记录使用 print("\n5.1 记录资源使用...") 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("✅ 资源使用记录成功") # 2. 获取使用统计 print("\n5.2 获取使用统计...") stats = manager.get_usage_stats(tenant_id) print("✅ 使用统计:") 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['usage_percentages']}") # 3. 检查资源限制 print("\n5.3 检查资源限制...") for resource in ["storage", "transcription", "api_calls", "projects", "entities", "members"]: allowed, current, limit = manager.check_resource_limit(tenant_id, resource) print(f" - {resource}: {current}/{limit} ({'✅' if allowed else '❌'})") return stats def cleanup(tenant_id: str, domain_id: str, member_ids: list) -> None: """清理测试数据""" print("\n" + " = " * 60) print("清理测试数据") print(" = " * 60) manager = get_tenant_manager() # 移除成员 for member_id in member_ids: if member_id: manager.remove_member(tenant_id, member_id) print(f"✅ 成员已移除: {member_id}") # 移除域名 if domain_id: manager.remove_domain(tenant_id, domain_id) print(f"✅ 域名已移除: {domain_id}") # 删除租户 manager.delete_tenant(tenant_id) print(f"✅ 租户已删除: {tenant_id}") def main() -> None: """主测试函数""" print("\n" + " = " * 60) print("InsightFlow Phase 8 Task 1 - 多租户 SaaS 架构测试") print(" = " * 60) tenant_id = None domain_id = None member_ids = [] try: # 运行所有测试 tenant_id = test_tenant_management() domain_id = test_domain_management(tenant_id) test_branding_management(tenant_id) m1, m2 = test_member_management(tenant_id) member_ids = [m1, m2] test_usage_tracking(tenant_id) print("\n" + " = " * 60) print("✅ 所有测试通过!") print(" = " * 60) except Exception as e: print(f"\n❌ 测试失败: {e}") import traceback traceback.print_exc() finally: # 清理 if tenant_id: try: cleanup(tenant_id, domain_id, member_ids) except Exception as e: print(f"⚠️ 清理失败: {e}") if __name__ == "__main__": main()