litellm/tests/test_health.py at main · BerriAI/litellm · GitHub

Name: litellm/tests/test_health.py at main · BerriAI/litellm · GitHub
Rating: 4.5 (8225 reviews)
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
# What this tests?
## Tests /health + /routes endpoints.

importpytest
importasyncio
importaiohttp


asyncdefhealth(session, call_key):
url="http://0.0.0.0:4000/health"
headers= {
"Authorization": f"Bearer {call_key}",
"Content-Type": "application/json",
 }

asyncwithsession.get(url, headers=headers) asresponse:
status=response.status
response_text=awaitresponse.text()

print(f"Response (Status code: {status}):")
print(response_text)
print()

ifstatus!=200:
raiseException(f"Request did not return a 200 status code: {status}")

returnawaitresponse.json()


asyncdefgenerate_key(session):
url="http://0.0.0.0:4000/key/generate"
headers= {"Authorization": "Bearer sk-1234", "Content-Type": "application/json"}
data= {
"models": ["gpt-4", "text-embedding-ada-002", "dall-e-2"],
"duration": None,
 }

asyncwithsession.post(url, headers=headers, json=data) asresponse:
status=response.status
response_text=awaitresponse.text()

print(response_text)
print()

ifstatus!=200:
raiseException(f"Request did not return a 200 status code: {status}")
returnawaitresponse.json()


@pytest.mark.asyncio
asyncdeftest_health():
"""
 - Call /health
 """
asyncwithaiohttp.ClientSession() assession:
# as admin #
all_healthy_models=awaithealth(session=session, call_key="sk-1234")
total_model_count= (
all_healthy_models["healthy_count"] +all_healthy_models["unhealthy_count"]
 )
asserttotal_model_count>0


@pytest.mark.asyncio
asyncdeftest_health_readiness():
"""
 Check if 200
 """
asyncwithaiohttp.ClientSession() assession:
url="http://0.0.0.0:4000/health/readiness"
asyncwithsession.get(url) asresponse:
status=response.status
response_json=awaitresponse.json()

print(response_json)
assert"litellm_version"inresponse_json
assert"status"inresponse_json

ifstatus!=200:
raiseException(f"Request did not return a 200 status code: {status}")


@pytest.mark.asyncio
asyncdeftest_health_liveliness():
"""
 Check if 200
 """
asyncwithaiohttp.ClientSession() assession:
url="http://0.0.0.0:4000/health/liveliness"
asyncwithsession.get(url) asresponse:
status=response.status
response_text=awaitresponse.text()

print(response_text)
print()

ifstatus!=200:
raiseException(f"Request did not return a 200 status code: {status}")


@pytest.mark.asyncio
asyncdeftest_routes():
"""
 Check if 200
 """
asyncwithaiohttp.ClientSession() assession:
url="http://0.0.0.0:4000/routes"
asyncwithsession.get(url) asresponse:
status=response.status
response_text=awaitresponse.text()

print(response_text)
print()

ifstatus!=200:
raiseException(f"Request did not return a 200 status code: {status}")