photools3/test/test_llm_session_mocked.py

from unittest.mock import MagicMock, patch

import openai
import pytest

from photools3.llm_session import LlmSession, LlmSessionCaptionsAndKeywords
from photools3.models import LlmResponseCaptionsAndKeywords


def test_llm_session_mocked(test_photo):
    # We want to mock OpenAI client's chat.completions.create
    with patch("photools3.llm_session.OpenAI") as mocked_openai:
        # Setup the mock response structure
        mock_client = MagicMock()
        mocked_openai.return_value = mock_client

        # Mock the models list to return a proper object with an ID string
        # This prevents Pydantic ValidationError when LlmSession tries to use model_id
        mock_model = MagicMock()
        mock_model.id = "gpt-4o"
        mock_client.models.list.return_value.data = [mock_model]

        # Create a response object: response.choices[0].message.content
        mock_response = MagicMock()
        mock_choice = MagicMock()
        mock_message = MagicMock()

        mock_message.content = '{"text": "some text"}'
        mock_choice.message = mock_message
        mock_response.choices = [mock_choice]

        # Configure the client to return our mock response
        mock_client.chat.completions.create.return_value = mock_response

        # Create a new LlmSession instance that will use the mocked OpenAI client
        mocked_session = LlmSession(system_prompt="You are a helper.")

        # Run the test
        result = mocked_session.call(
            prompt="keywords for the attached picture", photos=[test_photo]
        )

        # Verify the results
        assert result.text == "some text"

        # Verify that the mock was enough
        assert mock_client.chat.completions.create.called


def test_llm_session_captions_and_keywords_mocked(test_photo):
    # We want to mock OpenAI client's chat.completions.create
    with patch("photools3.llm_session.OpenAI") as mocked_openai:
        # Setup the mock response structure
        mock_client = MagicMock()
        mocked_openai.return_value = mock_client

        # Mock the models list to return a proper object with an ID string
        # This prevents Pydantic ValidationError when LlmSession tries to use model_id
        mock_model = MagicMock()
        mock_model.id = "gpt-4o"
        mock_client.models.list.return_value.data = [mock_model]

        # Create a response object: response.choices[0].message.content
        mock_response = MagicMock()
        mock_choice = MagicMock()
        mock_message = MagicMock()

        mock_message.content = (
            '{"caption": "A mocked caption", "tags": ["architektur"]}'
        )
        mock_choice.message = mock_message
        mock_response.choices = [mock_choice]

        # Configure the client to return our mock response
        mock_client.chat.completions.create.return_value = mock_response

        # Create a new LlmSessionCaptionsAndKeywords instance that will use the mocked OpenAI client
        from photools3.llm_session import LlmSessionCaptionsAndKeywords

        mocked_session = LlmSessionCaptionsAndKeywords()

        # Run the test
        result = mocked_session.call(photos=[test_photo])

        # Verify the results
        assert result.caption == "A mocked caption"
        assert "architektur" in result.tags

        # Verify that the mock was called
        assert mock_client.chat.completions.create.called


def test_llm_session_retries_from_original_request_without_accumulating_messages():
    with patch("photools3.llm_session.OpenAI") as mocked_openai:
        mock_client = MagicMock()
        mocked_openai.return_value = mock_client
        mock_model = MagicMock()
        mock_model.id = "gpt-4o"
        mock_client.models.list.return_value.data = [mock_model]

        def response(content):
            message = MagicMock()
            message.content = content
            choice = MagicMock()
            choice.message = message
            result = MagicMock()
            result.choices = [choice]
            return result

        mock_client.chat.completions.create.side_effect = [
            response('{"wrong": "first"}'),
            response('{"wrong": "second"}'),
            response('{"text": "correct"}'),
        ]

        session = LlmSession(system_prompt="You are a helper.")
        result = session.call(photos=[], prompt="Return JSON")

    assert result.text == "correct"
    requests = mock_client.chat.completions.create.call_args_list
    messages_per_attempt = [call.kwargs["messages"] for call in requests]
    assert [len(messages) for messages in messages_per_attempt] == [2, 3, 3]
    final_correction = messages_per_attempt[2][-1]["content"]
    assert "Field required" in final_correction
    assert '"wrong"' not in final_correction


def test_llm_session_uses_configured_retry_count(monkeypatch):
    monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 1)

    with patch("photools3.llm_session.OpenAI") as mocked_openai:
        mock_client = MagicMock()
        mocked_openai.return_value = mock_client
        mock_model = MagicMock()
        mock_model.id = "gpt-4o"
        mock_client.models.list.return_value.data = [mock_model]

        def response(content):
            message = MagicMock()
            message.content = content
            choice = MagicMock()
            choice.message = message
            result = MagicMock()
            result.choices = [choice]
            return result

        mock_client.chat.completions.create.side_effect = [
            response('{"wrong": "first"}'),
            response('{"wrong": "second"}'),
            response('{"text": "would be a third call"}'),
        ]
        session = LlmSession(system_prompt="You are a helper.")

        with (
            patch.object(session, "flush_model") as flush_model,
            pytest.raises(ValueError, match="after 1 retries"),
        ):
            session.call(photos=[])

    assert mock_client.chat.completions.create.call_count == 2
    flush_model.assert_called_once()


def test_llm_session_reloads_after_timeout_and_retries(monkeypatch):
    monkeypatch.setattr(
        "photools3.llm_session.settings.llm_request_timeout_seconds", 10
    )
    monkeypatch.setattr("photools3.llm_session.settings.llm_max_completion_tokens", 768)
    monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 1)

    with patch("photools3.llm_session.OpenAI") as mocked_openai:
        mock_client = MagicMock()
        mocked_openai.return_value = mock_client
        mock_model = MagicMock()
        mock_model.id = "gpt-4o"
        mock_client.models.list.return_value.data = [mock_model]

        response_message = MagicMock(content='{"text": "recovered"}')
        response_choice = MagicMock(message=response_message)
        success_response = MagicMock(choices=[response_choice])
        mock_client.chat.completions.create.side_effect = [
            openai.APITimeoutError(request=MagicMock()),
            success_response,
        ]

        session = LlmSession(system_prompt="You are a helper.")
        with patch.object(session, "flush_model") as flush_model:
            result = session.call(photos=[])

    assert result.text == "recovered"
    flush_model.assert_called_once()
    calls = mock_client.chat.completions.create.call_args_list
    assert len(calls) == 2
    assert all(call.kwargs["timeout"] == 10 for call in calls)
    assert all(call.kwargs["max_tokens"] == 768 for call in calls)
    mocked_openai.assert_called_once_with(
        base_url=session.base_url,
        api_key=session.api_key,
        max_retries=0,
    )


@pytest.mark.parametrize("failure_kind", ["invalid_response", "timeout"])
def test_llm_session_reloads_when_retry_limit_is_exhausted(monkeypatch, failure_kind):
    monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 0)

    with patch("photools3.llm_session.OpenAI") as mocked_openai:
        mock_client = MagicMock()
        mocked_openai.return_value = mock_client
        mock_model = MagicMock()
        mock_model.id = "gpt-4o"
        mock_client.models.list.return_value.data = [mock_model]

        if failure_kind == "timeout":
            mock_client.chat.completions.create.side_effect = openai.APITimeoutError(
                request=MagicMock()
            )
        else:
            response_message = MagicMock(content='{"wrong": "shape"}')
            response_choice = MagicMock(message=response_message)
            mock_client.chat.completions.create.return_value = MagicMock(
                choices=[response_choice]
            )

        session = LlmSession(system_prompt="You are a helper.")
        with (
            patch.object(session, "flush_model") as flush_model,
            pytest.raises(ValueError),
        ):
            session.call(photos=[])

    flush_model.assert_called_once()
    assert mock_client.chat.completions.create.call_count == 1


def test_invalid_keywords_get_one_correction_then_invalid_tags_are_dropped(monkeypatch):
    monkeypatch.setattr("photools3.llm_session.settings.llm_max_keywords", 3)
    monkeypatch.setattr("photools3.llm_session.settings.llm_max_retries", 5)

    with patch("photools3.llm_session.OpenAI") as mocked_openai:
        mock_client = MagicMock()
        mocked_openai.return_value = mock_client
        mock_model = MagicMock()
        mock_model.id = "gpt-4o"
        mock_client.models.list.return_value.data = [mock_model]

        def response(content):
            message = MagicMock(content=content)
            choice = MagicMock(message=message)
            return MagicMock(choices=[choice])

        mock_client.chat.completions.create.side_effect = [
            response(
                '{"caption": "A caption", "tags": ["not_allowed", "architektur"]}'
            ),
            response(
                '{"caption": "A caption", '
                '"tags": ["still_invalid", "architektur", "personen"]}'
            ),
        ]
        session = LlmSessionCaptionsAndKeywords()
        with patch.object(session, "flush_model") as flush_model:
            result = session.call(photos=[])

    assert result.tags == ["architektur", "personen"]
    assert mock_client.chat.completions.create.call_count == 2
    flush_model.assert_not_called()
    correction = mock_client.chat.completions.create.call_args_list[1].kwargs[
        "messages"
    ][-1]["content"]
    assert "not_allowed" in correction
    assert '"caption"' not in correction


def test_excess_keywords_are_truncated_to_configured_limit(monkeypatch):
    monkeypatch.setattr("photools3.llm_session.settings.llm_max_keywords", 2)

    with patch("photools3.llm_session.OpenAI") as mocked_openai:
        mock_client = MagicMock()
        mocked_openai.return_value = mock_client
        mock_model = MagicMock()
        mock_model.id = "gpt-4o"
        mock_client.models.list.return_value.data = [mock_model]
        message = MagicMock(
            content=(
                '{"caption": "A caption", '
                '"tags": ["architektur", "personen", "landschaft"]}'
            )
        )
        mock_client.chat.completions.create.return_value = MagicMock(
            choices=[MagicMock(message=message)]
        )

        result = LlmSessionCaptionsAndKeywords().call(photos=[])

    assert result.tags == ["architektur", "personen"]
    assert mock_client.chat.completions.create.call_count == 1


def test_keyword_response_schema_includes_configured_max_items():
    schema = LlmResponseCaptionsAndKeywords.model_json_schema()

    assert schema["properties"]["tags"]["maxItems"] == 15