insightflow/backend/test_phase8_task1.py

#!/usr/bin/env python3
"""
InsightFlow Phase 8 Task 1 - 多租户 SaaS 架构测试脚本

测试内容：
1. 租户创建和管理
2. 自定义域名绑定和验证
3. 品牌白标配置
4. 成员邀请和权限管理
5. 资源使用统计
"""

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, TenantRole, TenantStatus, TenantTier
)


def test_tenant_management():
    """测试租户管理功能"""
    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):
    """测试域名管理功能"""
    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(f"✅ 验证指导:")
    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):
    """测试品牌白标功能"""
    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(f"✅ 品牌配置更新成功")
    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(f"✅ 获取品牌配置成功")
    
    # 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):
    """测试成员管理功能"""
    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):
    """测试资源使用统计功能"""
    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(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['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):
    """清理测试数据"""
    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():
    """主测试函数"""
    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()