87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
"""Tests for Phase 5.1 decompose logging and fallback behavior."""
|
|
import logging
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from app.models.decompose import SubQuestions
|
|
from app.services.llm_client import LLMClientError
|
|
from app.services.query_decomposer import QueryDecomposer
|
|
|
|
|
|
class MockFallbackLLMClient:
|
|
"""LLM client where both structured and legacy paths fail for logging."""
|
|
|
|
def __init__(self, complete_response="not json"):
|
|
self._complete_response = complete_response
|
|
self.complete_called = False
|
|
self.complete_structured_called = False
|
|
|
|
async def complete(self, prompt, temperature=0.7, step_name="LLM"):
|
|
self.complete_called = True
|
|
return self._complete_response
|
|
|
|
async def complete_structured(self, prompt, pydantic_model, step_name="LLM"):
|
|
self.complete_structured_called = True
|
|
raise LLMClientError("structured output failed")
|
|
|
|
|
|
async def test_warning_logged_when_both_paths_fail(mock_prompt_service):
|
|
"""When both structured and legacy paths fail, warning should be logged."""
|
|
llm = MockFallbackLLMClient(complete_response="completely invalid response !!!")
|
|
decomposer = QueryDecomposer(llm, prompt_service=mock_prompt_service)
|
|
|
|
with patch("app.services.query_decomposer.logger") as mock_logger:
|
|
questions, _ = await decomposer.decompose("Test question")
|
|
|
|
assert questions == []
|
|
|
|
warning_calls = [
|
|
str(call)
|
|
for call in mock_logger.warning.call_args_list
|
|
]
|
|
assert any("Structured decomposition failed" in msg for msg in warning_calls)
|
|
assert any("Legacy decompose JSON parse failed" in msg for msg in warning_calls)
|
|
assert any("completely invalid response" in msg for msg in warning_calls)
|
|
|
|
|
|
async def test_warning_logged_when_structured_fails_but_legacy_succeeds(mock_prompt_service):
|
|
"""When structured fails but legacy succeeds, info should be logged."""
|
|
llm = MockFallbackLLMClient(complete_response='["q1", "q2"]')
|
|
llm._complete_response = '["q1", "q2"]'
|
|
decomposer = QueryDecomposer(llm, prompt_service=mock_prompt_service)
|
|
|
|
with patch("app.services.query_decomposer.logger") as mock_logger:
|
|
questions, _ = await decomposer.decompose("Test question")
|
|
|
|
assert questions == ["q1", "q2"]
|
|
|
|
info_calls = [
|
|
str(call)
|
|
for call in mock_logger.info.call_args_list
|
|
]
|
|
assert any("Legacy decompose succeeded after structured output failure" in msg for msg in info_calls)
|
|
|
|
|
|
async def test_no_logging_when_structured_succeeds(mock_prompt_service):
|
|
"""When structured output succeeds, no warning should be logged."""
|
|
llm = MockFallbackLLMClient()
|
|
|
|
async def successful_structured(prompt, pydantic_model, step_name="LLM"):
|
|
llm.complete_structured_called = True
|
|
return SubQuestions(questions=["Q1"])
|
|
|
|
llm.complete_structured = successful_structured
|
|
decomposer = QueryDecomposer(llm, prompt_service=mock_prompt_service)
|
|
|
|
with patch("app.services.query_decomposer.logger") as mock_logger:
|
|
questions, _ = await decomposer.decompose("Test question")
|
|
|
|
assert questions == ["Q1"]
|
|
|
|
warning_calls = [
|
|
str(call)
|
|
for call in mock_logger.warning.call_args_list
|
|
]
|
|
assert not any("Structured decomposition failed" in msg for msg in warning_calls)
|