flower/test_api_endpoints.py at development · bijankhazaei/flower · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
#!/usr/bin/env python3
"""
Test script for all API endpoints
"""

import asyncio
import json
from fastapi.testclient import TestClient
from app.Ship.Engine.app import create_app


def test_api_endpoints():
    """Test all API endpoints"""

    # Create test client
    app = create_app()
    client = TestClient(app)

    print("🧪 Testing API endpoints...")

    # Test health check
    print("\n1. Testing health check...")
    response = client.get("/health")
    assert response.status_code == 200
    print("✅ Health check passed")

    # Test node types listing
    print("\n2. Testing node types listing...")
    response = client.get("/api/nodes/types")
    assert response.status_code == 200
    data = response.json()
    assert "node_types" in data
    assert data["count"] > 0
    print(f"✅ Found {data['count']} node types")

    # Test specific node info
    print("\n3. Testing specific node info...")
    response = client.get("/api/nodes/types/TextProcessorNode")
    assert response.status_code == 200
    data = response.json()
    assert data["metadata"]["name"] == "Text Processor"
    print("✅ Node info retrieved successfully")

    # Test node validation
    print("\n4. Testing node validation...")
    node_config = {
        "node_type": "TextProcessorNode",
        "config": {
            "id": "test_processor",
            "parameters": {
                "operation": "uppercase"
            }
        }
    }
    response = client.post("/api/nodes/validate", json=node_config)
    assert response.status_code == 200
    data = response.json()
    assert data["valid"] == True
    print("✅ Node validation passed")

    # Test flow validation
    print("\n5. Testing flow validation...")
    flow_def = {
        "flow_definition": {
            "id": "test_flow",
            "name": "Test Flow",
            "nodes": [
                {
                    "id": "input_1",
                    "type": "InputNode",
                    "parameters": {"input_key": "text"}
                },
                {
                    "id": "output_1",
                    "type": "OutputNode",
                    "parameters": {"output_key": "result"},
                    "input_mappings": {
                        "input": {
                            "source_node": "input_1",
                            "source_output": "output"
                        }
                    }
                }
            ],
            "connections": [
                {"source": "input_1", "target": "output_1"}
            ]
        }
    }
    response = client.post("/api/compiler/validate", json=flow_def)
    assert response.status_code == 200
    data = response.json()
    assert data["valid"] == True
    print("✅ Flow validation passed")

    # Test flow compilation
    print("\n6. Testing flow compilation...")
    response = client.post("/api/compiler/compile", json=flow_def)
    assert response.status_code == 200
    data = response.json()
    assert data["success"] == True
    assert "code" in data
    print("✅ Flow compilation passed")

    # Test flow execution
    print("\n7. Testing flow execution...")
    execute_request = {
        **flow_def,
        "inputs": {"text": "hello world"}
    }
    response = client.post("/api/compiler/execute", json=execute_request)
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "completed"
    assert "execution_id" in data
    execution_id = data["execution_id"]
    print(f"✅ Flow execution completed: {execution_id}")

    # Test execution monitoring
    print("\n8. Testing execution monitoring...")
    response = client.get(f"/api/executions/{execution_id}")
    assert response.status_code == 200
    data = response.json()
    assert data["execution_id"] == execution_id
    assert data["status"] == "completed"
    print("✅ Execution monitoring passed")

    # Test executions listing
    print("\n9. Testing executions listing...")
    response = client.get("/api/executions/")
    assert response.status_code == 200
    data = response.json()
    assert "executions" in data
    assert data["count"] >= 1
    print(f"✅ Found {data['count']} executions")

    print("\n🎉 All API tests passed!")


if __name__ == "__main__":
    test_api_endpoints()