cloud-reporter/backend/test_task_api.py

#!/usr/bin/env python
"""
测试任务API调用

模拟实际的任务创建过程
"""
import os
import sys

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

def test_task_creation():
    """测试任务创建过程"""
    from app import create_app, db
    from app.models import Task, User, AWSCredential
    
    app = create_app('testing')
    
    with app.app_context():
        # 创建测试数据
        db.create_all()
        
        # 创建测试用户
        user = User(
            username='testuser',
            email='test@example.com',
            role='admin'
        )
        user.set_password('testpass')
        db.session.add(user)
        
        # 创建测试凭证
        credential = AWSCredential(
            name='Test Credential',
            credential_type='access_key',
            account_id='123456789012',
            access_key_id='AKIATEST123456789',
            is_active=True
        )
        credential.set_secret_access_key('test-secret-key')
        db.session.add(credential)
        db.session.commit()
        
        print(f"✅ 测试数据创建成功")
        print(f"   用户ID: {user.id}")
        print(f"   凭证ID: {credential.id}")
        
        # 模拟任务创建
        task = Task(
            name='Test Task',
            status='pending',
            progress=0,
            created_by=user.id
        )
        task.credential_ids = [credential.id]
        task.regions = ['us-east-1']
        task.project_metadata = {
            'clientName': 'Test Client',
            'projectName': 'Test Project'
        }
        
        db.session.add(task)
        db.session.commit()
        
        print(f"✅ 测试任务创建成功，任务ID: {task.id}")
        
        # 测试Celery任务提交
        try:
            print("🔍 尝试提交Celery任务...")
            from app.tasks.scan_tasks import scan_aws_resources
            
            celery_task = scan_aws_resources.delay(
                task_id=task.id,
                credential_ids=[credential.id],
                regions=['us-east-1'],
                project_metadata={
                    'clientName': 'Test Client',
                    'projectName': 'Test Project'
                }
            )
            
            print(f"✅ Celery任务提交成功: {celery_task.id}")
            return True
            
        except Exception as e:
            print(f"❌ Celery任务提交失败: {e}")
            print(f"   错误类型: {type(e).__name__}")
            
            # 尝试Mock模式
            try:
                print("🔄 尝试Mock模式...")
                from app.tasks.mock_tasks import scan_aws_resources
                
                celery_task = scan_aws_resources.delay(
                    task_id=task.id,
                    credential_ids=[credential.id],
                    regions=['us-east-1'],
                    project_metadata={
                        'clientName': 'Test Client',
                        'projectName': 'Test Project'
                    }
                )
                
                print(f"✅ Mock任务提交成功: {celery_task.id}")
                return True
                
            except Exception as e2:
                print(f"❌ Mock任务也失败: {e2}")
                return False


def main():
    """运行测试"""
    print("="*50)
    print("任务API测试")
    print("="*50)
    
    try:
        result = test_task_creation()
        if result:
            print("\n🎉 任务创建测试成功!")
        else:
            print("\n❌ 任务创建测试失败!")
    except Exception as e:
        print(f"\n❌ 测试异常: {e}")
        import traceback
        traceback.print_exc()
    
    print("="*50)


if __name__ == '__main__':
    main()