test_title_generator.py
1 """Tests for agent.title_generator — auto-generated session titles.""" 2 3 import threading 4 from unittest.mock import MagicMock, patch 5 6 import pytest 7 8 from agent.title_generator import ( 9 generate_title, 10 auto_title_session, 11 maybe_auto_title, 12 ) 13 14 15 class TestGenerateTitle: 16 """Unit tests for generate_title().""" 17 18 def test_returns_title_on_success(self): 19 mock_response = MagicMock() 20 mock_response.choices = [MagicMock()] 21 mock_response.choices[0].message.content = "Debugging Python Import Errors" 22 23 with patch("agent.title_generator.call_llm", return_value=mock_response): 24 title = generate_title("help me fix this import", "Sure, let me check...") 25 assert title == "Debugging Python Import Errors" 26 27 def test_strips_quotes(self): 28 mock_response = MagicMock() 29 mock_response.choices = [MagicMock()] 30 mock_response.choices[0].message.content = '"Setting Up Docker Environment"' 31 32 with patch("agent.title_generator.call_llm", return_value=mock_response): 33 title = generate_title("how do I set up docker", "First install...") 34 assert title == "Setting Up Docker Environment" 35 36 def test_strips_title_prefix(self): 37 mock_response = MagicMock() 38 mock_response.choices = [MagicMock()] 39 mock_response.choices[0].message.content = "Title: Kubernetes Pod Debugging" 40 41 with patch("agent.title_generator.call_llm", return_value=mock_response): 42 title = generate_title("my pod keeps crashing", "Let me look...") 43 assert title == "Kubernetes Pod Debugging" 44 45 def test_truncates_long_titles(self): 46 mock_response = MagicMock() 47 mock_response.choices = [MagicMock()] 48 mock_response.choices[0].message.content = "A" * 100 49 50 with patch("agent.title_generator.call_llm", return_value=mock_response): 51 title = generate_title("question", "answer") 52 assert len(title) == 80 53 assert title.endswith("...") 54 55 def test_returns_none_on_empty_response(self): 56 mock_response = MagicMock() 57 mock_response.choices = [MagicMock()] 58 mock_response.choices[0].message.content = "" 59 60 with patch("agent.title_generator.call_llm", return_value=mock_response): 61 assert generate_title("question", "answer") is None 62 63 def test_returns_none_on_exception(self): 64 with patch("agent.title_generator.call_llm", side_effect=RuntimeError("no provider")): 65 assert generate_title("question", "answer") is None 66 67 def test_invokes_failure_callback_on_exception(self): 68 """failure_callback must fire so the user sees a warning (issue #15775).""" 69 captured = [] 70 71 def _cb(task, exc): 72 captured.append((task, exc)) 73 74 exc = RuntimeError("openrouter 402: credits exhausted") 75 with patch("agent.title_generator.call_llm", side_effect=exc): 76 result = generate_title("question", "answer", failure_callback=_cb) 77 78 assert result is None 79 assert len(captured) == 1 80 assert captured[0][0] == "title generation" 81 assert captured[0][1] is exc 82 83 def test_failure_callback_errors_are_swallowed(self): 84 """A broken callback must not crash title generation.""" 85 86 def _bad_cb(task, exc): 87 raise ValueError("callback bug") 88 89 with patch("agent.title_generator.call_llm", side_effect=RuntimeError("nope")): 90 # Should return None without re-raising the callback error 91 assert generate_title("q", "a", failure_callback=_bad_cb) is None 92 93 def test_no_callback_matches_legacy_behavior(self): 94 """Omitting failure_callback preserves the silent-None return.""" 95 with patch("agent.title_generator.call_llm", side_effect=RuntimeError("nope")): 96 assert generate_title("q", "a") is None 97 98 def test_truncates_long_messages(self): 99 """Long user/assistant messages should be truncated in the LLM request.""" 100 captured_kwargs = {} 101 102 def mock_call_llm(**kwargs): 103 captured_kwargs.update(kwargs) 104 resp = MagicMock() 105 resp.choices = [MagicMock()] 106 resp.choices[0].message.content = "Short Title" 107 return resp 108 109 with patch("agent.title_generator.call_llm", side_effect=mock_call_llm): 110 generate_title("x" * 1000, "y" * 1000) 111 112 # The user content in the messages should be truncated 113 user_content = captured_kwargs["messages"][1]["content"] 114 assert len(user_content) < 1100 # 500 + 500 + formatting 115 116 117 class TestAutoTitleSession: 118 """Tests for auto_title_session() — the sync worker function.""" 119 120 def test_skips_if_no_session_db(self): 121 auto_title_session(None, "sess-1", "hi", "hello") # should not crash 122 123 def test_skips_if_title_exists(self): 124 db = MagicMock() 125 db.get_session_title.return_value = "Existing Title" 126 127 with patch("agent.title_generator.generate_title") as gen: 128 auto_title_session(db, "sess-1", "hi", "hello") 129 gen.assert_not_called() 130 131 def test_generates_and_sets_title(self): 132 db = MagicMock() 133 db.get_session_title.return_value = None 134 135 with patch("agent.title_generator.generate_title", return_value="New Title"): 136 auto_title_session(db, "sess-1", "hi", "hello") 137 db.set_session_title.assert_called_once_with("sess-1", "New Title") 138 139 def test_skips_if_generation_fails(self): 140 db = MagicMock() 141 db.get_session_title.return_value = None 142 143 with patch("agent.title_generator.generate_title", return_value=None): 144 auto_title_session(db, "sess-1", "hi", "hello") 145 db.set_session_title.assert_not_called() 146 147 148 class TestMaybeAutoTitle: 149 """Tests for maybe_auto_title() — the fire-and-forget entry point.""" 150 151 def test_skips_if_not_first_exchange(self): 152 """Should not fire for conversations with more than 2 user messages.""" 153 db = MagicMock() 154 history = [ 155 {"role": "user", "content": "first"}, 156 {"role": "assistant", "content": "response 1"}, 157 {"role": "user", "content": "second"}, 158 {"role": "assistant", "content": "response 2"}, 159 {"role": "user", "content": "third"}, 160 {"role": "assistant", "content": "response 3"}, 161 ] 162 163 with patch("agent.title_generator.auto_title_session") as mock_auto: 164 maybe_auto_title(db, "sess-1", "third", "response 3", history) 165 # Wait briefly for any thread to start 166 import time 167 time.sleep(0.1) 168 mock_auto.assert_not_called() 169 170 def test_fires_on_first_exchange(self): 171 """Should fire a background thread for the first exchange.""" 172 db = MagicMock() 173 db.get_session_title.return_value = None 174 history = [ 175 {"role": "user", "content": "hello"}, 176 {"role": "assistant", "content": "hi there"}, 177 ] 178 179 with patch("agent.title_generator.auto_title_session") as mock_auto: 180 maybe_auto_title(db, "sess-1", "hello", "hi there", history) 181 # Wait for the daemon thread to complete 182 import time 183 time.sleep(0.3) 184 mock_auto.assert_called_once_with( 185 db, "sess-1", "hello", "hi there", failure_callback=None, main_runtime=None 186 ) 187 188 def test_forwards_failure_callback_to_worker(self): 189 """maybe_auto_title must forward failure_callback into the thread.""" 190 db = MagicMock() 191 db.get_session_title.return_value = None 192 history = [ 193 {"role": "user", "content": "hello"}, 194 {"role": "assistant", "content": "hi there"}, 195 ] 196 197 def _cb(task, exc): 198 pass 199 200 with patch("agent.title_generator.auto_title_session") as mock_auto: 201 maybe_auto_title(db, "sess-1", "hello", "hi there", history, failure_callback=_cb) 202 import time 203 time.sleep(0.3) 204 mock_auto.assert_called_once_with( 205 db, "sess-1", "hello", "hi there", failure_callback=_cb, main_runtime=None 206 ) 207 208 def test_skips_if_no_response(self): 209 db = MagicMock() 210 maybe_auto_title(db, "sess-1", "hello", "", []) # empty response 211 212 def test_skips_if_no_session_db(self): 213 maybe_auto_title(None, "sess-1", "hello", "response", []) # no db