aiagent/backend/test_workflow_validation.py

"""
工作流验证功能测试
"""
import sys
import os

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from app.services.workflow_validator import validate_workflow


def test_valid_workflow():
    """测试有效的工作流"""
    print("=" * 60)
    print("测试1: 有效的工作流")
    print("=" * 60)
    
    nodes = [
        {"id": "start-1", "type": "start", "data": {"label": "开始"}},
        {"id": "llm-1", "type": "llm", "data": {"label": "LLM", "prompt": "测试", "provider": "openai", "model": "gpt-3.5-turbo"}},
        {"id": "end-1", "type": "end", "data": {"label": "结束"}}
    ]
    
    edges = [
        {"id": "e1", "source": "start-1", "target": "llm-1"},
        {"id": "e2", "source": "llm-1", "target": "end-1"}
    ]
    
    result = validate_workflow(nodes, edges)
    print(f"验证结果: {result['valid']}")
    print(f"错误: {result['errors']}")
    print(f"警告: {result['warnings']}")
    
    assert result['valid'] == True, "有效工作流应该通过验证"
    print("✅ 测试通过\n")
    return True


def test_missing_start_node():
    """测试缺少开始节点"""
    print("=" * 60)
    print("测试2: 缺少开始节点")
    print("=" * 60)
    
    nodes = [
        {"id": "llm-1", "type": "llm", "data": {"label": "LLM"}},
        {"id": "end-1", "type": "end", "data": {"label": "结束"}}
    ]
    
    edges = [
        {"id": "e1", "source": "llm-1", "target": "end-1"}
    ]
    
    result = validate_workflow(nodes, edges)
    print(f"验证结果: {result['valid']}")
    print(f"错误: {result['errors']}")
    
    assert result['valid'] == False, "缺少开始节点应该验证失败"
    assert any("开始节点" in error for error in result['errors']), "应该包含开始节点相关的错误"
    print("✅ 测试通过\n")
    return True


def test_cycle_detection():
    """测试循环检测"""
    print("=" * 60)
    print("测试3: 循环检测")
    print("=" * 60)
    
    nodes = [
        {"id": "start-1", "type": "start", "data": {"label": "开始"}},
        {"id": "node-1", "type": "default", "data": {"label": "节点1"}},
        {"id": "node-2", "type": "default", "data": {"label": "节点2"}}
    ]
    
    edges = [
        {"id": "e1", "source": "start-1", "target": "node-1"},
        {"id": "e2", "source": "node-1", "target": "node-2"},
        {"id": "e3", "source": "node-2", "target": "node-1"}  # 形成循环
    ]
    
    result = validate_workflow(nodes, edges)
    print(f"验证结果: {result['valid']}")
    print(f"错误: {result['errors']}")
    
    assert result['valid'] == False, "包含循环的工作流应该验证失败"
    assert any("循环" in error for error in result['errors']), "应该包含循环相关的错误"
    print("✅ 测试通过\n")
    return True


def test_unreachable_nodes():
    """测试不可达节点"""
    print("=" * 60)
    print("测试4: 不可达节点")
    print("=" * 60)
    
    nodes = [
        {"id": "start-1", "type": "start", "data": {"label": "开始"}},
        {"id": "node-1", "type": "default", "data": {"label": "节点1"}},
        {"id": "node-2", "type": "default", "data": {"label": "节点2"}}  # 不可达
    ]
    
    edges = [
        {"id": "e1", "source": "start-1", "target": "node-1"}
        # node-2 没有连接，不可达
    ]
    
    result = validate_workflow(nodes, edges)
    print(f"验证结果: {result['valid']}")
    print(f"警告: {result['warnings']}")
    
    assert result['valid'] == True, "不可达节点不应该导致验证失败（只是警告）"
    assert any("不可达" in warning for warning in result['warnings']), "应该包含不可达节点的警告"
    print("✅ 测试通过\n")
    return True


def test_condition_node_validation():
    """测试条件节点验证"""
    print("=" * 60)
    print("测试5: 条件节点验证")
    print("=" * 60)
    
    nodes = [
        {"id": "start-1", "type": "start", "data": {"label": "开始"}},
        {"id": "condition-1", "type": "condition", "data": {"label": "条件", "condition": "{value} > 10"}},
        {"id": "end-1", "type": "end", "data": {"label": "结束"}}
    ]
    
    edges = [
        {"id": "e1", "source": "start-1", "target": "condition-1"},
        {"id": "e2", "source": "condition-1", "target": "end-1", "sourceHandle": "true"}
        # 缺少false分支
    ]
    
    result = validate_workflow(nodes, edges)
    print(f"验证结果: {result['valid']}")
    print(f"警告: {result['warnings']}")
    
    assert result['valid'] == True, "缺少分支不应该导致验证失败（只是警告）"
    assert any("False分支" in warning for warning in result['warnings']), "应该包含缺少分支的警告"
    print("✅ 测试通过\n")
    return True


def main():
    """主测试函数"""
    print("\n🚀 开始工作流验证功能测试\n")
    
    results = []
    results.append(test_valid_workflow())
    results.append(test_missing_start_node())
    results.append(test_cycle_detection())
    results.append(test_unreachable_nodes())
    results.append(test_condition_node_validation())
    
    print("=" * 60)
    print("测试结果汇总")
    print("=" * 60)
    passed = sum(results)
    total = len(results)
    print(f"通过: {passed}/{total}")
    print(f"失败: {total - passed}/{total}")
    
    if passed == total:
        print("\n✅ 所有测试通过！工作流验证功能正常！")
    else:
        print(f"\n⚠️  有 {total - passed} 个测试失败")


if __name__ == "__main__":
    main()