forked from phoenix/litellm-mirror
84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
# What this tests?
|
|
## Tests /spend endpoints.
|
|
|
|
import pytest
|
|
import asyncio
|
|
import aiohttp
|
|
|
|
|
|
async def generate_key(session, models=[]):
|
|
url = "http://0.0.0.0:4000/key/generate"
|
|
headers = {"Authorization": "Bearer sk-1234", "Content-Type": "application/json"}
|
|
data = {
|
|
"models": models,
|
|
"duration": None,
|
|
}
|
|
|
|
async with session.post(url, headers=headers, json=data) as response:
|
|
status = response.status
|
|
response_text = await response.text()
|
|
|
|
print(response_text)
|
|
print()
|
|
|
|
if status != 200:
|
|
raise Exception(f"Request did not return a 200 status code: {status}")
|
|
return await response.json()
|
|
|
|
|
|
async def chat_completion(session, key):
|
|
url = "http://0.0.0.0:4000/chat/completions"
|
|
headers = {
|
|
"Authorization": f"Bearer {key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
data = {
|
|
"model": "gpt-3.5-turbo",
|
|
"messages": [
|
|
{"role": "system", "content": "You are a helpful assistant."},
|
|
{"role": "user", "content": "Hello!"},
|
|
],
|
|
}
|
|
|
|
async with session.post(url, headers=headers, json=data) as response:
|
|
status = response.status
|
|
response_text = await response.text()
|
|
|
|
print(response_text)
|
|
print()
|
|
|
|
if status != 200:
|
|
raise Exception(f"Request did not return a 200 status code: {status}")
|
|
|
|
return await response.json()
|
|
|
|
|
|
async def get_spend_logs(session, request_id):
|
|
url = f"http://0.0.0.0:4000/spend/logs?request_id={request_id}"
|
|
headers = {"Authorization": "Bearer sk-1234", "Content-Type": "application/json"}
|
|
|
|
async with session.get(url, headers=headers) as response:
|
|
status = response.status
|
|
response_text = await response.text()
|
|
|
|
print(response_text)
|
|
print()
|
|
|
|
if status != 200:
|
|
raise Exception(f"Request did not return a 200 status code: {status}")
|
|
return await response.json()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_spend_logs():
|
|
"""
|
|
- Create key
|
|
- Make call (makes sure it's in spend logs)
|
|
- Get request id from logs
|
|
"""
|
|
async with aiohttp.ClientSession() as session:
|
|
key_gen = await generate_key(session=session)
|
|
key = key_gen["key"]
|
|
response = await chat_completion(session=session, key=key)
|
|
await asyncio.sleep(5)
|
|
await get_spend_logs(session=session, request_id=response["id"])
|