cloud-reporter/backend/test_credential_update.py

#!/usr/bin/env python
"""
测试凭证更新功能

验证:
1. Account ID 字段是否可空
2. Access Key 类型凭证创建逻辑
3. 数据库模型更新
"""
import os
import sys

# 添加项目根目录到路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from app import create_app, db
from app.models import AWSCredential
from sqlalchemy import text

def test_database_schema():
    """测试数据库模式更新"""
    print("测试数据库模式...")
    
    app = create_app('testing')
    with app.app_context():
        # 创建测试表
        db.create_all()
        
        # 检查 account_id 字段是否可空
        try:
            # 尝试创建一个没有 account_id 的凭证
            test_credential = AWSCredential(
                name="Test Access Key",
                credential_type="access_key",
                account_id=None,  # 测试可空
                access_key_id="AKIATEST123456789",
                is_active=True
            )
            test_credential.set_secret_access_key("test-secret-key")
            
            db.session.add(test_credential)
            db.session.commit()
            
            print("✓ account_id 字段可空测试通过")
            
            # 清理测试数据
            db.session.delete(test_credential)
            db.session.commit()
            
        except Exception as e:
            print(f"❌ account_id 字段可空测试失败: {e}")
            return False
    
    return True

def test_model_methods():
    """测试模型方法"""
    print("测试模型方法...")
    
    app = create_app('testing')
    with app.app_context():
        db.create_all()
        
        try:
            # 测试 Assume Role 类型
            assume_role_cred = AWSCredential(
                name="Test Assume Role",
                credential_type="assume_role",
                account_id="123456789012",
                role_arn="arn:aws:iam::123456789012:role/TestRole",
                external_id="test-external-id",
                is_active=True
            )
            
            # 测试 Access Key 类型（无 account_id）
            access_key_cred = AWSCredential(
                name="Test Access Key",
                credential_type="access_key",
                account_id=None,
                access_key_id="AKIATEST123456789",
                is_active=True
            )
            access_key_cred.set_secret_access_key("test-secret-key")
            
            # 测试 to_dict 方法
            assume_role_dict = assume_role_cred.to_dict()
            access_key_dict = access_key_cred.to_dict()
            
            print("✓ Assume Role 凭证字典:", assume_role_dict)
            print("✓ Access Key 凭证字典:", access_key_dict)
            
            # 验证字段
            assert assume_role_dict['credential_type'] == 'assume_role'
            assert assume_role_dict['account_id'] == '123456789012'
            assert access_key_dict['credential_type'] == 'access_key'
            assert access_key_dict['account_id'] is None
            
            print("✓ 模型方法测试通过")
            
        except Exception as e:
            print(f"❌ 模型方法测试失败: {e}")
            return False
    
    return True

def test_api_validation():
    """测试API验证逻辑"""
    print("测试API验证逻辑...")
    
    from app.api.credentials import validate_account_id, validate_role_arn
    
    try:
        # 测试 account_id 验证
        assert validate_account_id("123456789012") == True
        assert validate_account_id("12345678901") == False  # 11位
        assert validate_account_id("1234567890123") == False  # 13位
        assert validate_account_id("12345678901a") == False  # 包含字母
        assert validate_account_id("") == False  # 空字符串
        assert validate_account_id(None) == False  # None
        
        # 测试 role_arn 验证
        assert validate_role_arn("arn:aws:iam::123456789012:role/TestRole") == True
        assert validate_role_arn("arn:aws:iam::123456789012:user/TestUser") == False  # 不是role
        assert validate_role_arn("invalid-arn") == False  # 无效格式
        assert validate_role_arn("") == False  # 空字符串
        assert validate_role_arn(None) == False  # None
        
        print("✓ API验证逻辑测试通过")
        
    except Exception as e:
        print(f"❌ API验证逻辑测试失败: {e}")
        return False
    
    return True

def main():
    """运行所有测试"""
    print("="*50)
    print("凭证更新功能测试")
    print("="*50)
    
    tests = [
        ("数据库模式", test_database_schema),
        ("模型方法", test_model_methods),
        ("API验证", test_api_validation),
    ]
    
    results = []
    for test_name, test_func in tests:
        print(f"\n{test_name}测试:")
        try:
            result = test_func()
            results.append((test_name, result))
        except Exception as e:
            print(f"❌ {test_name}测试异常: {e}")
            results.append((test_name, False))
    
    print("\n" + "="*50)
    print("测试结果:")
    print("="*50)
    
    all_passed = True
    for test_name, result in results:
        status = "✓ 通过" if result else "❌ 失败"
        print(f"  {test_name}: {status}")
        if not result:
            all_passed = False
    
    if all_passed:
        print("\n🎉 所有测试通过! 凭证更新功能正常。")
    else:
        print("\n⚠️  部分测试失败，请检查问题。")
    
    print("="*50)

if __name__ == '__main__':
    main()