import pytest import time from unittest.mock import Mock, patch, MagicMock import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent / "src")) from query_rewriter import QueryRewriter, get_query_rewriter, reset_query_rewriter @pytest.fixture def rewriter(): reset_query_rewriter() return QueryRewriter() @pytest.fixture def sample_history(): return [ {"role": "user", "content": "Atlas nedir?"}, {"role": "assistant", "content": "Atlas bir ERP sistemidir."} ] class TestFollowUpDetection: def test_detects_pronoun_bu(self, rewriter, sample_history): assert rewriter.is_follow_up("Bu nasil calisir?", sample_history) is True def test_detects_pronoun_bunun(self, rewriter, sample_history): assert rewriter.is_follow_up("Bunun ozellikleri neler?", sample_history) is True def test_detects_pronoun_o(self, rewriter, sample_history): assert rewriter.is_follow_up("Onu nasil kullanabilirim?", sample_history) is True def test_detects_boyle(self, rewriter, sample_history): assert rewriter.is_follow_up("Boyle bir sey var mi?", sample_history) is True def test_detects_peki_prefix(self, rewriter, sample_history): assert rewriter.is_follow_up("Peki fiyati ne?", sample_history) is True def test_standalone_not_follow_up(self, rewriter, sample_history): assert rewriter.is_follow_up("Atlas ERP sisteminin fiyati nedir?", sample_history) is False def test_no_history_not_follow_up(self, rewriter): assert rewriter.is_follow_up("Bu nedir?", []) is False def test_short_query_with_question_word(self, rewriter, sample_history): assert rewriter.is_follow_up("Ne zaman", sample_history) is True class TestHistoryFormatting: def test_format_empty_history(self, rewriter): assert rewriter.format_history([]) == "" def test_format_single_turn(self, rewriter, sample_history): formatted = rewriter.format_history(sample_history) assert "Kullanici: Atlas nedir?" in formatted assert "Asistan: Atlas bir ERP sistemidir." in formatted def test_format_respects_max_turns(self, rewriter): long_history = [] for i in range(10): long_history.append({"role": "user", "content": f"Soru {i}"}) long_history.append({"role": "assistant", "content": f"Cevap {i}"}) formatted = rewriter.format_history(long_history) assert "Soru 0" not in formatted assert "Soru 9" in formatted class TestGetLastTopic: def test_gets_last_user_message(self, rewriter, sample_history): topic = rewriter.get_last_topic(sample_history) assert topic == "Atlas nedir?" def test_empty_history_returns_none(self, rewriter): assert rewriter.get_last_topic([]) is None class TestSingleton: def test_singleton_returns_same_instance(self): reset_query_rewriter() r1 = get_query_rewriter() r2 = get_query_rewriter() assert r1 is r2 def test_reset_clears_instance(self): r1 = get_query_rewriter() reset_query_rewriter() r2 = get_query_rewriter() assert r1 is not r2 class TestLLMRewriting: def test_rewrite_returns_original_if_not_follow_up(self, rewriter, sample_history): query = "Atlas ERP sisteminin fiyati nedir?" result, debug = rewriter.rewrite(query, sample_history) assert result == query assert debug["is_follow_up"] is False assert debug["method"] == "none" def test_rewrite_returns_debug_info(self, rewriter, sample_history): query = "Atlas ERP sisteminin fiyati nedir?" result, debug = rewriter.rewrite(query, sample_history) assert "original_query" in debug assert "rewritten_query" in debug assert "is_follow_up" in debug assert "method" in debug assert "rewrite_time_ms" in debug @patch.object(QueryRewriter, '_llm_rewrite') def test_rewrite_calls_llm_for_follow_up(self, mock_llm, rewriter, sample_history): mock_llm.return_value = "Atlas ERP sisteminin ozellikleri nelerdir?" query = "Bunun ozellikleri neler?" result, debug = rewriter.rewrite(query, sample_history) assert mock_llm.called assert debug["is_follow_up"] is True assert debug["method"] == "llm" @patch.object(QueryRewriter, '_llm_rewrite') def test_rewrite_uses_cache_on_repeat(self, mock_llm, rewriter, sample_history): mock_llm.return_value = "Atlas ERP sisteminin ozellikleri nelerdir?" query = "Bunun ozellikleri neler?" result1, debug1 = rewriter.rewrite(query, sample_history) result2, debug2 = rewriter.rewrite(query, sample_history) assert mock_llm.call_count == 1 assert debug2["method"] == "cache" assert result1 == result2 @patch.object(QueryRewriter, '_llm_rewrite') def test_rewrite_handles_llm_error(self, mock_llm, rewriter, sample_history): mock_llm.side_effect = Exception("API Error") query = "Bunun ozellikleri neler?" result, debug = rewriter.rewrite(query, sample_history) assert debug["method"] == "rule" assert "llm_error" in debug assert "Atlas" in result class TestResponseCleaning: def test_clean_response_removes_prefix(self, rewriter): assert rewriter._clean_response("Soru: Atlas nedir?") == "Atlas nedir?" def test_clean_response_takes_first_line(self, rewriter): response = "Atlas nedir?\nBu bir aciklama" assert rewriter._clean_response(response) == "Atlas nedir?" def test_clean_response_strips_quotes(self, rewriter): assert rewriter._clean_response('"Atlas nedir?"') == "Atlas nedir?" class TestCacheKey: def test_cache_key_is_deterministic(self, rewriter, sample_history): key = rewriter._get_cache_key("test query", sample_history) assert len(key) == 32 def test_different_queries_different_keys(self, rewriter, sample_history): key1 = rewriter._get_cache_key("query 1", sample_history) key2 = rewriter._get_cache_key("query 2", sample_history) assert key1 != key2 class TestTopicExtraction: def test_extract_topic_from_simple_question(self, rewriter, sample_history): topic = rewriter._extract_topic(sample_history) assert topic is not None assert "Atlas" in topic def test_extract_topic_removes_stop_words(self, rewriter): history = [{"role": "user", "content": "Atlas ERP nedir?"}] topic = rewriter._extract_topic(history) assert "nedir" not in topic.lower() def test_extract_topic_empty_history(self, rewriter): assert rewriter._extract_topic([]) is None class TestRuleBasedRewrite: def test_replaces_bu_with_topic(self, rewriter, sample_history): result = rewriter._rule_based_rewrite("Bu nasil calisir?", sample_history) assert "Atlas" in result assert "bu" not in result.lower().split()[0] def test_replaces_bunun_with_topic(self, rewriter, sample_history): result = rewriter._rule_based_rewrite("Bunun ozellikleri neler?", sample_history) assert "Atlas" in result def test_no_replacement_without_pronoun(self, rewriter, sample_history): query = "Atlas fiyati nedir?" result = rewriter._rule_based_rewrite(query, sample_history) assert result == query def test_no_replacement_without_history(self, rewriter): query = "Bu nasil calisir?" result = rewriter._rule_based_rewrite(query, []) assert result == query class TestAppendContextFallback: def test_appends_topic_to_query(self, rewriter, sample_history): result = rewriter._append_context_fallback("Fiyati nedir?", sample_history) assert "hakkinda" in result assert "Atlas" in result def test_returns_original_without_history(self, rewriter): query = "Fiyati nedir?" result = rewriter._append_context_fallback(query, []) assert result == query class TestFallbackChain: @patch.object(QueryRewriter, '_llm_rewrite') def test_uses_rule_when_llm_fails(self, mock_llm, rewriter, sample_history): mock_llm.side_effect = Exception("API Error") query = "Bu nasil calisir?" result, debug = rewriter.rewrite(query, sample_history) assert debug["method"] == "rule" assert "Atlas" in result @patch.object(QueryRewriter, '_llm_rewrite') @patch.object(QueryRewriter, '_rule_based_rewrite') def test_uses_append_when_rule_fails(self, mock_rule, mock_llm, rewriter, sample_history): mock_llm.side_effect = Exception("API Error") mock_rule.side_effect = Exception("Rule Error") query = "Bunun ozellikleri neler?" result, debug = rewriter.rewrite(query, sample_history) assert debug["method"] == "append" assert "hakkinda" in result @patch.object(QueryRewriter, '_llm_rewrite') @patch.object(QueryRewriter, '_rule_based_rewrite') @patch.object(QueryRewriter, '_append_context_fallback') def test_returns_original_when_all_fail(self, mock_append, mock_rule, mock_llm, rewriter, sample_history): mock_llm.side_effect = Exception("API Error") mock_rule.side_effect = Exception("Rule Error") mock_append.side_effect = Exception("Append Error") query = "Bu nasil?" result, debug = rewriter.rewrite(query, sample_history) assert debug["method"] == "fallback_failed" assert result == query class TestCacheTTL: def test_cache_returns_value_within_ttl(self, rewriter, sample_history): rewriter._add_to_cache("test_key", "cached_value") assert rewriter._get_from_cache("test_key") == "cached_value" def test_cache_expires_after_ttl(self, rewriter, sample_history): rewriter._cache_duration = 1 rewriter._add_to_cache("test_key", "cached_value") time.sleep(1.1) assert rewriter._get_from_cache("test_key") is None def test_expired_entry_removed_from_both_dicts(self, rewriter): rewriter._cache_duration = 0 rewriter._add_to_cache("expire_key", "value") time.sleep(0.05) rewriter._get_from_cache("expire_key") assert "expire_key" not in rewriter._cache assert "expire_key" not in rewriter._cache_ttl def test_cache_miss_returns_none(self, rewriter): assert rewriter._get_from_cache("nonexistent") is None def test_ttl_timestamp_recorded(self, rewriter): before = time.time() rewriter._add_to_cache("ts_key", "value") after = time.time() assert before <= rewriter._cache_ttl["ts_key"] <= after class TestCacheLRUEviction: def test_evicts_oldest_when_full(self): rewriter = QueryRewriter(cache_duration_seconds=3600) rewriter._max_cache_size = 3 rewriter._add_to_cache("key1", "val1") time.sleep(0.01) rewriter._add_to_cache("key2", "val2") time.sleep(0.01) rewriter._add_to_cache("key3", "val3") time.sleep(0.01) rewriter._add_to_cache("key4", "val4") assert "key1" not in rewriter._cache assert len(rewriter._cache) == 3 def test_no_eviction_when_under_limit(self): rewriter = QueryRewriter(cache_duration_seconds=3600) rewriter._max_cache_size = 10 rewriter._add_to_cache("k1", "v1") rewriter._add_to_cache("k2", "v2") assert len(rewriter._cache) == 2 def test_overwrite_existing_key_no_eviction(self): rewriter = QueryRewriter(cache_duration_seconds=3600) rewriter._max_cache_size = 2 rewriter._add_to_cache("k1", "v1") rewriter._add_to_cache("k2", "v2") rewriter._add_to_cache("k1", "v1_updated") assert len(rewriter._cache) == 2 assert rewriter._cache["k1"] == "v1_updated" def test_eviction_removes_from_ttl_dict(self): rewriter = QueryRewriter(cache_duration_seconds=3600) rewriter._max_cache_size = 2 rewriter._add_to_cache("old", "v1") time.sleep(0.01) rewriter._add_to_cache("mid", "v2") time.sleep(0.01) rewriter._add_to_cache("new", "v3") assert "old" not in rewriter._cache_ttl class TestCacheStats: def test_initial_stats(self, rewriter): stats = rewriter.get_cache_stats() assert stats["total_entries"] == 0 assert stats["hit_rate"] == 0.0 assert stats["cache_hits"] == 0 assert stats["cache_misses"] == 0 def test_stats_after_hits_and_misses(self, rewriter): rewriter._add_to_cache("k1", "v1") rewriter._get_from_cache("k1") rewriter._get_from_cache("k1") rewriter._get_from_cache("missing") stats = rewriter.get_cache_stats() assert stats["cache_hits"] == 2 assert stats["cache_misses"] == 1 assert stats["hit_rate"] == pytest.approx(2 / 3, abs=0.01) def test_stats_entry_ages(self, rewriter): rewriter._add_to_cache("k1", "v1") time.sleep(0.1) stats = rewriter.get_cache_stats() assert stats["oldest_entry_age"] >= 0.1 assert stats["newest_entry_age"] >= 0.1 def test_stats_max_size_from_config(self): rewriter = QueryRewriter(cache_duration_seconds=7200) stats = rewriter.get_cache_stats() assert stats["ttl_seconds"] == 7200 def test_stats_empty_ages(self, rewriter): stats = rewriter.get_cache_stats() assert stats["oldest_entry_age"] == 0.0 assert stats["newest_entry_age"] == 0.0 class TestCacheWarming: def test_warm_cache_adds_entries(self, rewriter): pairs = [("Bunlarin toplami?", "Satislarin toplami nedir?"), ("Bu nedir?", "Atlas nedir?")] rewriter.warm_cache(pairs) assert len(rewriter._cache) == 2 def test_warm_cache_entries_retrievable(self, rewriter): pairs = [("test query", "rewritten query")] rewriter.warm_cache(pairs) values = list(rewriter._cache.values()) assert "rewritten query" in values def test_warm_cache_empty_list(self, rewriter): rewriter.warm_cache([]) assert len(rewriter._cache) == 0 def test_warm_cache_respects_size_limit(self): rewriter = QueryRewriter(cache_duration_seconds=3600) rewriter._max_cache_size = 3 pairs = [(f"q{i}", f"r{i}") for i in range(5)] rewriter.warm_cache(pairs) assert len(rewriter._cache) == 3 class TestResetMetrics: def test_reset_clears_counters(self, rewriter): rewriter._cache_hits = 10 rewriter._cache_misses = 5 rewriter.reset_metrics() assert rewriter._cache_hits == 0 assert rewriter._cache_misses == 0 class TestClearCacheWithTTL: def test_clear_removes_ttl_entries(self, rewriter): rewriter._add_to_cache("k1", "v1") rewriter._add_to_cache("k2", "v2") rewriter.clear_cache() assert len(rewriter._cache) == 0 assert len(rewriter._cache_ttl) == 0 class TestCacheKeyHashing: def test_cache_key_is_md5_hash(self, rewriter, sample_history): key = rewriter._get_cache_key("test", sample_history) assert len(key) == 32 assert all(c in "0123456789abcdef" for c in key) def test_same_inputs_same_hash(self, rewriter, sample_history): k1 = rewriter._get_cache_key("query", sample_history) k2 = rewriter._get_cache_key("query", sample_history) assert k1 == k2 def test_different_inputs_different_hash(self, rewriter, sample_history): k1 = rewriter._get_cache_key("query1", sample_history) k2 = rewriter._get_cache_key("query2", sample_history) assert k1 != k2 class TestRewriteDebugInfoEnhancements: @patch.object(QueryRewriter, '_llm_rewrite') def test_debug_info_includes_cache_hit_field(self, mock_llm, rewriter, sample_history): mock_llm.return_value = "Atlas ozellikleri nelerdir?" _, debug = rewriter.rewrite("Bunun ozellikleri neler?", sample_history) assert "cache_hit" in debug assert debug["cache_hit"] is False @patch.object(QueryRewriter, '_llm_rewrite') def test_debug_info_includes_cache_size(self, mock_llm, rewriter, sample_history): mock_llm.return_value = "Atlas ozellikleri nelerdir?" _, debug = rewriter.rewrite("Bunun ozellikleri neler?", sample_history) assert "cache_size" in debug assert debug["cache_size"] == 1 @patch.object(QueryRewriter, '_llm_rewrite') def test_cache_hit_debug_info(self, mock_llm, rewriter, sample_history): mock_llm.return_value = "Atlas ozellikleri nelerdir?" rewriter.rewrite("Bunun ozellikleri neler?", sample_history) _, debug = rewriter.rewrite("Bunun ozellikleri neler?", sample_history) assert debug["cache_hit"] is True assert debug["method"] == "cache" assert "cache_age_seconds" in debug if __name__ == "__main__": pytest.main([__file__, "-v"])