openai-responses 0.1.1

Automatically mock OpenAI requests

openai-responses-python

Automatically mock OpenAI requests

• openai-responses-python
  • Installation
  • Quick start
  • Endpoint Support
  • Mocks
    • Chat Completions
    • Embeddings
    • Files
    • Assistants
    • Threads
    • Messages
    • Runs
      • Sequences
      • Resource validation
  • Mocker Classes
    • Example Access
  • State
  • Async
  • Chaining
  • Route Access

[!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

Available on PyPi

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 to 0.0.
• failures - number of failures to simulate. Defaults to 0.

Some decorators will have additional arguments which are documented below.

[!TIP] See examples for more

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 to False.

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 to False.
• validate_assistant_exists: Whether to check the state store to see if the assistant exists. Defaults to False.

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
• RunStepsMock

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