"""Verify Gemini thinking content is properly extracted and wrapped for parse_thinking_trace. The google-genai SDK separates thinking content from visible text by marking parts with thought=True. The SDK filters these out of resp.text, so thinking monologues don't render in the Discussion Hub. _send_gemini now calls _extract_gemini_thoughts and wraps the extracted text in ... tags so thinking_parser can extract a ThinkingSegment. """ from unittest.mock import MagicMock, patch from google.genai.types import Part, Content, Candidate, GenerateContentResponse from src.ai_client import _extract_gemini_thoughts from src.thinking_parser import parse_thinking_trace def test_extract_gemini_thoughts_returns_thinking_only() -> None: """The helper must return concatenated thought=True parts and ignore thought=False parts.""" resp = GenerateContentResponse( candidates=[Candidate(content=Content(parts=[ Part(text="step 1 reasoning", thought=True), Part(text="visible text 1", thought=False), Part(text="step 2 reasoning", thought=True), Part(text="visible text 2"), ]))] ) thoughts = _extract_gemini_thoughts(resp) assert thoughts == "step 1 reasoningstep 2 reasoning" def test_extract_gemini_thoughts_returns_empty_when_no_thoughts() -> None: """No thought parts => empty string (the wrap is conditional).""" resp = GenerateContentResponse( candidates=[Candidate(content=Content(parts=[Part(text="just visible")]))] ) assert _extract_gemini_thoughts(resp) == "" def test_extract_gemini_thoughts_handles_missing_attributes() -> None: """Defensive: must not crash on objects without expected attributes.""" fake = MagicMock() fake.candidates = [MagicMock()] fake.candidates[0].content.parts = [MagicMock(thought=True, text="thinking text")] assert _extract_gemini_thoughts(fake) == "thinking text" fake.candidates = [] assert _extract_gemini_thoughts(fake) == "" def test_gemini_thinking_segment_extractable_after_wrap() -> None: """End-to-end: the wrapped output must be parseable by thinking_parser.parse_thinking_trace.""" resp = GenerateContentResponse( candidates=[Candidate(content=Content(parts=[ Part(text="my reasoning chain", thought=True), Part(text="final answer"), ]))] ) thoughts = _extract_gemini_thoughts(resp) wrapped = f"\n{thoughts}\n\n\nfinal answer" segments, response = parse_thinking_trace(wrapped) assert len(segments) == 1 assert segments[0].content == "my reasoning chain" assert segments[0].marker == "thinking" assert response == "final answer" def test_extract_gemini_thoughts_handles_none_resp() -> None: """Defensive: must not crash on None response.""" assert _extract_gemini_thoughts(None) == "" if __name__ == "__main__": test_extract_gemini_thoughts_returns_thinking_only() test_extract_gemini_thoughts_returns_empty_when_no_thoughts() test_extract_gemini_thoughts_handles_missing_attributes() test_gemini_thinking_segment_extractable_after_wrap() test_extract_gemini_thoughts_handles_none_resp() print("All Gemini thinking format tests passed!")