File size: 3,817 Bytes
4e10023
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
#!/usr/bin/env python3
"""
Test script for the AI Backend Service API endpoints
"""

import requests
import json
import time

BASE_URL = "http://localhost:8000"

def test_health():
    """Test health endpoint"""
    print("πŸ” Testing health endpoint...")
    response = requests.get(f"{BASE_URL}/health")
    print(f"Status: {response.status_code}")
    print(f"Response: {json.dumps(response.json(), indent=2)}")
    print()

def test_root():
    """Test root endpoint"""
    print("πŸ” Testing root endpoint...")
    response = requests.get(f"{BASE_URL}/")
    print(f"Status: {response.status_code}")
    print(f"Response: {json.dumps(response.json(), indent=2)}")
    print()

def test_models():
    """Test models endpoint"""
    print("πŸ” Testing models endpoint...")
    response = requests.get(f"{BASE_URL}/v1/models")
    print(f"Status: {response.status_code}")
    print(f"Response: {json.dumps(response.json(), indent=2)}")
    print()

def test_chat_completion():
    """Test chat completion endpoint"""
    print("πŸ” Testing chat completion endpoint...")
    data = {
        "model": "microsoft/DialoGPT-medium",
        "messages": [
            {"role": "user", "content": "Hello! How are you?"}
        ],
        "max_tokens": 100,
        "temperature": 0.7
    }
    
    response = requests.post(f"{BASE_URL}/v1/chat/completions", json=data)
    print(f"Status: {response.status_code}")
    print(f"Response: {json.dumps(response.json(), indent=2)}")
    print()

def test_completion():
    """Test completion endpoint"""
    print("πŸ” Testing completion endpoint...")
    data = {
        "prompt": "The weather today is",
        "max_tokens": 50,
        "temperature": 0.7
    }
    
    response = requests.post(f"{BASE_URL}/v1/completions", json=data)
    print(f"Status: {response.status_code}")
    print(f"Response: {json.dumps(response.json(), indent=2)}")
    print()

def test_streaming_chat():
    """Test streaming chat completion"""
    print("πŸ” Testing streaming chat completion...")
    data = {
        "model": "microsoft/DialoGPT-medium",
        "messages": [
            {"role": "user", "content": "Tell me a short joke"}
        ],
        "max_tokens": 100,
        "temperature": 0.7,
        "stream": True
    }
    
    response = requests.post(f"{BASE_URL}/v1/chat/completions", json=data, stream=True)
    print(f"Status: {response.status_code}")
    print("Streaming response:")
    
    for line in response.iter_lines():
        if line:
            line_str = line.decode('utf-8')
            if line_str.startswith('data: '):
                data_part = line_str[6:]  # Remove 'data: ' prefix
                if data_part == '[DONE]':
                    print("Stream completed!")
                    break
                try:
                    chunk_data = json.loads(data_part)
                    if 'choices' in chunk_data and chunk_data['choices']:
                        delta = chunk_data['choices'][0].get('delta', {})
                        if 'content' in delta:
                            print(delta['content'], end='', flush=True)
                except json.JSONDecodeError:
                    pass
    print("\n")

if __name__ == "__main__":
    print("πŸš€ Testing AI Backend Service API")
    print("=" * 50)
    
    # Wait a moment for service to be ready
    time.sleep(2)
    
    try:
        test_root()
        test_health()
        test_models()
        test_chat_completion()
        test_completion()
        test_streaming_chat()
        
        print("βœ… All tests completed!")
        
    except requests.exceptions.ConnectionError:
        print("❌ Could not connect to the service. Make sure it's running on localhost:8000")
    except Exception as e:
        print(f"❌ Test failed with error: {e}")