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_small_talk_returns_deterministic_fast_path_answer() -> None:
    gemini = MagicMock()
    gemini.is_configured = True
    gemini.fast_answer = AsyncMock(return_value=("unused", None))

    node = make_gemini_fast_node(gemini)
    state = {
        "query": "Hi",
        "is_followup": False,
        "conversation_history": [],
    }

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

    assert "Darshan's portfolio assistant" in result["answer"]
    assert result["path"] == "gemini_fast"
    gemini.fast_answer.assert_not_awaited()


@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_non_small_talk_routes_to_rag() -> None:
    gemini = MagicMock()
    gemini.is_configured = True
    gemini.fast_answer = AsyncMock(return_value=("unused", 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 "answer" not in result
    assert result["expanded_queries"] == ["What tech stack does he use?"]
    gemini.fast_answer.assert_not_awaited()