webseek-backend/test.py at master · datavisards/webseek-backend · 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
142
143
144
145
146
147
148
149
150
151
152
153
import requests
import json
import logging
from utils import setup_logging

setup_logging(logging.DEBUG)
logger = logging.getLogger("webseek_backend.test")

request = {
  "description": "What are in the instance view? Please summarize.",
  "user_preferences": {
    "summary_type": "comprehensive",
    "include_charts": False
  }
}

# send request to the server
def test_process_task():
    """Test the process-task endpoint."""
    url = "http://localhost:8000/api/process-task"

    try:
        logger.info("Sending request to server...")
        logger.info(f"Request: {json.dumps(request, indent=2)}")

        response = requests.post(url, json=request)

        logger.info(f"Status Code: {response.status_code}")

        if response.status_code == 200:
            result = response.json()
            logger.info("✅ Request successful!")
            logger.info(f"Result: {json.dumps(result, indent=2)}")
        else:
            logger.error(f"❌ Request failed with status {response.status_code}")
            logger.error(f"Error: {response.text}")

    except requests.exceptions.ConnectionError:
        logger.error("❌ Connection error: Make sure the server is running on localhost:8000")
    except Exception as e:
        logger.error(f"❌ Error: {e}")

def test_health_check():
    """Test the health check endpoint."""
    url = "http://localhost:8000/health"

    try:
        logger.info("\nTesting health check...")
        response = requests.get(url)

        if response.status_code == 200:
            result = response.json()
            logger.info("✅ Health check successful!")
            logger.info(f"Status: {result.get('status')}")
            logger.info(f"Agents: {result.get('agents', {})}")
            logger.info(f"WebSocket connections: {result.get('websocket_connections', 0)}")
        else:
            logger.error(f"❌ Health check failed with status {response.status_code}")

    except requests.exceptions.ConnectionError:
        logger.error("❌ Connection error: Make sure the server is running on localhost:8000")
    except Exception as e:
        logger.error(f"❌ Error: {e}")

def test_root_endpoint():
    """Test the root endpoint."""
    url = "http://localhost:8000/"

    try:
        logger.info("\nTesting root endpoint...")
        response = requests.get(url)

        if response.status_code == 200:
            result = response.json()
            logger.info("✅ Root endpoint successful!")
            logger.info(f"Message: {result.get('message')}")
            logger.info(f"Version: {result.get('version')}")
            logger.info(f"Status: {result.get('status')}")
            logger.info(f"WebSocket URL: {result.get('websocket_url')}")
        else:
            logger.error(f"❌ Root endpoint failed with status {response.status_code}")

    except requests.exceptions.ConnectionError:
        logger.error("❌ Connection error: Make sure the server is running on localhost:8000")
    except Exception as e:
        logger.error(f"❌ Error: {e}")

def test_render_vega_lite():
    """Test the render-vega-lite endpoint with a valid spec."""
    url = "http://localhost:8000/api/render-vega-lite/"

    # Create a valid test specification
    test_spec = {
        "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
        "data": {
            "values": [
                {"name": "Camera 1", "price": 82.99},
                {"name": "Camera 2", "price": 99.99}
            ]
        },
        "mark": "bar",
        "encoding": {
            "x": {"field": "name", "type": "nominal", "title": "Product"},
            "y": {"field": "price", "type": "quantitative", "title": "Price ($)"}
        },
        "width": 400,
        "height": 300
    }

    try:
        logger.info("\nTesting render-vega-lite endpoint...")
        logger.info(f"Test spec: {json.dumps(test_spec, indent=2)}")

        response = requests.post(url, json=test_spec)

        logger.info(f"Status Code: {response.status_code}")

        if response.status_code == 200:
            logger.info("✅ Vega-Lite rendering successful!")
            logger.info(f"Content-Type: {response.headers.get('content-type')}")
            logger.info(f"Content-Length: {len(response.content)} bytes")

            # Optionally save the SVG for inspection
            with open("test_chart.svg", "wb") as f:
                f.write(response.content)
            logger.info("📊 Chart saved as test_chart.svg")
        else:
            logger.error(f"❌ Vega-Lite rendering failed with status {response.status_code}")
            logger.error(f"Error: {response.text}")

    except requests.exceptions.ConnectionError:
        logger.error("❌ Connection error: Make sure the server is running on localhost:8000")
    except Exception as e:
        logger.error(f"❌ Error: {e}")

if __name__ == "__main__":
    logger.info("🧪 Testing WebSeek Backend API")
    logger.info("=" * 50)

    # Test root endpoint
    test_root_endpoint()

    # Test health check
    test_health_check()

    # Test main task processing
    test_process_task()

    # Test Vega-Lite rendering
    test_render_vega_lite()

    logger.info("\n" + "=" * 50)
    logger.info(" Test completed!")