tests/test_generate_not_found_fallback.py

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from app.pipeline.nodes.gemini_fast import make_gemini_fast_node

_WRITER_PATCH = "app.pipeline.nodes.gemini_fast.get_stream_writer"


@pytest.mark.asyncio
async def test_non_portfolio_conversational_query_uses_gemini_answer() -> None:
    gemini = MagicMock()
    gemini.is_configured = True
    gemini.fast_answer = AsyncMock(return_value=("Hey there!", None))

    node = make_gemini_fast_node(gemini)
    state = {
        "query": "I said hello.",
        "is_followup": False,
        "conversation_history": [],
    }

    with patch(_WRITER_PATCH, return_value=MagicMock()):
        result = await node(state)

    assert result["answer"] == "Hey there!"
    assert result["path"] == "gemini_fast"
    gemini.fast_answer.assert_awaited_once()


@pytest.mark.asyncio
async def test_portfolio_specific_query_forces_rag() -> None:
    gemini = MagicMock()
    gemini.is_configured = True
    gemini.fast_answer = AsyncMock(return_value=("unused", None))

    node = make_gemini_fast_node(gemini)
    state = {
        "query": "How does TextOps work?",
        "is_followup": False,
        "conversation_history": [],
    }

    with patch(_WRITER_PATCH, return_value=MagicMock()):
        result = await node(state)

    assert "answer" not in result
    assert result["expanded_queries"] == ["How does TextOps work?"]
    gemini.fast_answer.assert_not_awaited()

    @pytest.mark.asyncio
    async def test_broad_portfolio_intent_can_use_gemini_fast_path() -> None:
        gemini = MagicMock()
        gemini.is_configured = True
        gemini.fast_answer = AsyncMock(return_value=("He uses a broad stack.", None))

        node = make_gemini_fast_node(gemini)
        state = {
            "query": "What tech stack does he use?",
            "is_followup": False,
            "conversation_history": [],
        }

        with patch(_WRITER_PATCH, return_value=MagicMock()):
            result = await node(state)

        assert result["answer"] == "He uses a broad stack."
        assert result["path"] == "gemini_fast"
        gemini.fast_answer.assert_awaited_once()