Spaces:
Running
Running
| 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") | |
| ) | |
| 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}") | |