""" Tests for news/recommender/topic_based.py Tests cover: - TopicBasedRecommender initialization - generate_recommendations() - main recommendation flow - _get_trending_topics() - topic retrieval logic - _filter_topics_by_preferences() - preference-based filtering - _generate_topic_query() - query generation - _create_recommendation_card() - card creation from search results - SearchBasedRecommender behavior - Error handling and edge cases """ from unittest.mock import Mock, patch class TestTopicBasedRecommenderInit: """Tests for TopicBasedRecommender initialization.""" def test_inherits_from_base_recommender(self): """TopicBasedRecommender inherits from BaseRecommender.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) assert issubclass(TopicBasedRecommender, BaseRecommender) def test_init_sets_max_recommendations_default(self): """Initialization sets default max_recommendations to 5.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() assert recommender.max_recommendations == 5 def test_init_with_dependencies(self): """Initialization accepts all base recommender dependencies.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_pref_manager = Mock() mock_rating_system = Mock() mock_topic_registry = Mock() recommender = TopicBasedRecommender( preference_manager=mock_pref_manager, rating_system=mock_rating_system, topic_registry=mock_topic_registry, ) assert recommender.preference_manager is mock_pref_manager assert recommender.rating_system is mock_rating_system assert recommender.topic_registry is mock_topic_registry def test_strategy_name_is_class_name(self): """Strategy name is set to TopicBasedRecommender.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() assert recommender.strategy_name == "TopicBasedRecommender" class TestGetTrendingTopics: """Tests for _get_trending_topics method.""" def test_get_trending_topics_from_registry(self): """Gets topics from topic registry when available.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.return_value = ["AI", "Climate"] recommender = TopicBasedRecommender(topic_registry=mock_registry) topics = recommender._get_trending_topics(None) assert "AI" in topics assert "Climate" in topics mock_registry.get_trending_topics.assert_called_once_with( hours=24, limit=20 ) def test_get_trending_topics_with_context_news_topics(self): """Includes topics from context current_news_topics.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.return_value = ["AI"] context = {"current_news_topics": ["Technology", "Science"]} recommender = TopicBasedRecommender(topic_registry=mock_registry) topics = recommender._get_trending_topics(context) assert "AI" in topics assert "Technology" in topics assert "Science" in topics def test_get_trending_topics_fallback_defaults(self): """Uses fallback topics when no registry and no topics found.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = recommender._get_trending_topics(None) # Check some default topics are present assert len(topics) == 5 assert "artificial intelligence developments" in topics assert "cybersecurity threats" in topics def test_get_trending_topics_empty_registry_uses_fallback(self): """Uses fallback when registry returns empty list.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.return_value = [] recommender = TopicBasedRecommender(topic_registry=mock_registry) topics = recommender._get_trending_topics(None) # Should fall back to defaults assert len(topics) == 5 def test_get_trending_topics_context_with_category(self): """Handles context with current_category (currently pass-through).""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.return_value = ["AI"] context = {"current_category": "Technology"} recommender = TopicBasedRecommender(topic_registry=mock_registry) topics = recommender._get_trending_topics(context) # current_category is handled but currently just passes assert "AI" in topics class TestFilterTopicsByPreferences: """Tests for _filter_topics_by_preferences method.""" def test_filter_removes_disliked_topics(self): """Filters out topics that match disliked_topics.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = ["AI news", "Politics update", "Science discovery"] preferences = {"disliked_topics": ["politics"]} filtered = recommender._filter_topics_by_preferences( topics, preferences ) assert "Politics update" not in filtered assert "AI news" in filtered assert "Science discovery" in filtered def test_filter_case_insensitive_disliked(self): """Disliked topics filter is case-insensitive.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = ["POLITICS news", "AI Technology"] preferences = {"disliked_topics": ["Politics"]} filtered = recommender._filter_topics_by_preferences( topics, preferences ) assert "POLITICS news" not in filtered assert "AI Technology" in filtered def test_filter_boosts_interest_topics(self): """Topics matching interests are sorted to front.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = ["Sports news", "AI breakthrough", "Weather update"] preferences = {"interests": {"ai": 2.0, "weather": 1.5}} filtered = recommender._filter_topics_by_preferences( topics, preferences ) # AI should be boosted higher than weather assert filtered.index("AI breakthrough") < filtered.index( "Weather update" ) assert "Sports news" in filtered def test_filter_empty_preferences(self): """Empty preferences returns all topics.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = ["AI", "Politics", "Science"] preferences = {} filtered = recommender._filter_topics_by_preferences( topics, preferences ) assert len(filtered) == 3 def test_filter_partial_match_disliked(self): """Partial match on disliked topics works.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = ["political analysis", "AI politics", "Science"] preferences = {"disliked_topics": ["politic"]} filtered = recommender._filter_topics_by_preferences( topics, preferences ) assert "political analysis" not in filtered assert "AI politics" not in filtered assert "Science" in filtered def test_filter_multiple_interests_first_match_wins(self): """First matching interest determines boost value.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = ["AI technology news"] preferences = {"interests": {"ai": 3.0, "technology": 1.5}} filtered = recommender._filter_topics_by_preferences( topics, preferences ) # Topic should be present (boost applied internally) assert "AI technology news" in filtered class TestGenerateTopicQuery: """Tests for _generate_topic_query method.""" def test_generate_query_adds_news_context(self): """Query includes news-specific context words.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() query = recommender._generate_topic_query("AI") assert "AI" in query assert "latest" in query assert "news" in query assert "today" in query def test_generate_query_preserves_topic(self): """Original topic is preserved in query.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() query = recommender._generate_topic_query("climate change impacts") assert "climate change impacts" in query class TestCreateRecommendationCard: """Tests for _create_recommendation_card method.""" @patch("local_deep_research.config.search_config.get_search") @patch("local_deep_research.config.llm_config.get_llm") @patch( "local_deep_research.news.recommender.topic_based.AdvancedSearchSystem" ) @patch("local_deep_research.news.recommender.topic_based.CardFactory") def test_create_card_success( self, mock_factory, mock_search_class, _mock_get_llm, _mock_get_search ): """Successfully creates card from search results.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) # Mock search system mock_search = Mock() mock_search.analyze_topic.return_value = { "search_id": "search-123", "news_items": [ { "headline": "AI News", "impact_score": 8, "summary": "Summary", } ], "formatted_findings": "Big picture", } mock_search_class.return_value = mock_search # Mock card factory mock_card = Mock() mock_factory.create_news_card_from_analysis.return_value = mock_card recommender = TopicBasedRecommender() card = recommender._create_recommendation_card( "AI", "AI query", "user123" ) assert card is mock_card mock_card.add_version.assert_called_once() @patch("local_deep_research.config.search_config.get_search") @patch("local_deep_research.config.llm_config.get_llm") @patch( "local_deep_research.news.recommender.topic_based.AdvancedSearchSystem" ) def test_create_card_search_error( self, mock_search_class, _mock_get_llm, _mock_get_search ): """Returns None when search returns error.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_search = Mock() mock_search.analyze_topic.return_value = {"error": "Search failed"} mock_search_class.return_value = mock_search recommender = TopicBasedRecommender() card = recommender._create_recommendation_card( "AI", "AI query", "user123" ) assert card is None @patch("local_deep_research.config.search_config.get_search") @patch("local_deep_research.config.llm_config.get_llm") @patch( "local_deep_research.news.recommender.topic_based.AdvancedSearchSystem" ) def test_create_card_no_news_items( self, mock_search_class, _mock_get_llm, _mock_get_search ): """Returns None when no news items found.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_search = Mock() mock_search.analyze_topic.return_value = { "news_items": [], "formatted_findings": "", } mock_search_class.return_value = mock_search recommender = TopicBasedRecommender() card = recommender._create_recommendation_card( "AI", "AI query", "user123" ) assert card is None @patch("local_deep_research.config.search_config.get_search") @patch("local_deep_research.config.llm_config.get_llm") @patch( "local_deep_research.news.recommender.topic_based.AdvancedSearchSystem" ) def test_create_card_exception_handling( self, mock_search_class, _mock_get_llm, _mock_get_search ): """Returns None and logs on exception.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_search = Mock() mock_search.analyze_topic.side_effect = Exception("Search error") mock_search_class.return_value = mock_search recommender = TopicBasedRecommender() card = recommender._create_recommendation_card( "AI", "AI query", "user123" ) assert card is None @patch("local_deep_research.config.search_config.get_search") @patch("local_deep_research.config.llm_config.get_llm") @patch( "local_deep_research.news.recommender.topic_based.AdvancedSearchSystem" ) @patch("local_deep_research.news.recommender.topic_based.CardFactory") def test_create_card_selects_highest_impact( self, mock_factory, mock_search_class, _mock_get_llm, _mock_get_search ): """Selects news item with highest impact score.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_search = Mock() mock_search.analyze_topic.return_value = { "search_id": "search-123", "news_items": [ {"headline": "Low Impact", "impact_score": 3}, {"headline": "High Impact", "impact_score": 9}, {"headline": "Medium Impact", "impact_score": 6}, ], "formatted_findings": "", } mock_search_class.return_value = mock_search mock_card = Mock() mock_factory.create_news_card_from_analysis.return_value = mock_card recommender = TopicBasedRecommender() recommender._create_recommendation_card("AI", "AI query", "user123") # Verify highest impact item was selected call_args = mock_factory.create_news_card_from_analysis.call_args selected_item = call_args[1]["news_item"] assert selected_item["headline"] == "High Impact" class TestGenerateRecommendations: """Tests for generate_recommendations method.""" @patch.object( __import__( "local_deep_research.news.recommender.topic_based", fromlist=["TopicBasedRecommender"], ).TopicBasedRecommender, "_create_recommendation_card", ) def test_generate_recommendations_full_flow(self, mock_create_card): """Full recommendation flow creates cards for filtered topics.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_card = Mock() mock_create_card.return_value = mock_card mock_registry = Mock() mock_registry.get_trending_topics.return_value = ["AI", "Tech"] recommender = TopicBasedRecommender(topic_registry=mock_registry) recommendations = recommender.generate_recommendations("user123") assert len(recommendations) > 0 mock_create_card.assert_called() def test_generate_recommendations_respects_max_limit(self): """Only processes max_recommendations topics.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.return_value = [ f"Topic {i}" for i in range(20) ] recommender = TopicBasedRecommender(topic_registry=mock_registry) recommender.max_recommendations = 3 # Mock _create_recommendation_card to track calls create_card_calls = [] def mock_create_card(topic, query, user_id): create_card_calls.append(topic) return # Return None to avoid further processing recommender._create_recommendation_card = mock_create_card recommender.generate_recommendations("user123") # Should only process 3 topics assert len(create_card_calls) == 3 def test_generate_recommendations_handles_exception(self): """Returns empty list on exception.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.side_effect = Exception( "Registry error" ) recommender = TopicBasedRecommender(topic_registry=mock_registry) recommendations = recommender.generate_recommendations("user123") assert recommendations == [] def test_generate_recommendations_updates_progress(self): """Progress callback is called during recommendation generation.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.return_value = ["AI"] progress_calls = [] def progress_callback(message, percent, metadata): progress_calls.append((message, percent)) recommender = TopicBasedRecommender(topic_registry=mock_registry) recommender.set_progress_callback(progress_callback) recommender._create_recommendation_card = Mock(return_value=None) recommender.generate_recommendations("user123") # Should have progress updates assert len(progress_calls) > 0 # Final progress should be 100 assert any(p[1] == 100 for p in progress_calls) def test_generate_recommendations_applies_user_preferences(self): """User preferences are applied to filter topics.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_pref_manager = Mock() mock_pref_manager.get_preferences.return_value = { "disliked_topics": ["politics"] } mock_registry = Mock() mock_registry.get_trending_topics.return_value = ["AI", "Politics news"] topics_processed = [] def mock_create_card(topic, query, user_id): topics_processed.append(topic) return recommender = TopicBasedRecommender( preference_manager=mock_pref_manager, topic_registry=mock_registry ) recommender._create_recommendation_card = mock_create_card recommender.generate_recommendations("user123") # Politics should be filtered out assert "Politics news" not in topics_processed assert "AI" in topics_processed def test_generate_recommendations_skips_failed_cards(self): """Continues processing when card creation fails.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.return_value = [ "AI", "Tech", "Science", ] call_count = [0] def mock_create_card(topic, query, user_id): call_count[0] += 1 if topic == "Tech": raise Exception("Card creation failed") return Mock() if topic == "Science" else None recommender = TopicBasedRecommender(topic_registry=mock_registry) recommender._create_recommendation_card = mock_create_card recommendations = recommender.generate_recommendations("user123") # Should process all 3 topics despite failure assert call_count[0] == 3 # Should have 1 recommendation (Science) assert len(recommendations) == 1 def test_generate_recommendations_sorts_by_relevance(self): """Recommendations are sorted by relevance.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.return_value = ["AI"] mock_card_low = Mock() mock_card_low.impact_score = 3 mock_card_low.metadata = {} mock_card_high = Mock() mock_card_high.impact_score = 9 mock_card_high.metadata = {} cards_to_return = [mock_card_low, mock_card_high] def mock_create_card(topic, query, user_id): return cards_to_return.pop(0) if cards_to_return else None recommender = TopicBasedRecommender(topic_registry=mock_registry) recommender.max_recommendations = 2 mock_registry.get_trending_topics.return_value = ["AI", "Tech"] recommender._create_recommendation_card = mock_create_card recommendations = recommender.generate_recommendations("user123") # Higher impact should be first if len(recommendations) == 2: assert ( recommendations[0].impact_score > recommendations[1].impact_score ) def test_generate_recommendations_with_context(self): """Context is passed to _get_trending_topics.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() recommender._create_recommendation_card = Mock(return_value=None) context = {"current_news_topics": ["Custom Topic"]} recommender.generate_recommendations("user123", context=context) # Should have processed custom topic from context # (verified by the fact that no errors occurred) class TestSearchBasedRecommender: """Tests for SearchBasedRecommender class.""" def test_inherits_from_base_recommender(self): """SearchBasedRecommender inherits from BaseRecommender.""" from local_deep_research.news.recommender.topic_based import ( SearchBasedRecommender, ) from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) assert issubclass(SearchBasedRecommender, BaseRecommender) def test_generate_recommendations_returns_empty_list(self): """Returns empty list since search tracking is disabled.""" from local_deep_research.news.recommender.topic_based import ( SearchBasedRecommender, ) recommender = SearchBasedRecommender() recommendations = recommender.generate_recommendations("user123") assert recommendations == [] def test_generate_recommendations_with_context(self): """Accepts context parameter (unused currently).""" from local_deep_research.news.recommender.topic_based import ( SearchBasedRecommender, ) recommender = SearchBasedRecommender() recommendations = recommender.generate_recommendations( "user123", context={"page": "home"} ) assert recommendations == [] class TestEdgeCases: """Tests for edge cases and boundary conditions.""" def test_empty_user_id(self): """Handles empty user_id gracefully.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() recommender._create_recommendation_card = Mock(return_value=None) # Should not raise recommendations = recommender.generate_recommendations("") assert isinstance(recommendations, list) def test_none_context(self): """Handles None context.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = recommender._get_trending_topics(None) assert isinstance(topics, list) def test_empty_topics_list(self): """Handles empty topics list in filter.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() filtered = recommender._filter_topics_by_preferences([], {}) assert filtered == [] def test_unicode_topics(self): """Handles unicode characters in topics.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = ["AI 人工智能", "Climate 气候变化", "Tech"] preferences = {} filtered = recommender._filter_topics_by_preferences( topics, preferences ) assert len(filtered) == 3 def test_special_characters_in_topic(self): """Handles special characters in topic names.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() query = recommender._generate_topic_query("C++ & Python: What's new?") assert "C++ & Python: What's new?" in query def test_very_long_topic_name(self): """Handles very long topic names.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() long_topic = "A" * 1000 query = recommender._generate_topic_query(long_topic) assert long_topic in query def test_max_recommendations_zero(self): """Handles max_recommendations set to zero.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) mock_registry = Mock() mock_registry.get_trending_topics.return_value = ["AI", "Tech"] create_card_calls = [] def mock_create_card(topic, query, user_id): create_card_calls.append(topic) return Mock() recommender = TopicBasedRecommender(topic_registry=mock_registry) recommender.max_recommendations = 0 recommender._create_recommendation_card = mock_create_card recommender.generate_recommendations("user123") # Should not process any topics assert len(create_card_calls) == 0 def test_preferences_with_empty_lists(self): """Handles preferences with empty disliked_topics list.""" from local_deep_research.news.recommender.topic_based import ( TopicBasedRecommender, ) recommender = TopicBasedRecommender() topics = ["AI", "Tech"] preferences = {"disliked_topics": [], "interests": {}} filtered = recommender._filter_topics_by_preferences( topics, preferences ) assert len(filtered) == 2