Automatically mock OpenAI requests
Project description
openai-responses-python
Automatically mock OpenAI requests
[!NOTE] This project does not try to generate fake responses from the models. Any part of a response that would be generated by a model will need to be defined by the user or will be left as the fallback value (i.e. empty string for
str
type fields).
Installation
pip install openai-responses
Quick start
This is an example of how to mock a call to POST https://api.openai.com/v1/chat/completions
with a synthetic latency of 5 seconds for the call.
from openai import OpenAI
import openai_responses
@openai_responses.mock.chat.completions(
choices=[
{"message": {"content": "Hello, how can I help?"}},
{"message": {"content": "Hi! I'm here to help!"}},
{"message": {"content": "How can I help?"}},
],
latency=5,
)
def test_create_completion_with_multiple_choices():
client = OpenAI(api_key="fakeKey")
completion = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello!"},
],
n=3,
)
assert len(completion.choices) == 3
Endpoint Support
[!IMPORTANT] Currently, there is no support for streaming across the board. This is a top feature request so once I have time to tackle it I will.
❌ = Not supported
🟡 = Partially supported
✅ = Fully supported
Endpoint | Supported | Streaming Supported | Mock Type |
---|---|---|---|
Audio | ❌ | - | Stateless |
Chat | ✅ | ❌ | Stateless |
Embeddings | ✅ | - | Stateless |
Fine-tuning | ❌ | - | Stateful |
Files | 🟡[^1] | - | Stateful |
Images | ❌ | - | Stateless |
Models | ❌ | - | Stateless[^2] |
Moderations | ❌ | - | Stateless |
Assistants | 🟡[^3] | - | Stateful |
Threads | ✅ | - | Stateful |
Messages | 🟡[^3] | - | Stateful |
Runs | 🟡[^4][^5][^6] | ❌ | Stateful |
Completions | ❌ | ❌ | Stateless |
[^1]: Need to add support for retrieving file content [^2]: Stateless until fine-tuning is supported then it will need to be stateful [^3]: Need to add support for attached files [^4]: Need to add support for create thread and run [^5]: Fragile API for run steps [^6]: No state changes on submit tool call
Mocks
To activate the mock router, you just need to decorate a test function.
Each mock decorator will have the following arguments:
latency
- synthetic latency in seconds to introduce to the call(s). Defaults to0.0
.failures
- number of failures to simulate. Defaults to0
.
Some decorators will have additional arguments which are documented below.
[!TIP] For more examples see tests/examples
Chat Completions
To mock the Chat API you will use openai_responses.mock.chat.completions()
.
from openai import OpenAI
import openai_responses
@openai_responses.mock.chat.completions(
choices=[
{"message": {"content": "Hello, how can I help?"}},
]
)
def test_create_completion():
client = OpenAI(api_key="fakeKey")
completion = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello!"},
],
)
assert len(completion.choices) == 1
assert completion.choices[0].message.content == "Hello, how can I help?"
Additional Arguments
choices
: a list of choices the call should return. Defaults to[]
.
Mocked Routes
POST https://api.openai.com/v1/chat/completions
Embeddings
To mock the Embeddings API you will use openai_responses.mock.embeddings()
.
import random
import pytest
from openai import OpenAI, AsyncOpenAI
import openai_responses
EMBEDDING = [random.uniform(0.01, -0.01) for _ in range(100)]
@openai_responses.mock.embeddings(embedding=EMBEDDING)
def test_create_embeddings():
client = OpenAI(api_key="fakeKey")
embeddings = client.embeddings.create(
model="text-embedding-ada-002",
input="The food was delicious and the waiter...",
encoding_format="float",
)
assert len(embeddings.data) == 1
assert embeddings.data[0].embedding == EMBEDDING
assert embeddings.model == "text-embedding-ada-002"
Additional Arguments
embedding
: a list of floats that represent an embedding returned from the call. Defaults to[]
.
Mocked Routes
POST https://api.openai.com/v1/embeddings
Files
To mock the Files API you will use openai_responses.mock.files()
.
from openai import OpenAI
import openai_responses
@openai_responses.mock.files()
def test_upload_file():
client = OpenAI(api_key="fakeKey")
file = client.files.create(
file=open("tests/examples/example.json", "rb"),
purpose="assistants",
)
assert file.filename == "example.json"
assert file.purpose == "assistants"
Additional Arguments
state_store
: Optional state store override. See state below for more info.
Mocked Routes
POST https://api.openai.com/v1/files
GET https://api.openai.com/v1/files
GET https://api.openai.com/v1/files/{file_id}
DELETE https://api.openai.com/v1/files/{file_id}
Assistants
To mock the Assistants API you will use openai_responses.mock.beta.assistants()
.
from openai import OpenAI
import openai_responses
@openai_responses.mock.beta.assistants()
def test_create_assistant():
client = OpenAI(api_key="fakeKey")
my_assistant = client.beta.assistants.create(
instructions="You are a personal math tutor. When asked a question, write and run Python code to answer the question.",
name="Math Tutor",
tools=[{"type": "code_interpreter"}],
model="gpt-4",
)
assert my_assistant.name == "Math Tutor"
assert my_assistant.model == "gpt-4"
Additional Arguments
state_store
: Optional state store override. See state below for more info.
Mocked Routes
POST https://api.openai.com/v1/assistants
GET https://api.openai.com/v1/assistants
GET https://api.openai.com/v1/assistants/{assistant_id}
POST https://api.openai.com/v1/assistants/{assistant_id}
DELETE https://api.openai.com/v1/assistants/{assistant_id}
Threads
To mock the Threads API you will use openai_responses.mock.beta.threads()
.
from openai import OpenAI
import openai_responses
@openai_responses.mock.beta.threads()
def test_create_empty_thread():
client = OpenAI(api_key="fakeKey")
empty_thread = client.beta.threads.create()
assert empty_thread.id
Additional Arguments
state_store
: Optional state store override. See state below for more info.
Mocked Routes
POST https://api.openai.com/v1/threads
GET https://api.openai.com/v1/threads/{thread_id}
POST https://api.openai.com/v1/threads/{thread_id}
DELETE https://api.openai.com/v1/threads/{thread_id}
Messages
To mock the Messages API you will use openai_responses.mock.beta.threads.messages()
.
from openai import OpenAI
import openai_responses
@openai_responses.mock.beta.threads.messages()
def test_create_thread_message():
client = OpenAI(api_key="fakeKey")
thread_message = client.beta.threads.messages.create(
"thread_abc123",
role="user",
content="How does AI work? Explain it in simple terms.",
)
assert thread_message.id
Additional Arguments
state_store
: Optional state store override. See state below for more info.validate_thread_exists
: Whether to check the state store to see if the thread exists. Defaults toFalse
.
Mocked Routes
POST https://api.openai.com/v1/threads/{thread_id}/messages
GET https://api.openai.com/v1/threads/{thread_id}/messages
GET https://api.openai.com/v1/threads/{thread_id}/messages/{message_id}
POST https://api.openai.com/v1/threads/{thread_id}/messages/{message_id}
Runs
To mock the Runs API you will use openai_responses.mock.beta.threads.runs()
.
from openai import OpenAI
import openai_responses
@openai_responses.mock.beta.threads.runs(
sequence={
"create": [
{"status": "in_progress"},
],
"retrieve": [
{"status": "in_progress"},
{"status": "in_progress"},
{"status": "in_progress"},
{"status": "in_progress"},
{"status": "completed"},
],
}
)
def test_polled_get_status():
client = OpenAI(api_key="fakeKey")
run = client.beta.threads.runs.create(
thread_id="thread_abc123",
assistant_id="asst_abc123",
)
while run.status != "completed":
assert run.status == "in_progress"
run = client.beta.threads.runs.retrieve(
run.id,
thread_id="thread_abc123"
)
assert run.status == "completed"
Additional Arguments
sequence
: Sequence of run states that the calls will exhaust. Can be list or dict. Defaults to[]
.state_store
: Optional state store override. See state below for more info.validate_thread_exists
: Whether to check the state store to see if the thread exists. Defaults toFalse
.validate_assistant_exists
: Whether to check the state store to see if the assistant exists. Defaults toFalse
.
Mocked Routes
POST https://api.openai.com/v1/threads/{thread_id}/runs
GET https://api.openai.com/v1/threads/{thread_id}/runs
GET https://api.openai.com/v1/threads/{thread_id}/runs/{run_id}
POST https://api.openai.com/v1/threads/{thread_id}/runs/{run_id}
POST https://api.openai.com/v1/threads/{thread_id}/runs/{run_id}/cancel
GET https://api.openai.com/v1/threads/{thread_id}/runs/{run_id}/steps
GET https://api.openai.com/v1/threads/{thread_id}/runs/{run_id}/steps/{step_id}
POST https://api.openai.com/v1/threads/{thread_id}/runs/{run_id}/submit_tool_outputs
Sequences
Since run states can be changed at any time on the API server, the suggested usage pattern is to poll for updates. The sequence
argument allows you to define the sequence of states that will be returned for each corresponding call. Although you likely will not call create
more than once during standard API usage, this sequence is available to help you set the initial state.
Resource validation
If either validate_thread_exists
or validate_assistant_exists
then those related values will be used in the state so you do not need to set them in the sequence. If you do, your manual sequence entry will take precedent over the other values.
[!WARNING] Run step routes are mocked but API is volatile right now so it is not documented or included in the examples.
Mocker Classes
Each mock decorator also has an accompanying mocker class. These classes are provided as pytest fixtures and are always available. To access them from your test function, just include the mocker class name as snake case (e.g. access FilesMock
mocker with files_mock
).
Mockers
ChatCompletionMock
EmbeddingsMock
FilesMock
AssistantsMock
ThreadsMock
MessagesMock
RunsMock
Example Access
from openai import OpenAI
import openai_responses
from openai_responses import FilesMock
@openai_responses.mock.files(failures=2)
def test_upload_files_with_retries(files_mock: FilesMock):
client = OpenAI(api_key="fakeKey", max_retries=2, timeout=0)
file = client.files.create(
file=open("examples/example.json", "rb"),
purpose="assistants",
)
assert file.filename == "example.json"
assert file.purpose == "assistants"
assert files_mock.create.route.calls.call_count == 3
[!TIP] Learn more about route access below
State
If you want to establish state prior to a test run you can pass a state store instance into the decorator.
from openai import OpenAI
from openai.types.beta.assistant import Assistant
import openai_responses
from openai_responses.state import StateStore
custom_state_store = StateStore()
asst = Assistant(id="asst_abc123"...) # create assistant
custom_state_store.beta.assistants.put(asst) # put assistant in state store
@openai_responses.mock.assistants(state_store=custom_state_store):
def test_retrieve_assistant():
client = OpenAI(api_key="fakeKey")
found = client.beta.assistants.retrieve("asst_abc123")
Async
Async works exactly the same as sync so you don't have to change anything other than just marking your test function as async.
import pytest
from openai import AsyncOpenAI
import openai_responses
@pytest.mark.asyncio
@openai_responses.mock.files()
async def test_async_upload_file():
client = AsyncOpenAI(api_key="fakeKey")
file = await client.files.create(
file=open("examples/example.json", "rb"),
purpose="assistants",
)
assert file.filename == "example.json"
assert file.purpose == "assistants"
Chaining
To mock more than one API endpoint, you can chain decorators as much as you'd like.
@openai_responses.mock.beta.threads()
@openai_responses.mock.beta.threads.runs()
def test_list_runs(threads_mock: ThreadsMock, runs_mock: RunsMock):
client = OpenAI(api_key="fakeKey")
thread = client.beta.threads.create()
for _ in range(20):
client.beta.threads.runs.create(thread.id, assistant_id="asst_abc123")
runs = client.beta.threads.runs.list(thread.id)
assert len(runs.data) == 20
assert threads_mock.create.route.calls.call_count == 1
assert runs_mock.create.route.calls.call_count == 20
assert runs_mock.list.route.calls.call_count == 1
Route Access
Each endpoint method provides access to the underlying RESPX route for things like call history assertion, call retrieval, and more.
import pytest
from openai import OpenAI, NotFoundError
import openai_responses
from openai_responses import AssistantsMock
@openai_responses.mock.beta.assistants()
def test_retrieve_assistant(assistants_mock: AssistantsMock):
client = OpenAI(api_key="fakeKey")
with pytest.raises(NotFoundError):
client.beta.assistants.retrieve("invalid_id")
asst = client.beta.assistants.create(model="gpt-4")
found = client.beta.assistants.retrieve(asst.id)
assert found.id == asst.id
assert assistants_mock.retrieve.route.calls.call_count == 2 # should have been called twice
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for openai_responses-0.1.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | c8e9ab8c2f4feb3722bc09f74f7d0a9f83844f335458fabe1c375af60daa61e2 |
|
MD5 | e99f5c2151f94adeec0b81374cf9d403 |
|
BLAKE2b-256 | 634ae8f97a15c43a7a2415c3aa419d32ce863ff6cadbc8d9b95a29c82295f6d1 |