devplacepy/tests/unit/services/openai_gateway/responses.py

# retoor <retoor@molodetz.nl>

from devplacepy.services.openai_gateway.responses import (
    ChatToResponsesStream,
    ResponsesToChatStream,
    chat_completion_to_responses_object,
    chat_to_responses,
    responses_error_object,
    responses_failure_message,
    responses_object_to_chat_completion,
    responses_to_chat,
)


def _chat_body(**over):
    body = {
        "model": "m",
        "messages": [
            {"role": "system", "content": "Be terse."},
            {"role": "user", "content": [{"type": "text", "text": "hi"}]},
            {
                "role": "assistant",
                "content": None,
                "tool_calls": [
                    {
                        "id": "call_1",
                        "type": "function",
                        "function": {"name": "get_weather", "arguments": "{}"},
                    }
                ],
            },
            {"role": "tool", "tool_call_id": "call_1", "content": "sunny"},
        ],
        "tools": [
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "parameters": {"type": "object"},
                },
            }
        ],
        "tool_choice": "auto",
        "response_format": {
            "type": "json_schema",
            "json_schema": {"name": "r", "schema": {"type": "object"}},
        },
        "max_tokens": 10,
    }
    body.update(over)
    return body


def test_chat_to_responses_core_mapping():
    resp, warn = chat_to_responses(_chat_body())
    assert warn == []
    assert resp["model"] == "m"
    assert resp["instructions"] == "Be terse."
    assert [i.get("type") for i in resp["input"]] == [
        "message",
        "function_call",
        "function_call_output",
    ]
    assert resp["input"][0]["content"] == [
        {"type": "input_text", "text": "hi"}
    ]
    assert resp["input"][1]["call_id"] == "call_1"
    assert resp["input"][2]["output"] == "sunny"
    assert resp["tools"][0]["name"] == "get_weather"
    assert resp["tools"][0]["strict"] is False
    assert resp["text"]["format"]["type"] == "json_schema"
    assert resp["max_output_tokens"] == 10


def test_chat_to_responses_pins_strict_explicitly():
    resp, _ = chat_to_responses(
        {
            "model": "m",
            "messages": [{"role": "user", "content": "hi"}],
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": "f",
                        "strict": True,
                        "parameters": {
                            "type": "object",
                            "properties": {},
                            "additionalProperties": False,
                        },
                    },
                }
            ],
        }
    )
    assert resp["tools"][0]["strict"] is True


def test_chat_to_responses_drops_strict_the_schema_cannot_meet():
    def tool(parameters):
        return {
            "type": "function",
            "function": {"name": "f", "strict": True, "parameters": parameters},
        }

    nested_open = {
        "type": "object",
        "properties": {
            "words": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {"text": {"type": "string"}},
                    "required": ["text"],
                },
            }
        },
        "required": ["words"],
        "additionalProperties": False,
    }
    optional_property = {
        "type": "object",
        "properties": {"a": {"type": "string"}},
        "additionalProperties": False,
    }
    for parameters in (nested_open, optional_property, {"type": "object"}):
        resp, warn = chat_to_responses(
            {
                "model": "m",
                "messages": [{"role": "user", "content": "hi"}],
                "tools": [tool(parameters)],
            }
        )
        assert resp["tools"][0]["strict"] is False
        assert any("dropped strict" in w for w in warn)
    nested_open["properties"]["words"]["items"]["additionalProperties"] = False
    resp, warn = chat_to_responses(
        {
            "model": "m",
            "messages": [{"role": "user", "content": "hi"}],
            "tools": [tool(nested_open)],
        }
    )
    assert resp["tools"][0]["strict"] is True
    assert not any("dropped strict" in w for w in warn)


def test_chat_to_responses_warns_on_drops_and_unknowns():
    resp, warn = chat_to_responses(
        _chat_body(n=2, stop=["x"], seed=1, wat_field=True)
    )
    assert "n" not in resp
    assert "stop" not in resp
    assert "seed" not in resp
    assert any("field 'n'" in w for w in warn)
    assert any("field 'stop'" in w for w in warn)
    assert any("field 'seed'" in w for w in warn)
    assert any("unknown chat field 'wat_field'" in w for w in warn)


def test_chat_to_responses_drops_custom_tools_with_warning():
    resp, warn = chat_to_responses(
        {
            "model": "m",
            "messages": [{"role": "user", "content": "hi"}],
            "tools": [{"type": "custom", "custom": {"name": "x"}}],
        }
    )
    assert "tools" not in resp
    assert any("custom" in w for w in warn)


def test_chat_to_responses_named_tool_choice():
    resp, warn = chat_to_responses(
        {
            "model": "m",
            "messages": [{"role": "user", "content": "hi"}],
            "tool_choice": {
                "type": "function",
                "function": {"name": "get_weather"},
            },
        }
    )
    assert warn == []
    assert resp["tool_choice"] == {"type": "function", "name": "get_weather"}


def test_chat_to_responses_response_format_variants():
    resp, _ = chat_to_responses(
        {
            "model": "m",
            "messages": [{"role": "user", "content": "hi"}],
            "response_format": {"type": "json_object"},
        }
    )
    assert resp["text"] == {"format": {"type": "json_object"}}
    resp, _ = chat_to_responses(
        {
            "model": "m",
            "messages": [{"role": "user", "content": "hi"}],
            "response_format": {"type": "text"},
        }
    )
    assert resp["text"] == {"format": {"type": "text"}}


def test_chat_to_responses_reasoning_effort_and_image():
    resp, warn = chat_to_responses(
        {
            "model": "m",
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image_url",
                            "image_url": {
                                "url": "https://x/y.png",
                                "detail": "low",
                            },
                        }
                    ],
                }
            ],
            "reasoning_effort": "high",
        }
    )
    assert warn == []
    assert resp["reasoning"] == {"effort": "high"}
    assert resp["input"][0]["content"] == [
        {
            "type": "input_image",
            "image_url": "https://x/y.png",
            "detail": "low",
        }
    ]


def test_chat_to_responses_thinking_flags_map_to_reasoning():
    base = {"model": "m", "messages": [{"role": "user", "content": "hi"}]}
    resp, warn = chat_to_responses({**base, "think": False})
    assert warn == []
    assert resp["reasoning"] == {"effort": "minimal"}
    resp, warn = chat_to_responses(
        {**base, "thinking": {"type": "disabled"}}
    )
    assert warn == []
    assert resp["reasoning"] == {"effort": "minimal"}
    resp, warn = chat_to_responses({**base, "think": True})
    assert warn == []
    assert "reasoning" not in resp
    resp, warn = chat_to_responses(
        {**base, "reasoning": {"effort": "low", "max_tokens": 5}}
    )
    assert resp["reasoning"] == {"effort": "low"}
    assert any("reasoning keys" in w for w in warn)
    resp, warn = chat_to_responses(
        {**base, "reasoning_effort": "high", "think": False}
    )
    assert warn == []
    assert resp["reasoning"] == {"effort": "high"}


def test_chat_to_responses_audio_part_warns_and_drops():
    resp, warn = chat_to_responses(
        {
            "model": "m",
            "messages": [
                {
                    "role": "user",
                    "content": [{"type": "input_audio", "input_audio": {}}],
                }
            ],
        }
    )
    assert resp["input"][0]["content"] == []
    assert any("input_audio" in w for w in warn)


def test_responses_to_chat_round_trip():
    resp, _ = chat_to_responses(_chat_body())
    back, warn = responses_to_chat(resp)
    assert warn == []
    assert back["messages"][0] == {"role": "system", "content": "Be terse."}
    assert back["messages"][1]["role"] == "user"
    assert back["messages"][2]["tool_calls"][0]["id"] == "call_1"
    assert back["messages"][3] == {
        "role": "tool",
        "tool_call_id": "call_1",
        "content": "sunny",
    }
    assert back["tools"][0]["function"]["name"] == "get_weather"
    assert back["response_format"]["type"] == "json_schema"
    assert back["max_tokens"] == 10


def test_responses_to_chat_drops_namespace_and_state():
    back, warn = responses_to_chat(
        {
            "model": "m",
            "instructions": "sys",
            "input": [{"type": "message", "role": "user", "content": "hi"}],
            "tools": [{"type": "namespace", "name": "muse", "tools": []}],
            "store": False,
            "previous_response_id": "resp_1",
            "include": ["reasoning.encrypted_content"],
        }
    )
    assert back["messages"][0] == {"role": "system", "content": "sys"}
    assert "tools" not in back
    assert back["store"] is False
    assert any("namespace" in w for w in warn)
    assert any("previous_response_id" in w for w in warn)
    assert any("'include'" in w for w in warn)


def test_responses_to_chat_drops_reasoning_items():
    back, warn = responses_to_chat(
        {
            "model": "m",
            "input": [
                {"type": "reasoning", "id": "rs_1"},
                {"type": "message", "role": "user", "content": "hi"},
            ],
        }
    )
    assert [m["role"] for m in back["messages"]] == ["user"]
    assert any("reasoning" in w for w in warn)


def test_responses_to_chat_string_input_and_instructions_list():
    back, warn = responses_to_chat({"model": "m", "input": "hello"})
    assert warn == []
    assert back["messages"] == [{"role": "user", "content": "hello"}]
    back, warn = responses_to_chat(
        {
            "model": "m",
            "input": [{"type": "message", "role": "user", "content": "hi"}],
            "instructions": [
                {"type": "message", "role": "system", "content": "a"},
                {"type": "message", "role": "developer", "content": "b"},
            ],
        }
    )
    assert warn == []
    assert [m["role"] for m in back["messages"]] == [
        "system",
        "developer",
        "user",
    ]


def test_response_object_to_chat_completion():
    comp, warn = responses_object_to_chat_completion(
        {
            "id": "resp_1",
            "created_at": 7,
            "model": "m",
            "status": "completed",
            "output": [
                {
                    "type": "message",
                    "role": "assistant",
                    "content": [{"type": "output_text", "text": "ok"}],
                },
                {
                    "type": "function_call",
                    "call_id": "call_9",
                    "name": "f",
                    "arguments": "{}",
                },
            ],
            "usage": {"input_tokens": 3, "output_tokens": 1, "total_tokens": 4},
        }
    )
    assert warn == []
    assert comp["object"] == "chat.completion"
    assert comp["choices"][0]["message"]["content"] == "ok"
    assert comp["choices"][0]["message"]["tool_calls"][0]["id"] == "call_9"
    assert comp["choices"][0]["finish_reason"] == "tool_calls"
    assert comp["usage"] == {
        "prompt_tokens": 3,
        "completion_tokens": 1,
        "total_tokens": 4,
    }


def test_response_object_usage_details_pass_through():
    comp, _ = responses_object_to_chat_completion(
        {
            "id": "r",
            "output": [],
            "usage": {
                "input_tokens": 10,
                "output_tokens": 4,
                "total_tokens": 14,
                "input_tokens_details": {"cached_tokens": 6},
                "output_tokens_details": {"reasoning_tokens": 2},
            },
        }
    )
    assert comp["usage"]["prompt_tokens"] == 10
    assert comp["usage"]["prompt_tokens_details"] == {"cached_tokens": 6}
    assert comp["usage"]["completion_tokens_details"] == {"reasoning_tokens": 2}


def test_response_object_finish_reasons():
    comp, _ = responses_object_to_chat_completion(
        {
            "id": "r",
            "output": [],
            "incomplete_details": {"reason": "max_output_tokens"},
        }
    )
    assert comp["choices"][0]["finish_reason"] == "length"
    comp, _ = responses_object_to_chat_completion(
        {
            "id": "r",
            "output": [],
            "incomplete_details": {"reason": "content_filter"},
        }
    )
    assert comp["choices"][0]["finish_reason"] == "content_filter"
    comp, _ = responses_object_to_chat_completion(
        {
            "id": "r",
            "output": [
                {
                    "type": "message",
                    "role": "assistant",
                    "content": [{"type": "refusal", "refusal": "no"}],
                }
            ],
        }
    )
    assert comp["choices"][0]["message"]["refusal"] == "no"
    assert comp["choices"][0]["finish_reason"] == "stop"


def test_chat_completion_to_response_object_keeps_choice_zero():
    obj, warn = chat_completion_to_responses_object(
        {
            "id": "c",
            "created": 1,
            "model": "m",
            "choices": [
                {"message": {"role": "assistant", "content": "a"}},
                {"message": {"role": "assistant", "content": "b"}},
            ],
            "usage": {"prompt_tokens": 1, "completion_tokens": 1,
                      "total_tokens": 2},
        }
    )
    assert any("choice 0 of 2" in w for w in warn)
    assert obj["output"][0]["content"][0]["text"] == "a"
    assert obj["usage"]["input_tokens"] == 1


def test_responses_failure_and_error_shapes():
    assert responses_failure_message({"a": 1}) is None
    assert responses_failure_message("x") is None
    assert (
        responses_failure_message({"error": {"message": "bad", "type": "t"}})
        == "bad"
    )
    assert responses_failure_message(
        {"object": "response", "status": "failed",
         "incomplete_details": {"reason": "r"}}
    ) == "response failed: r"
    assert responses_error_object({"error": {"message": "m", "code": "c"}}) == {
        "error": {"message": "m", "code": "c"}
    }
    assert responses_error_object(
        {"error": {"message": "m", "type": "t", "code": "c", "param": "p"}}
    ) == {"error": {"message": "m", "type": "t", "code": "c", "param": "p"}}
    assert responses_error_object({"a": 1}) is None


def test_stream_responses_to_chat_text_and_done():
    stream = ResponsesToChatStream(model="m")
    chunks = stream.feed({"type": "response.output_text.delta", "delta": "Hi"})
    assert chunks[0]["choices"][0]["delta"] == {
        "role": "assistant",
        "content": "Hi",
    }
    chunks = stream.feed({"type": "response.output_text.delta", "delta": "!"})
    assert chunks[0]["choices"][0]["delta"] == {"content": "!"}
    done = stream.feed(
        {
            "type": "response.completed",
            "response": {
                "id": "r",
                "output": [],
                "usage": {"input_tokens": 1, "output_tokens": 1,
                          "total_tokens": 2},
            },
        }
    )
    assert done[0]["choices"][0]["finish_reason"] == "stop"
    assert "usage" not in done[0]
    assert done[1]["choices"] == []
    assert done[1]["usage"]["prompt_tokens"] == 1


def test_stream_responses_to_chat_tool_calls():
    stream = ResponsesToChatStream(model="m")
    chunks = stream.feed(
        {
            "type": "response.output_item.added",
            "item": {
                "type": "function_call",
                "call_id": "call_1",
                "name": "f",
            },
        }
    )
    call = chunks[0]["choices"][0]["delta"]["tool_calls"][0]
    assert call["id"] == "call_1"
    assert call["function"]["name"] == "f"
    chunks = stream.feed(
        {
            "type": "response.function_call_arguments.delta",
            "item_id": "call_1",
            "delta": '{"a":',
        }
    )
    call = chunks[0]["choices"][0]["delta"]["tool_calls"][0]
    assert call["index"] == 0
    assert call["function"] == {"arguments": '{"a":'}


def test_stream_responses_to_chat_error_and_unknown():
    stream = ResponsesToChatStream(model="m")
    assert stream.feed({"type": "response.in_progress"}) == []
    assert stream.feed("nope") == []
    out = stream.feed(
        {
            "type": "response.failed",
            "response": {
                "object": "response",
                "status": "failed",
                "error": {"message": "boom", "type": "server_error"},
            },
        }
    )
    assert out == [{"error": {"message": "boom", "type": "server_error"}}]
    assert stream.failed is not None


def test_stream_chat_to_responses_round_trip():
    stream = ChatToResponsesStream()
    events = stream.feed(
        {
            "model": "m",
            "choices": [
                {
                    "delta": {
                        "content": "Hi",
                        "tool_calls": [
                            {
                                "index": 0,
                                "id": "call_1",
                                "function": {"name": "f", "arguments": ""},
                            }
                        ],
                    },
                    "finish_reason": None,
                }
            ],
        }
    )
    assert [e["type"] for e in events] == [
        "response.output_item.added",
        "response.output_text.delta",
    ]
    events = stream.feed(
        {
            "model": "m",
            "choices": [
                {
                    "delta": {
                        "tool_calls": [
                            {"index": 0, "function": {"arguments": "{}"}}
                        ]
                    },
                    "finish_reason": "stop",
                }
            ],
        }
    )
    assert [e["type"] for e in events] == [
        "response.function_call_arguments.delta",
        "response.completed",
    ]
    assert events[0]["item_id"] == "call_1"
    assert events[1]["response"]["output"][0]["content"][0]["text"] == "Hi"
    assert stream.warnings == []