taiji-AI-PAD/services/mcp-server/tests/test_permissions.py

"""
权限系统测试用例
"""

import pytest
from httpx import AsyncClient


class TestPermissions:
    """权限系统测试类"""
    
    @pytest.mark.asyncio
    async def test_role_permissions_mapping(self):
        """测试角色权限映射"""
        from app.permissions import ROLE_PERMISSIONS, get_role_permissions
        
        # 测试超级管理员拥有所有权限
        super_admin_perms = get_role_permissions("super_admin")
        assert len(super_admin_perms) > 0
        assert "view:overview" in super_admin_perms
        assert "manage:billing" in super_admin_perms
        
        # 测试计费管理员权限
        billing_admin_perms = get_role_permissions("billing_admin")
        assert "view:billing" in billing_admin_perms
        assert "manage:billing" in billing_admin_perms
        assert "manage:tenants" not in billing_admin_perms
        
        # 测试运营管理员权限
        ops_admin_perms = get_role_permissions("operations_admin")
        assert "manage:tenants" in ops_admin_perms
        assert "manage:resources" in ops_admin_perms
        assert "manage:billing" not in ops_admin_perms
        
        # 测试管理员权限
        admin_perms = get_role_permissions("admin")
        assert "manage:tenants" in admin_perms
        assert "manage:billing" in admin_perms
        assert "manage:settings" in admin_perms
    
    @pytest.mark.asyncio
    async def test_has_permission(self):
        """测试权限检查函数"""
        from app.permissions import has_permission
        
        # 超级管理员应该有所有权限
        assert has_permission("super_admin", "view:overview")
        assert has_permission("super_admin", "manage:billing")
        assert has_permission("super_admin", "manage:tenants")
        
        # 计费管理员应该有计费权限，但没有租户管理权限
        assert has_permission("billing_admin", "manage:billing")
        assert not has_permission("billing_admin", "manage:tenants")
        
        # 运营管理员应该有租户管理权限，但没有计费管理权限
        assert has_permission("operations_admin", "manage:tenants")
        assert not has_permission("operations_admin", "manage:billing")
        
        # 普通用户应该只有查看权限
        assert has_permission("user", "view:overview")
        assert not has_permission("user", "manage:billing")


class TestAuthenticationAPI:
    """认证API测试类"""
    
    @pytest.mark.asyncio
    async def test_login_success(self, client: AsyncClient, test_users):
        """测试成功登录"""
        response = await client.post(
            "/api/auth/login",
            json={
                "email": "superadmin@test.com",
                "password": "super123",
                "role": "super_admin"
            }
        )
        
        assert response.status_code == 200
        data = response.json()
        assert data["success"] is True
        assert "token" in data["data"]
        assert "refresh_token" in data["data"]
        assert data["data"]["user"]["role"] == "super_admin"
    
    @pytest.mark.asyncio
    async def test_login_wrong_password(self, client: AsyncClient, test_users):
        """测试错误密码登录"""
        response = await client.post(
            "/api/auth/login",
            json={
                "email": "superadmin@test.com",
                "password": "wrongpassword",
                "role": "super_admin"
            }
        )
        
        assert response.status_code == 401
    
    @pytest.mark.asyncio
    async def test_login_wrong_role(self, client: AsyncClient, test_users):
        """测试错误角色登录"""
        # 尝试用普通用户身份登录管理员账号
        response = await client.post(
            "/api/auth/login",
            json={
                "email": "superadmin@test.com",
                "password": "super123",
                "role": "user"  # 错误的角色
            }
        )
        
        assert response.status_code == 403
    
    @pytest.mark.asyncio
    async def test_login_all_roles(self, client: AsyncClient, test_users):
        """测试所有角色登录"""
        test_cases = [
            ("superadmin@test.com", "super123", "super_admin"),
            ("admin@test.com", "admin123", "admin"),
            ("billing@test.com", "billing123", "billing_admin"),
            ("operations@test.com", "ops123", "operations_admin"),
            ("channel-admin@test.com", "channel123", "channel"),
            ("provider@test.com", "provider123", "provider"),
            ("user@test.com", "user123", "user"),
        ]
        
        for email, password, role in test_cases:
            response = await client.post(
                "/api/auth/login",
                json={
                    "email": email,
                    "password": password,
                    "role": role
                }
            )
            
            assert response.status_code == 200, f"登录失败: {email} as {role}"
            data = response.json()
            assert data["success"] is True
            assert "token" in data["data"]


class TestRoleBasedAccess:
    """基于角色的访问控制测试"""
    
    @pytest.mark.asyncio
    async def test_super_admin_access(
        self, 
        client: AsyncClient, 
        auth_tokens, 
        auth_headers
    ):
        """测试超级管理员访问权限"""
        token = auth_tokens.get("super_admin")
        headers = auth_headers(token)
        
        # 超级管理员应该能访问所有端点
        endpoints = [
            "/api/user/dashboard/stats",
            "/api/channel/dashboard/stats",
            "/api/admin/dashboard/stats",
            "/api/admin/channels",
            "/api/provider/models",
        ]
        
        for endpoint in endpoints:
            response = await client.get(endpoint, headers=headers)
            # 不应该返回403（权限不足）
            assert response.status_code != 403, f"超级管理员无法访问: {endpoint}"
    
    @pytest.mark.asyncio
    async def test_billing_admin_access(
        self, 
        client: AsyncClient, 
        auth_tokens, 
        auth_headers
    ):
        """测试计费管理员访问权限"""
        token = auth_tokens.get("billing_admin")
        headers = auth_headers(token)
        
        # 计费管理员应该能访问计费相关端点
        response = await client.get("/api/admin/billing/records", headers=headers)
        assert response.status_code != 403
        
        # 但不应该能访问租户管理端点
        response = await client.get("/api/admin/tenants", headers=headers)
        assert response.status_code == 403
    
    @pytest.mark.asyncio
    async def test_operations_admin_access(
        self, 
        client: AsyncClient, 
        auth_tokens, 
        auth_headers
    ):
        """测试运营管理员访问权限"""
        token = auth_tokens.get("operations_admin")
        headers = auth_headers(token)
        
        # 运营管理员应该能访问租户和资源管理端点
        response = await client.get("/api/admin/tenants", headers=headers)
        assert response.status_code != 403
        
        # 但不应该能执行充值操作
        response = await client.post(
            "/api/admin/billing/recharge",
            json={"user_id": "some-id", "amount": 100},
            headers=headers
        )
        assert response.status_code == 403
    
    @pytest.mark.asyncio
    async def test_user_limited_access(
        self, 
        client: AsyncClient, 
        auth_tokens, 
        auth_headers
    ):
        """测试普通用户的受限访问"""
        token = auth_tokens.get("user")
        headers = auth_headers(token)
        
        # 普通用户应该能访问自己的仪表板
        response = await client.get("/api/user/dashboard/stats", headers=headers)
        assert response.status_code != 403
        
        # 但不应该能访问管理员端点
        admin_endpoints = [
            "/api/admin/dashboard/stats",
            "/api/admin/channels",
            "/api/admin/tenants",
            "/api/channel/dashboard/stats",
        ]
        
        for endpoint in admin_endpoints:
            response = await client.get(endpoint, headers=headers)
            assert response.status_code == 403, f"普通用户不应该能访问: {endpoint}"


class TestAPIKeyAuthentication:
    """API密钥认证测试"""
    
    @pytest.mark.asyncio
    async def test_api_key_authentication(
        self, 
        client: AsyncClient, 
        test_session,
        test_users
    ):
        """测试API密钥认证"""
        from models import APIKey
        from passlib.context import CryptContext
        
        pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
        
        # 为超级管理员创建API密钥
        user = test_users["super_admin"]
        api_key = "sk-test-1234567890abcdef"
        api_key_obj = APIKey(
            user_id=user.id,
            api_key_hash=pwd_context.hash(api_key),
            api_key_prefix="sk-test-12",
            name="测试密钥"
        )
        test_session.add(api_key_obj)
        await test_session.commit()
        
        # 使用API密钥访问端点
        response = await client.get(
            "/api/user/dashboard/stats",
            headers={"X-API-Key": api_key}
        )
        
        assert response.status_code != 401  # 不应该返回未认证
    
    @pytest.mark.asyncio
    async def test_invalid_api_key(self, client: AsyncClient):
        """测试无效的API密钥"""
        response = await client.get(
            "/api/user/dashboard/stats",
            headers={"X-API-Key": "invalid-key"}
        )
        
        assert response.status_code == 401


class TestTokenRefresh:
    """Token刷新测试"""
    
    @pytest.mark.asyncio
    async def test_refresh_token(
        self, 
        client: AsyncClient, 
        test_users
    ):
        """测试刷新token"""
        # 先登录获取token
        login_response = await client.post(
            "/api/auth/login",
            json={
                "email": "superadmin@test.com",
                "password": "super123",
                "role": "super_admin"
            }
        )
        
        assert login_response.status_code == 200
        login_data = login_response.json()
        refresh_token = login_data["data"]["refresh_token"]
        
        # 使用refresh token获取新的access token
        refresh_response = await client.post(
            "/api/auth/refresh",
            json={"refresh_token": refresh_token}
        )
        
        assert refresh_response.status_code == 200
        refresh_data = refresh_response.json()
        assert "token" in refresh_data["data"]
        assert refresh_data["data"]["token"] != login_data["data"]["token"]


class TestPasswordChange:
    """密码修改测试"""
    
    @pytest.mark.asyncio
    async def test_change_password(
        self, 
        client: AsyncClient, 
        auth_tokens,
        auth_headers
    ):
        """测试修改密码"""
        token = auth_tokens.get("user")
        headers = auth_headers(token)
        
        response = await client.post(
            "/api/auth/change-password",
            json={
                "old_password": "user123",
                "new_password": "newpassword123"
            },
            headers=headers
        )
        
        # 应该成功或返回具体错误（取决于实现）
        assert response.status_code in [200, 400, 401]
    
    @pytest.mark.asyncio
    async def test_change_password_wrong_old_password(
        self, 
        client: AsyncClient, 
        auth_tokens,
        auth_headers
    ):
        """测试使用错误的旧密码修改密码"""
        token = auth_tokens.get("user")
        headers = auth_headers(token)
        
        response = await client.post(
            "/api/auth/change-password",
            json={
                "old_password": "wrongpassword",
                "new_password": "newpassword123"
            },
            headers=headers
        )
        
        assert response.status_code == 400


class TestLogout:
    """登出测试"""
    
    @pytest.mark.asyncio
    async def test_logout(
        self, 
        client: AsyncClient, 
        auth_tokens,
        auth_headers
    ):
        """测试登出"""
        token = auth_tokens.get("user")
        headers = auth_headers(token)
        
        response = await client.post("/api/auth/logout", headers=headers)
        
        # 应该成功登出
        assert response.status_code in [200, 204]
        
        # 登出后使用相同token应该无法访问
        response = await client.get("/api/user/dashboard/stats", headers=headers)
        # 根据实现，可能返回401或仍然有效（如果没有实现token黑名单）
        # 这里只检查不会崩溃
        assert response.status_code in [200, 401]


class TestCrossRoleAccess:
    """跨角色访问测试"""
    
    @pytest.mark.asyncio
    async def test_channel_admin_cannot_access_other_channels(
        self, 
        client: AsyncClient, 
        auth_tokens,
        auth_headers,
        test_session
    ):
        """测试渠道管理员不能访问其他渠道的数据"""
        # 创建第二个渠道和渠道管理员
        from models import Channel, User
        from passlib.context import CryptContext
        
        pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
        
        channel2 = Channel(
            name="测试渠道2",
            email="channel2@test.com",
            password_hash=pwd_context.hash("channel123"),
            contact_person="联系人2",
            contact_phone="13800138001",
            channel_credit=5000.0,
            status="active"
        )
        test_session.add(channel2)
        await test_session.commit()
        await test_session.refresh(channel2)
        
        # 渠道1的管理员尝试访问渠道2的数据
        token = auth_tokens.get("channel_admin")
        headers = auth_headers(token)
        
        response = await client.get(
            f"/api/channel/tenants?channel_id={channel2.id}",
            headers=headers
        )
        
        # 应该被拒绝或返回空数据
        assert response.status_code in [403, 200]
        if response.status_code == 200:
            data = response.json()
            # 如果返回200，数据应该为空或只包含自己渠道的数据
            assert len(data.get("data", [])) == 0 or \
                   all(t.get("channel_id") != str(channel2.id) for t in data.get("data", []))


class TestPermissionInheritance:
    """权限继承测试"""
    
    @pytest.mark.asyncio
    async def test_admin_has_billing_permissions(self):
        """测试管理员拥有计费管理员的权限"""
        from app.permissions import has_permission
        
        # 管理员应该拥有计费管理员的所有权限
        billing_permissions = ["view:billing", "manage:billing"]
        
        for perm in billing_permissions:
            assert has_permission("admin", perm), \
                f"管理员应该拥有权限: {perm}"
    
    @pytest.mark.asyncio
    async def test_admin_has_operations_permissions(self):
        """测试管理员拥有运营管理员的权限"""
        from app.permissions import has_permission
        
        # 管理员应该拥有运营管理员的所有权限
        ops_permissions = ["manage:tenants", "manage:resources"]
        
        for perm in ops_permissions:
            assert has_permission("admin", perm), \
                f"管理员应该拥有权限: {perm}"
    
    @pytest.mark.asyncio
    async def test_super_admin_has_all_permissions(self):
        """测试超级管理员拥有所有权限"""
        from app.permissions import PERMISSIONS, has_permission
        
        # 超级管理员应该拥有所有权限
        for perm in PERMISSIONS.values():
            assert has_permission("super_admin", perm), \
                f"超级管理员应该拥有权限: {perm}"


if __name__ == "__main__":
    pytest.main([__file__, "-v"])