litellm/tests/local_testing/test_async_fn.py
devin-ai-integration[bot] dfe4df8c8a
test: move whole-unit legacy test files into tests/unit and delete dead skips (#44807)
* test: delete unconditionally skipped legacy tests

* test: move whole-unit legacy test files into tests/unit

* test: keep moved legacy tests free of import-time global state

* test: keep the module-level invocation scan pointed at tests/local_testing

* ci: drop the agent_testing CircleCI job emptied by the move

* test: fix moved-test isolation and router coverage

* test: add Tinyfish search package marker

* test: isolate moved tests from logger state leaks

* test: isolate Helicone logging fixture state

* test: isolate Vertex pass-through credentials between moved tests

* test: cancel S3 periodic flush tasks started by moved tests

* ci: restore CircleCI assistant test selection after move

* test: prevent Datadog datetime import shadowing

* ci: drop the litellm_assistants_api_testing CircleCI job emptied by the move

* test: deduplicate imports in rebased unit tests

* test: remove duplicate passthrough router patch import

* test: remove moved legacy source files after rebase

* test: align moved tests with rebased main

* test: carry main's legacy-file edits into moved destinations

* test: make the moved cost map fallback tests assert the fetch and the backup

The four fallback cases only checked the result was non-empty, so they still
passed with integrity validation disabled. They now inject a mock client, assert
one fetch happened, and assert the result is exactly the local backup with the
fallback reason recorded.

---------

Co-authored-by: yuneng <yuneng@berri.ai>
2026-10-07 05:02:28 +00:00

93 lines
2.1 KiB
Python

#### What this tests ####
# This tests the the acompletion function #
import asyncio
import pytest
import litellm
from litellm import acompletion
litellm.num_retries = 3
# test_sync_response_anyscale()
def test_async_response_openai():
litellm.set_verbose = True
async def test_get_response():
user_message = "Hello, how are you?"
messages = [{"content": user_message, "role": "user"}]
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["location"],
},
},
}
]
try:
response = await acompletion(
model="gpt-3.5-turbo",
messages=messages,
tools=tools,
parallel_tool_calls=True,
timeout=5,
)
print(f"response: {response}")
print(f"response ms: {response._response_ms}")
except litellm.Timeout as e:
pass
except Exception as e:
pytest.fail(f"An exception occurred: {e}")
print(e)
asyncio.run(test_get_response())
# test_async_response_openai()
# test_async_anyscale_response()
# test_async_completion_cloudflare()
# test_get_cloudflare_response_streaming()
# test_get_response_streaming()
# test_get_response_non_openai_streaming()