ha-openai-compatible/tests/test_entity.py

"""Tests for the fork's chat log conversion fixes in entity.py."""

from unittest.mock import MagicMock

from homeassistant.components import conversation
from homeassistant.core import Context
from homeassistant.helpers import llm

from custom_components.openai_compatible.const import (
    CONF_WEB_SEARCH,
    CONF_WEB_SEARCH_COUNTRY,
    CONF_WEB_SEARCH_TIMEZONE,
    CONF_WEB_SEARCH_USER_LOCATION,
)
from custom_components.openai_compatible.entity import (
    _convert_content_to_param,
    _transform_stream,
)

from . import create_function_tool_call_item, create_message_item


async def _stream(events):
    for event in events:
        yield event


async def test_tool_call_without_arguments() -> None:
    """Test an empty argument string means a call without parameters.

    Some OpenAI-compatible endpoints send "" instead of "{}" for tools that
    take no parameters, e.g. the timer status intent.
    """
    events = create_function_tool_call_item(
        id="fc_1",
        arguments=[],
        call_id="call_1",
        name="HassTimerStatus",
        output_index=0,
    )

    deltas = [delta async for delta in _transform_stream(MagicMock(), _stream(events))]

    tool_calls = [call for delta in deltas for call in delta.get("tool_calls", [])]
    assert len(tool_calls) == 1
    assert tool_calls[0].tool_name == "HassTimerStatus"
    assert tool_calls[0].tool_args == {}


def test_web_search_not_replayed() -> None:
    """Test earlier web searches are left out of the history.

    The chat log keeps only a search's status, not its results. Replaying the
    bare search makes the model disown the answer it based on that search.
    """
    content = [
        conversation.UserContent(content="How did the match go?"),
        conversation.AssistantContent(
            agent_id="conversation.test",
            tool_calls=[
                llm.ToolInput(
                    id="ws_1",
                    tool_name="web_search_call",
                    tool_args={"action": {"type": "search", "query": "match"}},
                    external=True,
                )
            ],
        ),
        conversation.ToolResultContent(
            agent_id="conversation.test",
            tool_call_id="ws_1",
            tool_name="web_search_call",
            result=llm.ToolResult(data={"status": "completed"}),
        ),
        conversation.AssistantContent(
            agent_id="conversation.test",
            content="It ended 2 to 2.",
            tool_calls=[
                llm.ToolInput(id="call_1", tool_name="GetLiveContext", tool_args={})
            ],
        ),
        conversation.ToolResultContent(
            agent_id="conversation.test",
            tool_call_id="call_1",
            tool_name="GetLiveContext",
            result=llm.ToolResult(data={"success": True}),
        ),
    ]

    messages = _convert_content_to_param(content)

    assert [message["type"] for message in messages] == [
        "message",
        "message",
        "function_call",
        "function_call_output",
    ]
    assert messages[1]["content"] == "It ended 2 to 2."
    assert messages[2]["call_id"] == "call_1"


async def test_web_search_location_leaves_out_empty_fields(
    hass, mock_config_entry, mock_init_component, mock_create_stream
) -> None:
    """Test unknown location fields are left out of the web search tool.

    A failed location lookup stores no city or region; endpoints in front of
    Anthropic models reject empty strings there.
    """
    subentry = next(iter(mock_config_entry.subentries.values()))
    hass.config_entries.async_update_subentry(
        mock_config_entry,
        subentry,
        data={
            **subentry.data,
            CONF_WEB_SEARCH: True,
            CONF_WEB_SEARCH_USER_LOCATION: True,
            CONF_WEB_SEARCH_COUNTRY: "DE",
            CONF_WEB_SEARCH_TIMEZONE: "Europe/Berlin",
        },
    )
    await hass.config_entries.async_reload(mock_config_entry.entry_id)
    mock_create_stream.return_value = [
        create_message_item(id="msg_A", text="Hi", output_index=0)
    ]

    await conversation.async_converse(
        hass, "hello", None, Context(), agent_id="conversation.openai_conversation"
    )

    tools = mock_create_stream.mock_calls[0][2]["tools"]
    web_search = next(tool for tool in tools if tool["type"] == "web_search")
    assert web_search["user_location"] == {
        "type": "approximate",
        "country": "DE",
        "timezone": "Europe/Berlin",
    }