Spaces:
Running
Running
File size: 2,559 Bytes
314597a f4d0a5c 314597a e78c9e1 314597a e78c9e1 314597a e78c9e1 314597a e78c9e1 314597a |
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 |
import pytest
import httpx
from openai import OpenAI
import os
import time
import random
# Configure the OpenAI client to point to the local server.
# The API key is required by the library but not used by the local server.
client = OpenAI(
base_url="https://solarumasteridion-lenith.hf.space/",
api_key=os.environ.get("OPENAI_API_KEY", "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")
)
@pytest.mark.asyncio
async def test_chat_completion_streaming_openai():
"""
Test the /chat/completions endpoint for streaming requests.
This test first fetches an available model from the /models endpoint.
"""
messages = [{"role": "user", "content": "bionomial thoerem calc in python concise code but explain before the code block"}]
print(f"\nTesting streaming chat completion with model 'Gemini 2.0 Flash'...")
accumulated_content = ""
deltas_received = 0
start_time = time.time()
try:
stream = client.chat.completions.create(
model="Gemini 2.0 Flash",
messages=messages,
stream=True
)
for chunk in stream:
assert chunk.id is not None
assert chunk.object == "chat.completion.chunk"
assert len(chunk.choices) > 0
choice = chunk.choices[0]
delta = choice.delta
if delta.role is not None:
print(f"\nReceived role: {delta.role}")
assert delta.role == "assistant"
if delta.content is not None:
delta_content = delta.content
assert isinstance(delta_content, str)
accumulated_content += delta_content
deltas_received += 1
print(delta_content, end="", flush=True)
if choice.finish_reason is not None:
print(f"\nReceived finish_reason: {choice.finish_reason}")
assert choice.finish_reason == "stop"
break
end_time = time.time()
print(f"\n--- Streaming complete ---")
print(f"Total deltas received: {deltas_received}")
print(f"Full response ({len(accumulated_content)} chars) received in {end_time - start_time:.2f}s:")
print(accumulated_content)
assert deltas_received > 0, "No content deltas were received."
assert len(accumulated_content) > 0, "Accumulated content is empty."
except Exception as e:
pytest.fail(f"Streaming chat completion failed: {e}")
|