""" Tests for news/recommender/base_recommender.py Tests cover: - BaseRecommender initialization - Progress callback handling - User preference access - Abstract method requirements """ import pytest from unittest.mock import Mock from abc import ABC class TestBaseRecommenderInit: """Tests for BaseRecommender initialization.""" def test_base_recommender_is_abstract(self): """BaseRecommender is an abstract class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) assert issubclass(BaseRecommender, ABC) def test_base_recommender_has_abstract_method(self): """BaseRecommender requires generate_recommendations.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) assert hasattr(BaseRecommender, "generate_recommendations") class TestConcreteRecommender: """Tests using a concrete implementation of BaseRecommender.""" @pytest.fixture def mock_preference_manager(self): """Create mock preference manager.""" mock = Mock() mock.get_preferences.return_value = {"topic": "test"} return mock @pytest.fixture def mock_rating_system(self): """Create mock rating system.""" mock = Mock() mock.get_user_ratings.return_value = [] return mock @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] return TestRecommender def test_recommender_initialization_with_defaults( self, concrete_recommender ): """Recommender initializes with default None values.""" recommender = concrete_recommender() assert recommender.preference_manager is None assert recommender.rating_system is None assert recommender.topic_registry is None assert recommender.search_system is None assert recommender.progress_callback is None def test_recommender_initialization_with_dependencies( self, concrete_recommender, mock_preference_manager, mock_rating_system ): """Recommender initializes with provided dependencies.""" recommender = concrete_recommender( preference_manager=mock_preference_manager, rating_system=mock_rating_system, ) assert recommender.preference_manager is mock_preference_manager assert recommender.rating_system is mock_rating_system def test_strategy_name_is_class_name(self, concrete_recommender): """Strategy name is set to class name.""" recommender = concrete_recommender() assert recommender.strategy_name == "TestRecommender" class TestProgressCallback: """Tests for progress callback functionality.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): self._update_progress("Processing", 50, {"step": 1}) return [] return TestRecommender def test_set_progress_callback(self, concrete_recommender): """Progress callback can be set.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) assert recommender.progress_callback is callback def test_update_progress_calls_callback(self, concrete_recommender): """_update_progress calls the callback when set.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) recommender._update_progress("Test message", 50, {"key": "value"}) callback.assert_called_once_with("Test message", 50, {"key": "value"}) def test_update_progress_does_nothing_without_callback( self, concrete_recommender ): """_update_progress doesn't fail without callback.""" recommender = concrete_recommender() # Should not raise recommender._update_progress("Test message", 50, {}) def test_update_progress_default_metadata(self, concrete_recommender): """_update_progress uses empty dict for default metadata.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) recommender._update_progress("Test message", 50) callback.assert_called_once_with("Test message", 50, {}) class TestUserPreferences: """Tests for user preference handling.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def get_prefs(self, user_id): return self._get_user_preferences(user_id) return TestRecommender def test_get_user_preferences_with_manager(self, concrete_recommender): """_get_user_preferences returns preferences when manager available.""" mock_manager = Mock() mock_manager.get_preferences.return_value = {"topic": "test"} recommender = concrete_recommender(preference_manager=mock_manager) prefs = recommender.get_prefs("user123") assert prefs == {"topic": "test"} mock_manager.get_preferences.assert_called_once_with("user123") def test_get_user_preferences_without_manager(self, concrete_recommender): """_get_user_preferences returns empty dict without manager.""" recommender = concrete_recommender() prefs = recommender.get_prefs("user123") assert prefs == {} class TestGenerateRecommendations: """Tests for the generate_recommendations abstract method.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [{"id": 1, "topic": "test"}] return TestRecommender def test_generate_recommendations_returns_list(self, concrete_recommender): """generate_recommendations returns a list.""" recommender = concrete_recommender() result = recommender.generate_recommendations("user123") assert isinstance(result, list) def test_generate_recommendations_accepts_context( self, concrete_recommender ): """generate_recommendations accepts optional context.""" recommender = concrete_recommender() # Should not raise result = recommender.generate_recommendations( "user123", context={"page": "home"} ) assert isinstance(result, list) # ============================================================================= # Tests for _get_user_ratings # ============================================================================= class TestGetUserRatings: """Tests for the _get_user_ratings method.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def get_ratings(self, user_id, limit=50): return self._get_user_ratings(user_id, limit) return TestRecommender def test_returns_ratings_from_system(self, concrete_recommender): """_get_user_ratings returns ratings from rating_system.""" mock_rating_system = Mock() mock_rating_system.get_recent_ratings.return_value = [ {"card_id": "card1", "rating": 5}, {"card_id": "card2", "rating": 3}, ] recommender = concrete_recommender(rating_system=mock_rating_system) ratings = recommender.get_ratings("user123") assert len(ratings) == 2 assert ratings[0]["card_id"] == "card1" mock_rating_system.get_recent_ratings.assert_called_once_with( "user123", 50 ) def test_respects_limit_parameter(self, concrete_recommender): """_get_user_ratings passes limit to rating_system.""" mock_rating_system = Mock() mock_rating_system.get_recent_ratings.return_value = [] recommender = concrete_recommender(rating_system=mock_rating_system) recommender.get_ratings("user123", limit=10) mock_rating_system.get_recent_ratings.assert_called_once_with( "user123", 10 ) def test_returns_empty_when_no_rating_system(self, concrete_recommender): """_get_user_ratings returns empty list when no rating_system.""" recommender = concrete_recommender() ratings = recommender.get_ratings("user123") assert ratings == [] # ============================================================================= # Tests for _execute_search # ============================================================================= class TestExecuteSearch: """Tests for the _execute_search method.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def do_search(self, query, strategy=None): return self._execute_search(query, strategy) return TestRecommender def test_returns_error_when_no_search_system(self, concrete_recommender): """_execute_search returns error dict when no search_system.""" recommender = concrete_recommender() result = recommender.do_search("test query") assert "error" in result assert result["error"] == "Search system not configured" def test_uses_news_aggregation_strategy(self, concrete_recommender): """_execute_search uses news_aggregation as default strategy.""" mock_search = Mock() mock_search.analyze_topic.return_value = {"results": []} recommender = concrete_recommender(search_system=mock_search) # Strategy defaults to "news_aggregation" but analyze_topic is called recommender.do_search("test query") mock_search.analyze_topic.assert_called_once_with("test query") def test_calls_analyze_topic(self, concrete_recommender): """_execute_search calls analyze_topic with the query.""" mock_search = Mock() mock_search.analyze_topic.return_value = {"findings": ["test"]} recommender = concrete_recommender(search_system=mock_search) result = recommender.do_search("climate change news") mock_search.analyze_topic.assert_called_once_with("climate change news") assert result == {"findings": ["test"]} def test_handles_exception_gracefully(self, concrete_recommender): """_execute_search returns error dict on exception.""" mock_search = Mock() mock_search.analyze_topic.side_effect = Exception("Search failed") recommender = concrete_recommender(search_system=mock_search) result = recommender.do_search("test query") assert "error" in result assert result["error"] == "Recommendation search failed" def test_logs_search_execution(self, concrete_recommender): """_execute_search logs errors on failure.""" from unittest.mock import patch mock_search = Mock() mock_search.analyze_topic.side_effect = Exception("Network error") recommender = concrete_recommender(search_system=mock_search) with patch( "local_deep_research.news.recommender.base_recommender.logger" ) as mock_logger: recommender.do_search("test query") mock_logger.exception.assert_called_once() # ============================================================================= # Tests for _filter_by_preferences (CRITICAL) # ============================================================================= class TestFilterByPreferences: """Tests for the _filter_by_preferences method - core filtering logic.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender @pytest.fixture def sample_cards(self): """Create sample NewsCard objects for testing.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="AI in healthcare", source=source, user_id="user1", category="Technology", impact_score=8, ), NewsCard( topic="Climate change policy", source=source, user_id="user1", category="Environment", impact_score=6, ), NewsCard( topic="Stock market update", source=source, user_id="user1", category="Finance", impact_score=4, ), NewsCard( topic="Machine learning breakthrough", source=source, user_id="user1", category="Technology", impact_score=9, ), ] return cards def test_returns_all_cards_with_empty_preferences( self, concrete_recommender, sample_cards ): """Empty preferences returns all cards unchanged.""" recommender = concrete_recommender() result = recommender.filter_cards(sample_cards, {}) assert len(result) == len(sample_cards) def test_adds_preference_boost_for_liked_categories( self, concrete_recommender, sample_cards ): """Cards in liked_categories get preference_boost in metadata.""" recommender = concrete_recommender() preferences = {"liked_categories": ["Technology"]} result = recommender.filter_cards(sample_cards, preferences) # Technology cards should have boost tech_cards = [c for c in result if c.category == "Technology"] assert len(tech_cards) == 2 for card in tech_cards: assert card.metadata.get("preference_boost") == 1.2 # Non-technology cards should not have boost non_tech_cards = [c for c in result if c.category != "Technology"] for card in non_tech_cards: assert "preference_boost" not in card.metadata def test_filters_low_impact_cards(self, concrete_recommender, sample_cards): """Cards below impact_threshold are filtered out.""" recommender = concrete_recommender() preferences = {"impact_threshold": 5} result = recommender.filter_cards(sample_cards, preferences) # Only cards with impact_score >= 5 should remain assert len(result) == 3 for card in result: assert card.impact_score >= 5 def test_removes_cards_with_disliked_topics( self, concrete_recommender, sample_cards ): """Cards matching disliked_topics are removed.""" recommender = concrete_recommender() preferences = {"disliked_topics": ["stock", "market"]} result = recommender.filter_cards(sample_cards, preferences) # "Stock market update" should be filtered out assert len(result) == 3 topics = [c.topic for c in result] assert "Stock market update" not in topics def test_topic_matching_converts_topic_to_lowercase( self, concrete_recommender, sample_cards ): """Disliked topics must be lowercase since topic.lower() is used.""" recommender = concrete_recommender() # Disliked topics must be lowercase to match (topic is lowercased) preferences = {"disliked_topics": ["ai"]} result = recommender.filter_cards(sample_cards, preferences) # "AI in healthcare" should be filtered out (topic is lowercased) topics = [c.topic for c in result] assert "AI in healthcare" not in topics def test_handles_multiple_filter_criteria( self, concrete_recommender, sample_cards ): """Multiple filter criteria are applied together.""" recommender = concrete_recommender() preferences = { "liked_categories": ["Technology"], "impact_threshold": 7, "disliked_topics": ["stock"], } result = recommender.filter_cards(sample_cards, preferences) # Should filter by impact >= 7 AND not contain "stock" # Remaining: AI in healthcare (8), Machine learning breakthrough (9) assert len(result) == 2 for card in result: assert card.impact_score >= 7 def test_handles_empty_cards_list(self, concrete_recommender): """Empty cards list returns empty list.""" recommender = concrete_recommender() preferences = {"impact_threshold": 5} result = recommender.filter_cards([], preferences) assert result == [] def test_handles_none_values_in_safe_fields( self, concrete_recommender, sample_cards ): """None values in liked_categories and disliked_topics are handled safely.""" recommender = concrete_recommender() # Note: None in impact_threshold will raise TypeError (comparison with None) # Only liked_categories and disliked_topics handle None gracefully preferences = { "liked_categories": None, "disliked_topics": None, } # Should not raise result = recommender.filter_cards(sample_cards, preferences) # All cards should be returned (no filtering applied for None values) assert len(result) == len(sample_cards) def test_preserves_original_card_order( self, concrete_recommender, sample_cards ): """Filtering preserves the original order of cards.""" recommender = concrete_recommender() # Filter out one card preferences = {"disliked_topics": ["stock"]} original_order = [ c.topic for c in sample_cards if "stock" not in c.topic.lower() ] result = recommender.filter_cards(sample_cards, preferences) result_order = [c.topic for c in result] assert result_order == original_order def test_filters_partial_topic_matches( self, concrete_recommender, sample_cards ): """Disliked topics filter on substring matches.""" recommender = concrete_recommender() # "machine" should match "Machine learning breakthrough" preferences = {"disliked_topics": ["machine"]} result = recommender.filter_cards(sample_cards, preferences) topics = [c.topic for c in result] assert "Machine learning breakthrough" not in topics # ============================================================================= # Tests for _sort_by_relevance # ============================================================================= class TestSortByRelevance: """Tests for the _sort_by_relevance method.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender @pytest.fixture def sample_cards(self): """Create sample NewsCard objects with varying scores.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Low impact", source=source, user_id="user1", impact_score=3, ), NewsCard( topic="High impact", source=source, user_id="user1", impact_score=9, ), NewsCard( topic="Medium impact", source=source, user_id="user1", impact_score=6, ), ] return cards def test_sorts_by_impact_score_descending( self, concrete_recommender, sample_cards ): """Cards are sorted by impact_score in descending order.""" recommender = concrete_recommender() result = recommender.sort_cards(sample_cards, "user123") # Should be ordered: 9, 6, 3 assert result[0].impact_score == 9 assert result[1].impact_score == 6 assert result[2].impact_score == 3 def test_applies_preference_boost(self, concrete_recommender, sample_cards): """Preference boost affects sorting order.""" recommender = concrete_recommender() # Give the low impact card a boost sample_cards[0].metadata["preference_boost"] = 5.0 # Low impact (3) result = recommender.sort_cards(sample_cards, "user123") # Low impact card (3 * 5.0 = 15) should now be first # Score calculation: (impact/10) * boost # Low: (3/10) * 5.0 = 1.5 # High: (9/10) * 1.0 = 0.9 # Medium: (6/10) * 1.0 = 0.6 assert result[0].topic == "Low impact" def test_handles_equal_scores_stably(self, concrete_recommender): """Cards with equal scores maintain stable sort.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="First", source=source, user_id="user1", impact_score=5 ), NewsCard( topic="Second", source=source, user_id="user1", impact_score=5 ), NewsCard( topic="Third", source=source, user_id="user1", impact_score=5 ), ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") # Python's sort is stable - equal elements maintain relative order assert len(result) == 3 def test_handles_empty_list(self, concrete_recommender): """Empty list returns empty list.""" recommender = concrete_recommender() result = recommender.sort_cards([], "user123") assert result == [] def test_score_calculation(self, concrete_recommender): """Score is calculated as (impact/10) * boost.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") # Card with impact 8 and boost 1.5 # Score = (8/10) * 1.5 = 1.2 card_boosted = NewsCard( topic="Boosted", source=source, user_id="user1", impact_score=8 ) card_boosted.metadata["preference_boost"] = 1.5 # Card with impact 10 and no boost (default 1.0) # Score = (10/10) * 1.0 = 1.0 card_high = NewsCard( topic="High", source=source, user_id="user1", impact_score=10 ) cards = [card_high, card_boosted] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") # Boosted card (1.2) should rank higher than high impact (1.0) assert result[0].topic == "Boosted" assert result[1].topic == "High" # ============================================================================= # Tests for get_strategy_info # ============================================================================= class TestGetStrategyInfo: """Tests for the get_strategy_info method.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class MyCustomRecommender(BaseRecommender): """Custom recommender for testing info display.""" def generate_recommendations(self, user_id, context=None): return [] return MyCustomRecommender def test_returns_correct_structure(self, concrete_recommender): """get_strategy_info returns dict with expected keys.""" recommender = concrete_recommender() info = recommender.get_strategy_info() assert "name" in info assert "has_preference_manager" in info assert "has_rating_system" in info assert "has_search_system" in info assert "description" in info def test_boolean_flags_reflect_dependencies(self, concrete_recommender): """has_* flags accurately reflect dependency availability.""" # Without dependencies recommender_empty = concrete_recommender() info_empty = recommender_empty.get_strategy_info() assert info_empty["has_preference_manager"] is False assert info_empty["has_rating_system"] is False assert info_empty["has_search_system"] is False # With dependencies recommender_full = concrete_recommender( preference_manager=Mock(), rating_system=Mock(), search_system=Mock(), ) info_full = recommender_full.get_strategy_info() assert info_full["has_preference_manager"] is True assert info_full["has_rating_system"] is True assert info_full["has_search_system"] is True def test_uses_class_name_as_strategy(self, concrete_recommender): """Strategy name is derived from class name.""" recommender = concrete_recommender() info = recommender.get_strategy_info() assert info["name"] == "MyCustomRecommender" # ============================================================================= # Tests for topic_registry attribute # ============================================================================= class TestTopicRegistry: """Tests for topic_registry attribute handling.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] return TestRecommender def test_topic_registry_is_none_by_default(self, concrete_recommender): """topic_registry is None when not provided.""" recommender = concrete_recommender() assert recommender.topic_registry is None def test_topic_registry_can_be_set(self, concrete_recommender): """topic_registry can be set during initialization.""" mock_registry = Mock() mock_registry.get_topics.return_value = ["AI", "Climate"] recommender = concrete_recommender(topic_registry=mock_registry) assert recommender.topic_registry is mock_registry assert recommender.topic_registry.get_topics() == ["AI", "Climate"] # ============================================================================= # Additional tests for _execute_search # ============================================================================= class TestExecuteSearchAdditional: """Additional tests for _execute_search edge cases.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def do_search(self, query, strategy=None): return self._execute_search(query, strategy) return TestRecommender def test_ignores_strategy_parameter(self, concrete_recommender): """_execute_search calls analyze_topic regardless of strategy parameter.""" mock_search = Mock() mock_search.analyze_topic.return_value = {"results": []} recommender = concrete_recommender(search_system=mock_search) # Strategy parameter is set but analyze_topic is still used recommender.do_search("test query", strategy="custom_strategy") mock_search.analyze_topic.assert_called_once_with("test query") def test_handles_none_result_from_search(self, concrete_recommender): """_execute_search handles None return from analyze_topic.""" mock_search = Mock() mock_search.analyze_topic.return_value = None recommender = concrete_recommender(search_system=mock_search) result = recommender.do_search("test query") assert result is None def test_handles_empty_dict_result(self, concrete_recommender): """_execute_search handles empty dict return.""" mock_search = Mock() mock_search.analyze_topic.return_value = {} recommender = concrete_recommender(search_system=mock_search) result = recommender.do_search("test query") assert result == {} def test_handles_various_exception_types(self, concrete_recommender): """_execute_search handles various exception types.""" mock_search = Mock() recommender = concrete_recommender(search_system=mock_search) # Test with different exception types for exc_type in [ValueError, RuntimeError, KeyError, TimeoutError]: mock_search.analyze_topic.side_effect = exc_type("Error") result = recommender.do_search("test query") assert "error" in result assert result["error"] == "Recommendation search failed" def test_logs_warning_when_no_search_system(self, concrete_recommender): """_execute_search logs warning when search system is not available.""" from unittest.mock import patch recommender = concrete_recommender() with patch( "local_deep_research.news.recommender.base_recommender.logger" ) as mock_logger: recommender.do_search("test query") mock_logger.warning.assert_called_once() # ============================================================================= # Additional tests for _filter_by_preferences # ============================================================================= class TestFilterByPreferencesAdditional: """Additional tests for _filter_by_preferences edge cases.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender @pytest.fixture def sample_cards(self): """Create sample NewsCard objects.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") return [ NewsCard( topic="Python programming tips", source=source, user_id="user1", category="Programming", impact_score=7, ), NewsCard( topic="JavaScript frameworks", source=source, user_id="user1", category="Programming", impact_score=5, ), NewsCard( topic="Data science trends", source=source, user_id="user1", category="Data Science", impact_score=8, ), ] def test_only_impact_threshold_filter( self, concrete_recommender, sample_cards ): """Test filtering with only impact_threshold preference.""" recommender = concrete_recommender() preferences = {"impact_threshold": 6} result = recommender.filter_cards(sample_cards, preferences) # Should keep cards with score >= 6 assert len(result) == 2 topics = [c.topic for c in result] assert "JavaScript frameworks" not in topics # score 5 def test_only_liked_categories_filter( self, concrete_recommender, sample_cards ): """Test filtering with only liked_categories preference.""" recommender = concrete_recommender() preferences = {"liked_categories": ["Data Science"]} result = recommender.filter_cards(sample_cards, preferences) # All cards returned, but Data Science cards get boost assert len(result) == 3 data_science_cards = [c for c in result if c.category == "Data Science"] programming_cards = [c for c in result if c.category == "Programming"] assert data_science_cards[0].metadata.get("preference_boost") == 1.2 for card in programming_cards: assert "preference_boost" not in card.metadata def test_only_disliked_topics_filter( self, concrete_recommender, sample_cards ): """Test filtering with only disliked_topics preference.""" recommender = concrete_recommender() preferences = {"disliked_topics": ["javascript"]} result = recommender.filter_cards(sample_cards, preferences) assert len(result) == 2 topics = [c.topic for c in result] assert "JavaScript frameworks" not in topics def test_disliked_topics_with_multiple_keywords( self, concrete_recommender, sample_cards ): """Test disliked_topics with multiple keywords.""" recommender = concrete_recommender() preferences = {"disliked_topics": ["python", "javascript"]} result = recommender.filter_cards(sample_cards, preferences) assert len(result) == 1 assert result[0].topic == "Data science trends" def test_empty_liked_categories_list( self, concrete_recommender, sample_cards ): """Empty liked_categories list applies no boost.""" recommender = concrete_recommender() preferences = {"liked_categories": []} result = recommender.filter_cards(sample_cards, preferences) # All cards returned, none with boost assert len(result) == 3 for card in result: assert "preference_boost" not in card.metadata def test_empty_disliked_topics_list( self, concrete_recommender, sample_cards ): """Empty disliked_topics list filters nothing.""" recommender = concrete_recommender() preferences = {"disliked_topics": []} result = recommender.filter_cards(sample_cards, preferences) assert len(result) == 3 def test_all_cards_filtered_out(self, concrete_recommender, sample_cards): """All cards can be filtered out if threshold too high.""" recommender = concrete_recommender() preferences = {"impact_threshold": 100} result = recommender.filter_cards(sample_cards, preferences) assert len(result) == 0 def test_boost_does_not_modify_impact_score( self, concrete_recommender, sample_cards ): """Preference boost goes to metadata, not impact_score.""" recommender = concrete_recommender() preferences = {"liked_categories": ["Programming"]} original_scores = [c.impact_score for c in sample_cards] result = recommender.filter_cards(sample_cards, preferences) result_scores = [c.impact_score for c in result] assert original_scores == result_scores def test_unrecognized_preference_keys_ignored( self, concrete_recommender, sample_cards ): """Unrecognized preference keys are ignored.""" recommender = concrete_recommender() preferences = { "unknown_key": "value", "another_unknown": [1, 2, 3], "impact_threshold": 6, } result = recommender.filter_cards(sample_cards, preferences) # Only impact_threshold applied assert len(result) == 2 # ============================================================================= # Additional tests for _sort_by_relevance # ============================================================================= class TestSortByRelevanceAdditional: """Additional tests for _sort_by_relevance edge cases.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_single_card_returns_same_card(self, concrete_recommender): """Single card list returns that card.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Single", source=source, user_id="user1", impact_score=5 ) recommender = concrete_recommender() result = recommender.sort_cards([card], "user123") assert len(result) == 1 assert result[0].topic == "Single" def test_sort_with_zero_impact_scores(self, concrete_recommender): """Cards with zero impact scores are sorted correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Zero", source=source, user_id="user1", impact_score=0 ), NewsCard( topic="Five", source=source, user_id="user1", impact_score=5 ), NewsCard( topic="Zero2", source=source, user_id="user1", impact_score=0 ), ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") assert result[0].topic == "Five" assert result[0].impact_score == 5 def test_sort_with_negative_boost(self, concrete_recommender): """Negative boost reduces effective score.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card_boosted = NewsCard( topic="Negative Boost", source=source, user_id="user1", impact_score=10, ) card_boosted.metadata["preference_boost"] = 0.1 # Very low boost card_normal = NewsCard( topic="Normal", source=source, user_id="user1", impact_score=5 ) recommender = concrete_recommender() result = recommender.sort_cards([card_boosted, card_normal], "user123") # Normal (5/10 * 1.0 = 0.5) > Negative Boost (10/10 * 0.1 = 0.1) assert result[0].topic == "Normal" def test_sort_with_very_large_boost(self, concrete_recommender): """Very large boost values work correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card_huge_boost = NewsCard( topic="Huge Boost", source=source, user_id="user1", impact_score=1 ) card_huge_boost.metadata["preference_boost"] = 100.0 card_high_score = NewsCard( topic="High Score", source=source, user_id="user1", impact_score=10 ) recommender = concrete_recommender() result = recommender.sort_cards( [card_high_score, card_huge_boost], "user123" ) # Huge Boost (1/10 * 100 = 10) > High Score (10/10 * 1 = 1) assert result[0].topic == "Huge Boost" def test_sort_preserves_card_objects(self, concrete_recommender): """Sorting returns the same card objects, not copies.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") original_card = NewsCard( topic="Original", source=source, user_id="user1", impact_score=5 ) recommender = concrete_recommender() result = recommender.sort_cards([original_card], "user123") assert result[0] is original_card # ============================================================================= # Additional progress callback tests # ============================================================================= class TestProgressCallbackAdditional: """Additional tests for progress callback functionality.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): self._update_progress("Starting", 0) self._update_progress("Processing", 50, {"step": "middle"}) self._update_progress("Complete", 100, {"step": "done"}) return [] return TestRecommender def test_callback_receives_all_updates(self, concrete_recommender): """Callback receives all progress updates in order.""" recommender = concrete_recommender() calls = [] def track_callback(message, progress, metadata): calls.append((message, progress, metadata)) recommender.set_progress_callback(track_callback) recommender.generate_recommendations("user123") assert len(calls) == 3 assert calls[0] == ("Starting", 0, {}) assert calls[1] == ("Processing", 50, {"step": "middle"}) assert calls[2] == ("Complete", 100, {"step": "done"}) def test_callback_can_be_replaced(self, concrete_recommender): """Progress callback can be replaced.""" recommender = concrete_recommender() first_callback = Mock() second_callback = Mock() recommender.set_progress_callback(first_callback) recommender.set_progress_callback(second_callback) recommender._update_progress("Test", 50) first_callback.assert_not_called() second_callback.assert_called_once() def test_callback_can_be_removed(self, concrete_recommender): """Progress callback can be set to None to remove it.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) recommender.set_progress_callback(None) # Should not raise recommender._update_progress("Test", 50) callback.assert_not_called() def test_update_progress_with_none_percent(self, concrete_recommender): """_update_progress works with None progress percent.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) recommender._update_progress("Status update", None, {"key": "value"}) callback.assert_called_once_with( "Status update", None, {"key": "value"} ) # ============================================================================= # Tests for get_strategy_info additional coverage # ============================================================================= class TestGetStrategyInfoAdditional: """Additional tests for get_strategy_info.""" def test_description_from_docstring(self): """get_strategy_info uses class docstring for description.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class DocumentedRecommender(BaseRecommender): """This is a custom recommendation strategy for testing.""" def generate_recommendations(self, user_id, context=None): return [] recommender = DocumentedRecommender() info = recommender.get_strategy_info() assert ( info["description"] == "This is a custom recommendation strategy for testing." ) def test_description_fallback_when_no_docstring(self): """get_strategy_info uses fallback when no docstring.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class UndocumentedRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] # Remove docstring UndocumentedRecommender.__doc__ = None recommender = UndocumentedRecommender() info = recommender.get_strategy_info() assert info["description"] == "No description available" # ============================================================================= # Integration tests for recommender # ============================================================================= class TestRecommenderIntegration: """Integration tests for recommender workflow.""" @pytest.fixture def full_recommender(self): """Create a recommender with all dependencies.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class FullRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): # Get preferences prefs = self._get_user_preferences(user_id) # Get ratings (exercises the method, result not used in this test) self._get_user_ratings(user_id, limit=10) # Create some cards from local_deep_research.news.core.base_card import ( NewsCard, CardSource, ) source = CardSource(type="recommendation") cards = [ NewsCard( topic="AI News", source=source, user_id=user_id, category="Technology", impact_score=8, ), NewsCard( topic="Sports Update", source=source, user_id=user_id, category="Sports", impact_score=6, ), NewsCard( topic="Weather Report", source=source, user_id=user_id, category="Weather", impact_score=4, ), ] # Filter and sort filtered = self._filter_by_preferences(cards, prefs) sorted_cards = self._sort_by_relevance(filtered, user_id) return sorted_cards return FullRecommender def test_full_workflow_without_preferences(self, full_recommender): """Test complete workflow without user preferences.""" mock_pref_manager = Mock() mock_pref_manager.get_preferences.return_value = {} mock_rating_system = Mock() mock_rating_system.get_recent_ratings.return_value = [] recommender = full_recommender( preference_manager=mock_pref_manager, rating_system=mock_rating_system, ) result = recommender.generate_recommendations("user123") # All cards returned, sorted by impact score assert len(result) == 3 assert result[0].impact_score == 8 # AI News assert result[1].impact_score == 6 # Sports assert result[2].impact_score == 4 # Weather def test_full_workflow_with_category_boost(self, full_recommender): """Test workflow with category boost preference.""" mock_pref_manager = Mock() mock_pref_manager.get_preferences.return_value = { "liked_categories": ["Sports"] } recommender = full_recommender(preference_manager=mock_pref_manager) result = recommender.generate_recommendations("user123") # Sports gets 1.2x boost: 6/10 * 1.2 = 0.72 # AI: 8/10 * 1.0 = 0.8 # Sports should still be second, but has boost assert len(result) == 3 assert result[0].topic == "AI News" # Still highest sports_card = [c for c in result if c.topic == "Sports Update"][0] assert sports_card.metadata.get("preference_boost") == 1.2 def test_full_workflow_with_impact_filter(self, full_recommender): """Test workflow with impact threshold filter.""" mock_pref_manager = Mock() mock_pref_manager.get_preferences.return_value = {"impact_threshold": 5} recommender = full_recommender(preference_manager=mock_pref_manager) result = recommender.generate_recommendations("user123") # Weather (4) filtered out assert len(result) == 2 topics = [c.topic for c in result] assert "Weather Report" not in topics def test_full_workflow_with_disliked_topics(self, full_recommender): """Test workflow with disliked topics filter.""" mock_pref_manager = Mock() mock_pref_manager.get_preferences.return_value = { "disliked_topics": ["sports"] } recommender = full_recommender(preference_manager=mock_pref_manager) result = recommender.generate_recommendations("user123") # Sports filtered out assert len(result) == 2 topics = [c.topic for c in result] assert "Sports Update" not in topics def test_full_workflow_with_all_filters(self, full_recommender): """Test workflow with all filter types combined.""" mock_pref_manager = Mock() mock_pref_manager.get_preferences.return_value = { "liked_categories": ["Technology"], "impact_threshold": 5, "disliked_topics": ["weather"], } recommender = full_recommender(preference_manager=mock_pref_manager) result = recommender.generate_recommendations("user123") # Weather filtered by disliked topics # Low impact items filtered by threshold # Technology boosted assert len(result) == 2 # AI News should be first (boosted and high impact) assert result[0].topic == "AI News" assert result[0].metadata.get("preference_boost") == 1.2 # ============================================================================= # Tests for edge cases with NewsCard creation # ============================================================================= class TestNewsCardEdgeCases: """Tests for edge cases when working with NewsCard objects.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_filter_with_none_category(self, concrete_recommender): """Filter handles cards with None category.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="No category", source=source, user_id="user1", category=None, impact_score=5, ) recommender = concrete_recommender() preferences = {"liked_categories": ["Technology"]} # Should not raise result = recommender.filter_cards([card], preferences) assert len(result) == 1 assert "preference_boost" not in result[0].metadata def test_sort_handles_cards_with_existing_metadata( self, concrete_recommender ): """Sort handles cards that already have metadata.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="With metadata", source=source, user_id="user1", impact_score=5, ) card.metadata["existing_key"] = "existing_value" card.metadata["preference_boost"] = 2.0 recommender = concrete_recommender() result = recommender.sort_cards([card], "user123") # Original metadata preserved assert result[0].metadata["existing_key"] == "existing_value" assert result[0].metadata["preference_boost"] == 2.0 # ============================================================================= # Parameterized tests for impact threshold # ============================================================================= class TestImpactThresholdParameterized: """Parameterized tests for impact threshold filtering.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender @pytest.fixture def cards_with_range_of_scores(self): """Create cards with impact scores from 1 to 10.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") return [ NewsCard( topic=f"Score {i}", source=source, user_id="user1", impact_score=i, ) for i in range(1, 11) ] @pytest.mark.parametrize( "threshold,expected_count", [ (1, 10), # All cards (5, 6), # Scores 5-10 (7, 4), # Scores 7-10 (10, 1), # Only score 10 (11, 0), # None pass ], ) def test_threshold_filters_correctly( self, concrete_recommender, cards_with_range_of_scores, threshold, expected_count, ): """Verify threshold filters out cards below it.""" recommender = concrete_recommender() preferences = {"impact_threshold": threshold} result = recommender.filter_cards( cards_with_range_of_scores, preferences ) assert len(result) == expected_count for card in result: assert card.impact_score >= threshold # ============================================================================= # Parameterized tests for preference boost # ============================================================================= class TestPreferenceBoostParameterized: """Parameterized tests for preference boost behavior.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender @pytest.mark.parametrize( "impact,boost,expected_score", [ (10, 1.0, 1.0), # 10/10 * 1.0 = 1.0 (10, 2.0, 2.0), # 10/10 * 2.0 = 2.0 (5, 1.0, 0.5), # 5/10 * 1.0 = 0.5 (5, 2.0, 1.0), # 5/10 * 2.0 = 1.0 (8, 1.5, 1.2), # 8/10 * 1.5 = 1.2 (0, 1.0, 0.0), # 0/10 * 1.0 = 0.0 (10, 0.0, 0.0), # 10/10 * 0.0 = 0.0 ], ) def test_score_calculation_formula( self, concrete_recommender, impact, boost, expected_score ): """Verify score = (impact/10) * boost formula.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=impact ) card.metadata["preference_boost"] = boost recommender = concrete_recommender() # Sort with single card to verify it processes correctly result = recommender.sort_cards([card], "user123") # Verify by checking the card is returned (score calculation happened) assert len(result) == 1 # ============================================================================= # Parameterized tests for disliked topics matching # ============================================================================= class TestDislikedTopicsParameterized: """Parameterized tests for disliked topics matching.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender @pytest.mark.parametrize( "card_topic,disliked_topics,should_be_filtered", [ ("AI in Healthcare", ["ai"], True), # Lowercase match ( "AI in Healthcare", ["AI"], False, ), # Case-sensitive (topic lowercased) ("Machine Learning News", ["machine"], True), # Partial match ("Deep Learning", ["learning"], True), # Word match ("Python Tips", ["java"], False), # No match ("JavaScript Guide", ["script"], True), # Substring match ("Data Science", ["data", "science"], True), # Multiple matches ("Weather Report", ["tech", "sports"], False), # No matches ], ) def test_topic_matching_behavior( self, concrete_recommender, card_topic, disliked_topics, should_be_filtered, ): """Verify topic matching with various patterns.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic=card_topic, source=source, user_id="user1", impact_score=5 ) recommender = concrete_recommender() preferences = {"disliked_topics": disliked_topics} result = recommender.filter_cards([card], preferences) if should_be_filtered: assert len(result) == 0, f"Expected {card_topic} to be filtered" else: assert len(result) == 1, f"Expected {card_topic} to NOT be filtered" # ============================================================================= # Tests for sorting stability # ============================================================================= class TestSortingStability: """Tests for sorting stability and correctness.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_sort_many_cards(self, concrete_recommender): """Sort handles many cards correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic=f"Card {i}", source=source, user_id="user1", impact_score=i % 10 + 1, # Scores 1-10 repeating ) for i in range(100) ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") assert len(result) == 100 # Verify descending order for i in range(len(result) - 1): assert result[i].impact_score >= result[i + 1].impact_score def test_sort_with_identical_cards(self, concrete_recommender): """Sort handles many identical cards.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Same", source=source, user_id="user1", impact_score=5, ) for _ in range(10) ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") assert len(result) == 10 def test_sort_with_mixed_boosts(self, concrete_recommender): """Sort correctly handles mixed boost values.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") # Create cards with various impact scores and boosts card1 = NewsCard( topic="Low Impact High Boost", source=source, user_id="user1", impact_score=2, ) card1.metadata["preference_boost"] = 5.0 # Score: 2/10 * 5 = 1.0 card2 = NewsCard( topic="High Impact No Boost", source=source, user_id="user1", impact_score=9, ) # Score: 9/10 * 1 = 0.9 card3 = NewsCard( topic="Medium Both", source=source, user_id="user1", impact_score=6, ) card3.metadata["preference_boost"] = 1.5 # Score: 6/10 * 1.5 = 0.9 recommender = concrete_recommender() result = recommender.sort_cards([card2, card1, card3], "user123") # card1 should be first (highest score: 1.0) assert result[0].topic == "Low Impact High Boost" # ============================================================================= # Tests for method chaining scenarios # ============================================================================= class TestMethodChaining: """Tests for realistic method chaining scenarios.""" @pytest.fixture def full_recommender(self): """Create recommender with all operations exposed.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class ChainableRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_and_sort(self, cards, preferences, user_id): """Chain filter and sort operations.""" filtered = self._filter_by_preferences(cards, preferences) sorted_cards = self._sort_by_relevance(filtered, user_id) return sorted_cards return ChainableRecommender def test_filter_then_sort(self, full_recommender): """Filter and sort chain works correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Sports News", source=source, user_id="user1", category="Sports", impact_score=9, ), NewsCard( topic="Tech News", source=source, user_id="user1", category="Technology", impact_score=7, ), NewsCard( topic="Low Impact Tech", source=source, user_id="user1", category="Technology", impact_score=3, ), ] recommender = full_recommender() preferences = { "liked_categories": ["Technology"], "impact_threshold": 5, } result = recommender.filter_and_sort(cards, preferences, "user123") # Low Impact Tech filtered out (score 3 < 5) assert len(result) == 2 # Tech News should be first (boosted and above threshold) # Score: 7/10 * 1.2 = 0.84 # Sports: 9/10 * 1.0 = 0.9 # But wait - Sports has higher base score without boost # Let's verify the order topics = [c.topic for c in result] assert "Low Impact Tech" not in topics def test_empty_after_all_filtered(self, full_recommender): """Empty result after all cards filtered is sorted correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Low Score", source=source, user_id="user1", impact_score=1, ), ] recommender = full_recommender() preferences = {"impact_threshold": 10} result = recommender.filter_and_sort(cards, preferences, "user123") assert result == [] # ============================================================================= # Tests for dependency injection patterns # ============================================================================= class TestDependencyInjection: """Tests for various dependency injection scenarios.""" def test_all_dependencies_none(self): """Recommender works with all dependencies as None.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class MinimalRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): # Exercise all methods to verify they don't crash self._get_user_preferences(user_id) self._get_user_ratings(user_id) self._execute_search("test") return [] recommender = MinimalRecommender() # All should work without crashing result = recommender.generate_recommendations("user123") assert result == [] def test_partial_dependencies(self): """Recommender works with some dependencies set.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) mock_pref_manager = Mock() mock_pref_manager.get_preferences.return_value = {"key": "value"} class PartialRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): prefs = self._get_user_preferences(user_id) ratings = self._get_user_ratings(user_id) # No rating system return [prefs, ratings] recommender = PartialRecommender(preference_manager=mock_pref_manager) result = recommender.generate_recommendations("user123") assert result[0] == {"key": "value"} # From preference manager assert result[1] == [] # Empty from missing rating system def test_dependencies_can_be_mocked(self): """All dependencies can be replaced with mocks.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) mock_pref = Mock() mock_pref.get_preferences.return_value = {"pref": True} mock_rating = Mock() mock_rating.get_recent_ratings.return_value = [{"rating": 5}] mock_search = Mock() mock_search.analyze_topic.return_value = {"results": ["item"]} mock_registry = Mock() mock_registry.get_topics.return_value = ["topic1"] class FullMockedRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] recommender = FullMockedRecommender( preference_manager=mock_pref, rating_system=mock_rating, search_system=mock_search, topic_registry=mock_registry, ) info = recommender.get_strategy_info() assert info["has_preference_manager"] is True assert info["has_rating_system"] is True assert info["has_search_system"] is True # ============================================================================= # Tests for card attribute edge cases # ============================================================================= class TestCardAttributeEdgeCases: """Tests for handling cards with unusual attribute values.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_filter_with_empty_topic(self, concrete_recommender): """Filter handles cards with empty topic string.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["test"]} result = recommender.filter_cards([card], preferences) # Empty topic should not match any disliked topic assert len(result) == 1 def test_filter_with_whitespace_topic(self, concrete_recommender): """Filter handles cards with whitespace-only topic.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic=" ", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["test"]} result = recommender.filter_cards([card], preferences) assert len(result) == 1 def test_sort_with_maximum_impact_score(self, concrete_recommender): """Sort handles maximum impact score correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Max Score", source=source, user_id="user1", impact_score=100, # Very high score ) recommender = concrete_recommender() result = recommender.sort_cards([card], "user123") assert len(result) == 1 assert result[0].impact_score == 100 def test_filter_with_special_chars_in_category(self, concrete_recommender): """Filter handles categories with special characters.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", category="Tech & Science", impact_score=5, ) recommender = concrete_recommender() preferences = {"liked_categories": ["Tech & Science"]} result = recommender.filter_cards([card], preferences) assert len(result) == 1 assert result[0].metadata.get("preference_boost") == 1.2 # ============================================================================= # Tests for preference edge cases # ============================================================================= class TestPreferenceEdgeCases: """Tests for unusual preference configurations.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender @pytest.fixture def sample_cards(self): """Create sample cards for testing.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") return [ NewsCard( topic="Test Card", source=source, user_id="user1", category="Technology", impact_score=5, ), ] def test_empty_string_in_disliked_topics( self, concrete_recommender, sample_cards ): """Empty string in disliked_topics matches everything.""" recommender = concrete_recommender() preferences = {"disliked_topics": [""]} result = recommender.filter_cards(sample_cards, preferences) # Empty string is in every topic.lower() assert len(result) == 0 def test_very_long_disliked_topic(self, concrete_recommender, sample_cards): """Very long disliked topic that won't match anything.""" recommender = concrete_recommender() preferences = {"disliked_topics": ["a" * 1000]} result = recommender.filter_cards(sample_cards, preferences) assert len(result) == 1 # No match def test_disliked_topic_with_special_regex_chars( self, concrete_recommender, sample_cards ): """Disliked topics with regex special chars work as literals.""" recommender = concrete_recommender() # These are regex special chars but should be treated as literals preferences = {"disliked_topics": ["[test]", ".*", "^$"]} result = recommender.filter_cards(sample_cards, preferences) # None of these regex patterns should match literal topic text assert len(result) == 1 def test_negative_impact_threshold( self, concrete_recommender, sample_cards ): """Negative impact threshold allows all cards.""" recommender = concrete_recommender() preferences = {"impact_threshold": -10} result = recommender.filter_cards(sample_cards, preferences) assert len(result) == 1 # All cards pass def test_very_high_impact_threshold( self, concrete_recommender, sample_cards ): """Very high impact threshold filters all cards.""" recommender = concrete_recommender() preferences = {"impact_threshold": 1000} result = recommender.filter_cards(sample_cards, preferences) assert len(result) == 0 def test_multiple_matching_categories(self, concrete_recommender): """Card matching multiple liked categories gets boost once.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", category="Technology", impact_score=5, ) recommender = concrete_recommender() # Category appears multiple times in list preferences = { "liked_categories": ["Technology", "Technology", "Science"] } result = recommender.filter_cards([card], preferences) # Should only get 1.2 boost, not compounded assert result[0].metadata.get("preference_boost") == 1.2 # ============================================================================= # Tests for rating system integration # ============================================================================= class TestRatingSystemIntegration: """Tests for rating system integration.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def get_ratings(self, user_id, limit=50): return self._get_user_ratings(user_id, limit) return TestRecommender def test_rating_system_called_with_correct_args(self, concrete_recommender): """Rating system is called with correct user_id and limit.""" mock_rating_system = Mock() mock_rating_system.get_recent_ratings.return_value = [] recommender = concrete_recommender(rating_system=mock_rating_system) recommender.get_ratings("user456", limit=25) mock_rating_system.get_recent_ratings.assert_called_once_with( "user456", 25 ) def test_rating_system_returns_complex_data(self, concrete_recommender): """Rating system can return complex rating data.""" mock_rating_system = Mock() mock_rating_system.get_recent_ratings.return_value = [ {"card_id": "c1", "rating": 5, "timestamp": "2024-01-15T12:00:00"}, {"card_id": "c2", "rating": 3, "feedback": "good article"}, { "card_id": "c3", "rating": 1, "disliked": True, "topics": ["spam"], }, ] recommender = concrete_recommender(rating_system=mock_rating_system) result = recommender.get_ratings("user123") assert len(result) == 3 assert result[0]["rating"] == 5 assert result[2]["disliked"] is True # ============================================================================= # Tests for search system edge cases # ============================================================================= class TestSearchSystemEdgeCases: """Tests for search system edge cases.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def search(self, query, strategy=None): return self._execute_search(query, strategy) return TestRecommender def test_search_with_empty_query(self, concrete_recommender): """Search handles empty query string.""" mock_search = Mock() mock_search.analyze_topic.return_value = {"results": []} recommender = concrete_recommender(search_system=mock_search) result = recommender.search("") mock_search.analyze_topic.assert_called_once_with("") assert result == {"results": []} def test_search_with_unicode_query(self, concrete_recommender): """Search handles unicode query string.""" mock_search = Mock() mock_search.analyze_topic.return_value = {"results": ["日本語"]} recommender = concrete_recommender(search_system=mock_search) result = recommender.search("日本語ニュース") mock_search.analyze_topic.assert_called_once_with("日本語ニュース") assert "日本語" in result["results"] def test_search_returns_large_result(self, concrete_recommender): """Search handles large result set.""" mock_search = Mock() mock_search.analyze_topic.return_value = { "results": [f"item_{i}" for i in range(1000)] } recommender = concrete_recommender(search_system=mock_search) result = recommender.search("test") assert len(result["results"]) == 1000 # ============================================================================= # Tests for complex filter combinations # ============================================================================= class TestComplexFilterCombinations: """Tests for complex combinations of filters.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender @pytest.fixture def diverse_cards(self): """Create diverse set of cards for testing.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") return [ NewsCard( topic="Breaking: AI Revolution", source=source, user_id="user1", category="Technology", impact_score=9, ), NewsCard( topic="Sports Update: Football Finals", source=source, user_id="user1", category="Sports", impact_score=7, ), NewsCard( topic="Weather: Storm Warning", source=source, user_id="user1", category="Weather", impact_score=4, ), NewsCard( topic="Finance: Stock Market Crash", source=source, user_id="user1", category="Finance", impact_score=8, ), NewsCard( topic="Tech News: Python Update", source=source, user_id="user1", category="Technology", impact_score=5, ), ] def test_filter_boost_then_threshold( self, concrete_recommender, diverse_cards ): """Boost is applied before threshold filtering.""" recommender = concrete_recommender() preferences = { "liked_categories": ["Technology"], "impact_threshold": 6, } result = recommender.filter_cards(diverse_cards, preferences) # Weather (4) and Python Update (5) filtered by threshold assert len(result) == 3 # AI Revolution should have boost ai_card = [c for c in result if "AI" in c.topic][0] assert ai_card.metadata.get("preference_boost") == 1.2 def test_filter_disliked_then_threshold( self, concrete_recommender, diverse_cards ): """Disliked topics and threshold both apply.""" recommender = concrete_recommender() preferences = { "disliked_topics": ["stock", "crash"], "impact_threshold": 5, } result = recommender.filter_cards(diverse_cards, preferences) # Finance filtered by disliked, Weather filtered by threshold assert len(result) == 3 topics = [c.topic for c in result] assert not any("Stock" in t or "Crash" in t for t in topics) def test_all_filters_combined(self, concrete_recommender, diverse_cards): """All filter types applied together.""" recommender = concrete_recommender() preferences = { "liked_categories": ["Technology"], "disliked_topics": ["weather", "storm"], "impact_threshold": 5, } result = recommender.filter_cards(diverse_cards, preferences) # Weather filtered (disliked + threshold) # Remaining: AI, Sports, Finance, Python # But Python also filtered by threshold (5 >= 5, so it passes) assert len(result) == 4 # Verify AI and Python have boosts tech_cards = [c for c in result if c.category == "Technology"] for card in tech_cards: assert card.metadata.get("preference_boost") == 1.2 def test_filter_then_sort_integration( self, concrete_recommender, diverse_cards ): """Filter and sort work together correctly.""" recommender = concrete_recommender() preferences = { "liked_categories": ["Finance"], "impact_threshold": 5, } filtered = recommender.filter_cards(diverse_cards, preferences) sorted_cards = recommender.sort_cards(filtered, "user123") # Verify sorted by score (impact/10 * boost) # AI: 9/10 * 1.0 = 0.9 # Sports: 7/10 * 1.0 = 0.7 # Finance: 8/10 * 1.2 = 0.96 (boosted) # Python: 5/10 * 1.0 = 0.5 assert ( sorted_cards[0].category == "Finance" ) # Highest score due to boost # ============================================================================= # Tests for strategy name inheritance # ============================================================================= class TestStrategyNameInheritance: """Tests for strategy name behavior across inheritance.""" def test_strategy_name_from_subclass(self): """Strategy name is derived from the actual subclass.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class MySpecialRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] recommender = MySpecialRecommender() assert recommender.strategy_name == "MySpecialRecommender" def test_strategy_name_in_nested_inheritance(self): """Strategy name works with nested inheritance.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class IntermediateRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] class FinalRecommender(IntermediateRecommender): pass recommender = FinalRecommender() assert recommender.strategy_name == "FinalRecommender" def test_strategy_info_includes_correct_name(self): """get_strategy_info returns the correct strategy name.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class AnalyticsRecommender(BaseRecommender): """Recommender using analytics data.""" def generate_recommendations(self, user_id, context=None): return [] recommender = AnalyticsRecommender() info = recommender.get_strategy_info() assert info["name"] == "AnalyticsRecommender" assert "analytics" in info["description"].lower() # ============================================================================= # Tests for context parameter handling # ============================================================================= class TestContextParameterHandling: """Tests for context parameter in generate_recommendations.""" def test_context_is_passed_to_implementation(self): """Context parameter is available in implementation.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class ContextAwareRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): # Return context for testing return [{"received_context": context}] recommender = ContextAwareRecommender() context = {"page": "home", "device": "mobile", "time": "morning"} result = recommender.generate_recommendations( "user123", context=context ) assert result[0]["received_context"] == context def test_context_defaults_to_none(self): """Context defaults to None when not provided.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class ContextCheckRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [{"context_was_none": context is None}] recommender = ContextCheckRecommender() result = recommender.generate_recommendations("user123") assert result[0]["context_was_none"] is True def test_context_with_complex_data(self): """Context can contain complex nested data.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class NestedContextRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [{"context": context}] recommender = NestedContextRecommender() context = { "user_history": [{"page": "news", "duration": 120}], "preferences": {"theme": "dark"}, "session": {"id": "abc123", "start": "2024-01-15T12:00:00"}, } result = recommender.generate_recommendations( "user123", context=context ) assert result[0]["context"]["user_history"][0]["duration"] == 120 assert result[0]["context"]["preferences"]["theme"] == "dark" # ============================================================================= # Stress tests for recommender # ============================================================================= class TestRecommenderStress: """Stress tests for recommender with large data sets.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_filter_1000_cards(self, concrete_recommender): """Filter handles 1000 cards efficiently.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic=f"News Item {i}", source=source, user_id="user1", category=f"Category{i % 10}", impact_score=i % 10 + 1, ) for i in range(1000) ] recommender = concrete_recommender() preferences = {"impact_threshold": 5} result = recommender.filter_cards(cards, preferences) # Cards with impact 5-10 should pass (600 cards) assert len(result) == 600 def test_sort_1000_cards(self, concrete_recommender): """Sort handles 1000 cards efficiently.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic=f"News Item {i}", source=source, user_id="user1", impact_score=i % 100, ) for i in range(1000) ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") assert len(result) == 1000 # Verify descending order for i in range(len(result) - 1): assert result[i].impact_score >= result[i + 1].impact_score def test_filter_and_sort_large_dataset(self, concrete_recommender): """Combined filter and sort on large dataset.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") categories = ["Tech", "Sports", "Finance", "Health", "Entertainment"] cards = [ NewsCard( topic=f"Article about {categories[i % 5]} topic {i}", source=source, user_id="user1", category=categories[i % 5], impact_score=(i % 10) + 1, ) for i in range(500) ] recommender = concrete_recommender() preferences = { "liked_categories": ["Tech", "Finance"], "impact_threshold": 3, } filtered = recommender.filter_cards(cards, preferences) sorted_cards = recommender.sort_cards(filtered, "user123") # Verify filtering worked for card in sorted_cards: assert card.impact_score >= 3 # Verify sorting worked for i in range(len(sorted_cards) - 1): score_i = ( sorted_cards[i].impact_score / 10.0 * sorted_cards[i].metadata.get("preference_boost", 1.0) ) score_next = ( sorted_cards[i + 1].impact_score / 10.0 * sorted_cards[i + 1].metadata.get("preference_boost", 1.0) ) assert score_i >= score_next # ============================================================================= # Tests for abstract method enforcement in recommender # ============================================================================= class TestRecommenderAbstractEnforcement: """Tests for abstract method enforcement in BaseRecommender.""" def test_cannot_instantiate_base_class(self): """BaseRecommender cannot be instantiated directly.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) with pytest.raises(TypeError) as exc_info: BaseRecommender() assert "abstract" in str(exc_info.value).lower() def test_must_implement_generate_recommendations(self): """Subclass must implement generate_recommendations.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class IncompleteRecommender(BaseRecommender): pass with pytest.raises(TypeError): IncompleteRecommender() # ============================================================================= # Tests for data validation in recommender # ============================================================================= class TestRecommenderDataValidation: """Tests for data validation in recommender operations.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_filter_returns_list(self, concrete_recommender): """Filter always returns a list.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, ) ] recommender = concrete_recommender() result = recommender.filter_cards(cards, {}) assert isinstance(result, list) result = recommender.filter_cards([], {}) assert isinstance(result, list) def test_sort_returns_list(self, concrete_recommender): """Sort always returns a list.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, ) ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") assert isinstance(result, list) result = recommender.sort_cards([], "user123") assert isinstance(result, list) def test_get_strategy_info_returns_dict(self, concrete_recommender): """get_strategy_info always returns a dict.""" recommender = concrete_recommender() result = recommender.get_strategy_info() assert isinstance(result, dict) def test_get_strategy_info_has_required_keys(self, concrete_recommender): """get_strategy_info contains all required keys.""" recommender = concrete_recommender() result = recommender.get_strategy_info() required_keys = [ "name", "has_preference_manager", "has_rating_system", "has_search_system", "description", ] for key in required_keys: assert key in result, f"Missing required key: {key}" # ============================================================================= # Tests for multiple users # ============================================================================= class TestMultipleUsers: """Tests for handling multiple users.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): prefs = self._get_user_preferences(user_id) return [{"user_id": user_id, "prefs": prefs}] return TestRecommender def test_different_users_get_different_preferences( self, concrete_recommender ): """Different users receive their own preferences.""" mock_pref_manager = Mock() mock_pref_manager.get_preferences.side_effect = lambda uid: { "user_id": uid, "setting": f"value_for_{uid}", } recommender = concrete_recommender(preference_manager=mock_pref_manager) result1 = recommender.generate_recommendations("user1") result2 = recommender.generate_recommendations("user2") assert result1[0]["prefs"]["user_id"] == "user1" assert result2[0]["prefs"]["user_id"] == "user2" assert result1[0]["prefs"]["setting"] == "value_for_user1" assert result2[0]["prefs"]["setting"] == "value_for_user2" def test_user_id_passed_to_sort(self, concrete_recommender): """User ID is passed to sort method.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TrackingRecommender(BaseRecommender): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.sort_user_ids = [] def generate_recommendations(self, user_id, context=None): return [] def sort_cards(self, cards, user_id): self.sort_user_ids.append(user_id) return self._sort_by_relevance(cards, user_id) from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Test", source=source, user_id="user1", impact_score=5 ) ] recommender = TrackingRecommender() recommender.sort_cards(cards, "user_abc") recommender.sort_cards(cards, "user_xyz") assert "user_abc" in recommender.sort_user_ids assert "user_xyz" in recommender.sort_user_ids # ============================================================================= # Tests for callback error handling # ============================================================================= class TestCallbackErrorHandling: """Tests for handling errors in callbacks.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): self._update_progress("Step 1", 50) return [] return TestRecommender def test_callback_exception_propagates(self, concrete_recommender): """Exception in callback propagates to caller.""" recommender = concrete_recommender() def failing_callback(msg, pct, meta): raise ValueError("Callback failed!") recommender.set_progress_callback(failing_callback) with pytest.raises(ValueError) as exc_info: recommender.generate_recommendations("user123") assert "Callback failed!" in str(exc_info.value) # ============================================================================= # Tests for category boost behavior # ============================================================================= class TestCategoryBoostBehavior: """Tests for category boost application in filtering.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class with filter method.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_boost_is_exactly_1_2(self, concrete_recommender): """Preference boost for liked categories is exactly 1.2.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category="tech", ) recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} result = recommender.filter_cards([card], preferences) assert result[0].metadata.get("preference_boost") == 1.2 def test_no_boost_for_non_liked_category(self, concrete_recommender): """Cards not in liked categories get no boost.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category="sports", ) recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} result = recommender.filter_cards([card], preferences) assert result[0].metadata.get("preference_boost") is None def test_multiple_liked_categories(self, concrete_recommender): """Multiple liked categories all receive boost.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Tech news", source=source, user_id="user1", impact_score=5, category="tech", ), NewsCard( topic="Science news", source=source, user_id="user1", impact_score=5, category="science", ), NewsCard( topic="Sports news", source=source, user_id="user1", impact_score=5, category="sports", ), ] recommender = concrete_recommender() preferences = {"liked_categories": ["tech", "science"]} result = recommender.filter_cards(cards, preferences) assert result[0].metadata.get("preference_boost") == 1.2 # tech assert result[1].metadata.get("preference_boost") == 1.2 # science assert result[2].metadata.get("preference_boost") is None # sports def test_empty_liked_categories_list(self, concrete_recommender): """Empty liked_categories list doesn't apply boost.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category="tech", ) recommender = concrete_recommender() preferences = {"liked_categories": []} result = recommender.filter_cards([card], preferences) # Empty list is falsy, so no boost should be applied assert result[0].metadata.get("preference_boost") is None # ============================================================================= # Tests for impact threshold filtering # ============================================================================= class TestImpactThresholdFiltering: """Tests for impact threshold filtering behavior.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class with filter method.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_filters_below_threshold(self, concrete_recommender): """Cards below threshold are filtered out.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Low impact", source=source, user_id="user1", impact_score=3, ), NewsCard( topic="High impact", source=source, user_id="user1", impact_score=7, ), ] recommender = concrete_recommender() preferences = {"impact_threshold": 5} result = recommender.filter_cards(cards, preferences) assert len(result) == 1 assert result[0].topic == "High impact" def test_threshold_is_inclusive(self, concrete_recommender): """Cards exactly at threshold are kept.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="At threshold", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"impact_threshold": 5} result = recommender.filter_cards([card], preferences) assert len(result) == 1 def test_threshold_zero_keeps_all(self, concrete_recommender): """Threshold of zero keeps all cards.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Zero impact", source=source, user_id="user1", impact_score=0, ), NewsCard( topic="Some impact", source=source, user_id="user1", impact_score=3, ), ] recommender = concrete_recommender() preferences = {"impact_threshold": 0} result = recommender.filter_cards(cards, preferences) assert len(result) == 2 # ============================================================================= # Tests for disliked topics filtering # ============================================================================= class TestDislikedTopicsFiltering: """Tests for disliked topics filtering behavior.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class with filter method.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_exact_match_filtered(self, concrete_recommender): """Exact topic match is filtered.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="politics", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["politics"]} result = recommender.filter_cards([card], preferences) assert len(result) == 0 def test_substring_match_filtered(self, concrete_recommender): """Substring match is filtered.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="US politics today", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["politics"]} result = recommender.filter_cards([card], preferences) assert len(result) == 0 def test_case_insensitive_matching(self, concrete_recommender): """Topic matching is case insensitive.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="POLITICS NEWS", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["politics"]} result = recommender.filter_cards([card], preferences) assert len(result) == 0 def test_multiple_disliked_topics(self, concrete_recommender): """Multiple disliked topics are all filtered.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Politics update", source=source, user_id="user1", impact_score=5, ), NewsCard( topic="Celebrity gossip", source=source, user_id="user1", impact_score=5, ), NewsCard( topic="Tech news", source=source, user_id="user1", impact_score=5, ), ] recommender = concrete_recommender() preferences = {"disliked_topics": ["politics", "celebrity"]} result = recommender.filter_cards(cards, preferences) assert len(result) == 1 assert result[0].topic == "Tech news" # ============================================================================= # Tests for sort score calculation # ============================================================================= class TestSortScoreCalculation: """Tests for the score calculation in sorting.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class with sort method.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_higher_impact_ranks_first(self, concrete_recommender): """Cards with higher impact score rank first.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Low impact", source=source, user_id="user1", impact_score=3, ), NewsCard( topic="High impact", source=source, user_id="user1", impact_score=9, ), NewsCard( topic="Medium impact", source=source, user_id="user1", impact_score=6, ), ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") assert result[0].topic == "High impact" assert result[1].topic == "Medium impact" assert result[2].topic == "Low impact" def test_boost_affects_ranking(self, concrete_recommender): """Preference boost affects final ranking.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") # Card with lower impact but boost card1 = NewsCard( topic="Boosted card", source=source, user_id="user1", impact_score=5, ) card1.metadata["preference_boost"] = 2.0 # Strong boost # Card with higher impact but no boost card2 = NewsCard( topic="Normal card", source=source, user_id="user1", impact_score=8, ) recommender = concrete_recommender() result = recommender.sort_cards([card1, card2], "user123") # 5/10 * 2.0 = 1.0 vs 8/10 * 1.0 = 0.8 assert result[0].topic == "Boosted card" assert result[1].topic == "Normal card" def test_score_formula(self, concrete_recommender): """Score formula is (impact_score / 10) * boost.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") # Create cards with known scores # Card A: 10/10 * 1.0 = 1.0 card_a = NewsCard( topic="A", source=source, user_id="user1", impact_score=10, ) # Card B: 5/10 * 1.5 = 0.75 card_b = NewsCard( topic="B", source=source, user_id="user1", impact_score=5, ) card_b.metadata["preference_boost"] = 1.5 # Card C: 6/10 * 1.2 = 0.72 card_c = NewsCard( topic="C", source=source, user_id="user1", impact_score=6, ) card_c.metadata["preference_boost"] = 1.2 recommender = concrete_recommender() result = recommender.sort_cards([card_b, card_c, card_a], "user123") # Expected order: A (1.0), B (0.75), C (0.72) assert result[0].topic == "A" assert result[1].topic == "B" assert result[2].topic == "C" # ============================================================================= # Tests for search system integration # ============================================================================= class TestSearchSystemIntegration: """Tests for search system integration.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def execute_search(self, query, strategy=None): return self._execute_search(query, strategy) return TestRecommender def test_uses_news_aggregation_by_default(self, concrete_recommender): """Search uses news_aggregation strategy by default.""" mock_search = Mock() mock_search.analyze_topic.return_value = {"results": []} recommender = concrete_recommender(search_system=mock_search) recommender.execute_search("test query") mock_search.analyze_topic.assert_called_once_with("test query") def test_handles_search_exception(self, concrete_recommender): """Search exception returns error dict.""" mock_search = Mock() mock_search.analyze_topic.side_effect = Exception("Search failed") recommender = concrete_recommender(search_system=mock_search) result = recommender.execute_search("test query") assert "error" in result def test_no_search_system_returns_error(self, concrete_recommender): """No search system returns error dict.""" recommender = concrete_recommender(search_system=None) result = recommender.execute_search("test query") assert "error" in result assert "not configured" in result["error"] # ============================================================================= # Tests for preference manager integration # ============================================================================= class TestPreferenceManagerIntegration: """Tests for preference manager integration.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def get_prefs(self, user_id): return self._get_user_preferences(user_id) return TestRecommender def test_calls_preference_manager_with_user_id(self, concrete_recommender): """Preference manager is called with correct user ID.""" mock_pref = Mock() mock_pref.get_preferences.return_value = {"setting": "value"} recommender = concrete_recommender(preference_manager=mock_pref) recommender.get_prefs("user_abc") mock_pref.get_preferences.assert_called_once_with("user_abc") def test_returns_preferences_from_manager(self, concrete_recommender): """Returns preferences from manager.""" mock_pref = Mock() mock_pref.get_preferences.return_value = { "liked_categories": ["tech"], "impact_threshold": 5, } recommender = concrete_recommender(preference_manager=mock_pref) result = recommender.get_prefs("user123") assert result["liked_categories"] == ["tech"] assert result["impact_threshold"] == 5 def test_no_preference_manager_returns_empty_dict( self, concrete_recommender ): """No preference manager returns empty dict.""" recommender = concrete_recommender(preference_manager=None) result = recommender.get_prefs("user123") assert result == {} # ============================================================================= # Tests for filter and sort pipeline # ============================================================================= class TestFilterSortPipeline: """Tests for the complete filter and sort pipeline.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def process_cards(self, cards, user_id, preferences): filtered = self._filter_by_preferences(cards, preferences) sorted_cards = self._sort_by_relevance(filtered, user_id) return sorted_cards return TestRecommender def test_complete_pipeline(self, concrete_recommender): """Complete filter and sort pipeline works correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Politics low", source=source, user_id="user1", impact_score=8, ), NewsCard( topic="Tech high", source=source, user_id="user1", impact_score=7, category="tech", ), NewsCard( topic="Science medium", source=source, user_id="user1", impact_score=6, ), NewsCard( topic="Low impact", source=source, user_id="user1", impact_score=2, ), ] recommender = concrete_recommender() preferences = { "liked_categories": ["tech"], "disliked_topics": ["politics"], "impact_threshold": 5, } result = recommender.process_cards(cards, "user123", preferences) # Should filter out: "Politics low" (disliked), "Low impact" (below threshold) # Should boost: "Tech high" (liked category) assert len(result) == 2 # Tech high with boost: 7/10 * 1.2 = 0.84 # Science medium: 6/10 * 1.0 = 0.6 assert result[0].topic == "Tech high" assert result[1].topic == "Science medium" def test_pipeline_with_empty_input(self, concrete_recommender): """Pipeline handles empty input.""" recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} result = recommender.process_cards([], "user123", preferences) assert result == [] def test_pipeline_with_empty_preferences(self, concrete_recommender): """Pipeline handles empty preferences.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Card 1", source=source, user_id="user1", impact_score=5, ), NewsCard( topic="Card 2", source=source, user_id="user1", impact_score=3, ), ] recommender = concrete_recommender() result = recommender.process_cards(cards, "user123", {}) # No filtering, just sorting by impact assert len(result) == 2 assert result[0].topic == "Card 1" assert result[1].topic == "Card 2" # ============================================================================= # Tests for rating system limit parameter # ============================================================================= class TestRatingSystemLimitParameter: """Tests for the limit parameter in _get_user_ratings.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def get_ratings(self, user_id, limit=50): return self._get_user_ratings(user_id, limit) return TestRecommender def test_default_limit_is_50(self, concrete_recommender): """Default limit is 50.""" mock_rating = Mock() mock_rating.get_recent_ratings.return_value = [] recommender = concrete_recommender(rating_system=mock_rating) recommender.get_ratings("user123") mock_rating.get_recent_ratings.assert_called_once_with("user123", 50) def test_custom_limit_is_passed(self, concrete_recommender): """Custom limit is passed to rating system.""" mock_rating = Mock() mock_rating.get_recent_ratings.return_value = [] recommender = concrete_recommender(rating_system=mock_rating) recommender.get_ratings("user123", limit=100) mock_rating.get_recent_ratings.assert_called_once_with("user123", 100) def test_small_limit(self, concrete_recommender): """Small limit of 1 works correctly.""" mock_rating = Mock() mock_rating.get_recent_ratings.return_value = [{"id": 1, "rating": 5}] recommender = concrete_recommender(rating_system=mock_rating) result = recommender.get_ratings("user123", limit=1) assert len(result) == 1 mock_rating.get_recent_ratings.assert_called_once_with("user123", 1) def test_zero_limit(self, concrete_recommender): """Zero limit works correctly.""" mock_rating = Mock() mock_rating.get_recent_ratings.return_value = [] recommender = concrete_recommender(rating_system=mock_rating) result = recommender.get_ratings("user123", limit=0) assert result == [] mock_rating.get_recent_ratings.assert_called_once_with("user123", 0) # ============================================================================= # Tests for topic registry dependency # ============================================================================= class TestTopicRegistryDependency: """Tests for topic registry dependency handling.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] return TestRecommender def test_topic_registry_stored(self, concrete_recommender): """Topic registry is stored correctly.""" mock_registry = Mock() recommender = concrete_recommender(topic_registry=mock_registry) assert recommender.topic_registry is mock_registry def test_topic_registry_none_by_default(self, concrete_recommender): """Topic registry is None by default.""" recommender = concrete_recommender() assert recommender.topic_registry is None def test_topic_registry_in_strategy_info(self, concrete_recommender): """Topic registry presence is reflected in strategy info.""" # Note: get_strategy_info doesn't include topic_registry currently # but has_search_system is there. This tests the pattern. mock_registry = Mock() recommender = concrete_recommender(topic_registry=mock_registry) info = recommender.get_strategy_info() # Verify the strategy info structure assert "has_search_system" in info # ============================================================================= # Tests for metadata handling in cards # ============================================================================= class TestMetadataHandlingInCards: """Tests for metadata handling when filtering cards.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class with filter method.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_existing_metadata_preserved(self, concrete_recommender): """Existing metadata is preserved when adding boost.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category="tech", ) card.metadata["existing_key"] = "existing_value" recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} result = recommender.filter_cards([card], preferences) assert result[0].metadata["existing_key"] == "existing_value" assert result[0].metadata["preference_boost"] == 1.2 def test_metadata_not_mutated_for_non_liked(self, concrete_recommender): """Metadata is not modified for non-liked categories.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category="sports", ) card.metadata["original"] = "value" recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} result = recommender.filter_cards([card], preferences) assert result[0].metadata == {"original": "value"} def test_boost_overwrites_existing_boost(self, concrete_recommender): """Preference boost overwrites any existing boost.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category="tech", ) card.metadata["preference_boost"] = 0.5 # Old boost recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} result = recommender.filter_cards([card], preferences) assert result[0].metadata["preference_boost"] == 1.2 # ============================================================================= # Tests for multiple preference combinations # ============================================================================= class TestMultiplePreferenceCombinations: """Tests for complex preference combinations.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_boost_then_dislike_filter(self, concrete_recommender): """Boosted card can still be filtered by disliked topics.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Tech Politics", source=source, user_id="user1", impact_score=5, category="tech", ) recommender = concrete_recommender() preferences = { "liked_categories": ["tech"], "disliked_topics": ["politics"], } result = recommender.filter_cards([card], preferences) # Should be filtered out despite being in liked category assert len(result) == 0 def test_boost_then_threshold_filter(self, concrete_recommender): """Boosted card can still be filtered by impact threshold.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Tech News", source=source, user_id="user1", impact_score=3, category="tech", ) recommender = concrete_recommender() preferences = { "liked_categories": ["tech"], "impact_threshold": 5, } result = recommender.filter_cards([card], preferences) # Boost is applied but card is still filtered by threshold assert len(result) == 0 def test_all_preferences_combined_pass(self, concrete_recommender): """Card passes all combined preference filters.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="AI Innovation", source=source, user_id="user1", impact_score=8, category="tech", ) recommender = concrete_recommender() preferences = { "liked_categories": ["tech"], "disliked_topics": ["politics", "sports"], "impact_threshold": 5, } result = recommender.filter_cards([card], preferences) assert len(result) == 1 assert result[0].metadata.get("preference_boost") == 1.2 # ============================================================================= # Tests for docstring in strategy info # ============================================================================= class TestDocstringInStrategyInfo: """Tests for docstring handling in get_strategy_info.""" def test_class_docstring_used(self): """Class docstring is used as description.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class DocumentedRecommender(BaseRecommender): """This is a custom recommender for testing purposes.""" def generate_recommendations(self, user_id, context=None): return [] recommender = DocumentedRecommender() info = recommender.get_strategy_info() assert "custom recommender" in info["description"].lower() def test_no_docstring_returns_default(self): """No docstring returns default message.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class UndocumentedRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] # Remove docstring explicitly UndocumentedRecommender.__doc__ = None recommender = UndocumentedRecommender() info = recommender.get_strategy_info() assert info["description"] == "No description available" # ============================================================================= # Tests for user_id in methods # ============================================================================= class TestUserIdInMethods: """Tests for user_id parameter handling in methods.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def get_prefs(self, user_id): return self._get_user_preferences(user_id) def get_ratings(self, user_id, limit=50): return self._get_user_ratings(user_id, limit) return TestRecommender def test_uuid_user_id_in_preferences(self, concrete_recommender): """UUID user_id is passed correctly to preferences.""" mock_pref = Mock() mock_pref.get_preferences.return_value = {} recommender = concrete_recommender(preference_manager=mock_pref) user_id = "550e8400-e29b-41d4-a716-446655440000" recommender.get_prefs(user_id) mock_pref.get_preferences.assert_called_once_with(user_id) def test_email_user_id_in_ratings(self, concrete_recommender): """Email user_id is passed correctly to ratings.""" mock_rating = Mock() mock_rating.get_recent_ratings.return_value = [] recommender = concrete_recommender(rating_system=mock_rating) user_id = "user@example.com" recommender.get_ratings(user_id) mock_rating.get_recent_ratings.assert_called_once_with(user_id, 50) def test_empty_user_id(self, concrete_recommender): """Empty user_id is handled without error.""" mock_pref = Mock() mock_pref.get_preferences.return_value = {} recommender = concrete_recommender(preference_manager=mock_pref) # Should not raise recommender.get_prefs("") mock_pref.get_preferences.assert_called_once_with("") # ============================================================================= # Tests for card ordering preservation # ============================================================================= class TestCardOrderingPreservation: """Tests for order preservation in filtering.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_order_preserved_with_no_filters(self, concrete_recommender): """Order is preserved when no filters apply.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic=f"Topic {i}", source=source, user_id="user1", impact_score=5, ) for i in range(10) ] recommender = concrete_recommender() result = recommender.filter_cards(cards, {}) # Order should be preserved for i, card in enumerate(result): assert card.topic == f"Topic {i}" def test_order_preserved_with_boost(self, concrete_recommender): """Order is preserved when applying category boost.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic=f"Topic {i}", source=source, user_id="user1", impact_score=5, category="tech", ) for i in range(5) ] recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} result = recommender.filter_cards(cards, preferences) # Order should be preserved for i, card in enumerate(result): assert card.topic == f"Topic {i}" # ============================================================================= # Tests for impact score edge values # ============================================================================= class TestImpactScoreEdgeValues: """Tests for edge values in impact score handling.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_zero_impact_score(self, concrete_recommender): """Zero impact score is handled correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Zero impact", source=source, user_id="user1", impact_score=0, ) recommender = concrete_recommender() result = recommender.sort_cards([card], "user123") assert len(result) == 1 def test_negative_impact_score_in_sort(self, concrete_recommender): """Negative impact score is handled in sorting.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Negative", source=source, user_id="user1", impact_score=-5, ), NewsCard( topic="Positive", source=source, user_id="user1", impact_score=5, ), ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") # Positive should come first assert result[0].topic == "Positive" assert result[1].topic == "Negative" def test_very_high_impact_score(self, concrete_recommender): """Very high impact score is handled correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="High impact", source=source, user_id="user1", impact_score=1000000, ) recommender = concrete_recommender() result = recommender.sort_cards([card], "user123") assert len(result) == 1 # ============================================================================= # Tests for float impact thresholds # ============================================================================= class TestFloatImpactThresholds: """Tests for float impact threshold values.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_float_threshold(self, concrete_recommender): """Float threshold works correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Below", source=source, user_id="user1", impact_score=5, ), NewsCard( topic="Above", source=source, user_id="user1", impact_score=6, ), ] recommender = concrete_recommender() preferences = {"impact_threshold": 5.5} result = recommender.filter_cards(cards, preferences) assert len(result) == 1 assert result[0].topic == "Above" # ============================================================================= # Tests for search result variations # ============================================================================= class TestSearchResultVariations: """Tests for handling various search result formats.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def execute_search(self, query, strategy=None): return self._execute_search(query, strategy) return TestRecommender def test_search_returns_dict(self, concrete_recommender): """Search returns dict results correctly.""" mock_search = Mock() mock_search.analyze_topic.return_value = { "results": ["r1", "r2"], "count": 2, } recommender = concrete_recommender(search_system=mock_search) result = recommender.execute_search("test") assert result["results"] == ["r1", "r2"] assert result["count"] == 2 def test_search_returns_empty_dict(self, concrete_recommender): """Search returns empty dict correctly.""" mock_search = Mock() mock_search.analyze_topic.return_value = {} recommender = concrete_recommender(search_system=mock_search) result = recommender.execute_search("test") assert result == {} def test_search_returns_nested_results(self, concrete_recommender): """Search returns nested results correctly.""" mock_search = Mock() mock_search.analyze_topic.return_value = { "data": { "articles": [{"title": "A"}, {"title": "B"}], "metadata": {"total": 100}, } } recommender = concrete_recommender(search_system=mock_search) result = recommender.execute_search("test") assert result["data"]["articles"][0]["title"] == "A" assert result["data"]["metadata"]["total"] == 100 def test_search_with_none_result(self, concrete_recommender): """Search handles None result from analyze_topic.""" mock_search = Mock() mock_search.analyze_topic.return_value = None recommender = concrete_recommender(search_system=mock_search) result = recommender.execute_search("test") assert result is None # ============================================================================= # Tests for progress callback variations # ============================================================================= class TestProgressCallbackVariations: """Tests for various progress callback scenarios.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def send_progress(self, msg, pct, meta=None): self._update_progress(msg, pct, meta) return TestRecommender def test_callback_with_none_percent(self, concrete_recommender): """Callback handles None percent value.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) recommender.send_progress("Message", None, {"key": "value"}) callback.assert_called_once_with("Message", None, {"key": "value"}) def test_callback_with_zero_percent(self, concrete_recommender): """Callback handles 0 percent value.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) recommender.send_progress("Starting", 0, {}) callback.assert_called_once_with("Starting", 0, {}) def test_callback_with_100_percent(self, concrete_recommender): """Callback handles 100 percent value.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) recommender.send_progress("Complete", 100, {}) callback.assert_called_once_with("Complete", 100, {}) def test_callback_called_multiple_times(self, concrete_recommender): """Callback can be called multiple times.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) recommender.send_progress("Step 1", 25) recommender.send_progress("Step 2", 50) recommender.send_progress("Step 3", 75) recommender.send_progress("Done", 100) assert callback.call_count == 4 def test_callback_can_be_changed(self, concrete_recommender): """Callback can be replaced with a new one.""" recommender = concrete_recommender() callback1 = Mock() callback2 = Mock() recommender.set_progress_callback(callback1) recommender.send_progress("First", 50) recommender.set_progress_callback(callback2) recommender.send_progress("Second", 75) callback1.assert_called_once() callback2.assert_called_once() def test_callback_can_be_cleared(self, concrete_recommender): """Callback can be cleared by setting to None.""" recommender = concrete_recommender() callback = Mock() recommender.set_progress_callback(callback) recommender.send_progress("First", 50) recommender.set_progress_callback(None) recommender.send_progress("Second", 75) # Should not raise callback.assert_called_once() # ============================================================================= # Tests for category matching behavior # ============================================================================= class TestCategoryMatchingBehavior: """Tests for category matching in filtering.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_category_exact_match_required(self, concrete_recommender): """Category matching requires exact match (not substring).""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category="technology", ) recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} # Partial match result = recommender.filter_cards([card], preferences) # Should NOT boost because "tech" != "technology" assert result[0].metadata.get("preference_boost") is None def test_category_case_sensitive(self, concrete_recommender): """Category matching is case-sensitive.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category="Tech", ) recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} # Different case result = recommender.filter_cards([card], preferences) # Should NOT boost because "Tech" != "tech" assert result[0].metadata.get("preference_boost") is None def test_category_none_handling(self, concrete_recommender): """Cards with None category are not boosted.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category=None, ) recommender = concrete_recommender() preferences = {"liked_categories": ["tech"]} result = recommender.filter_cards([card], preferences) assert result[0].metadata.get("preference_boost") is None # ============================================================================= # Tests for disliked topic edge cases # ============================================================================= class TestDislikedTopicEdgeCases: """Tests for edge cases in disliked topic filtering.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_disliked_topic_at_start(self, concrete_recommender): """Disliked topic at start of card topic is filtered.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Politics: Today's News", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["politics"]} result = recommender.filter_cards([card], preferences) assert len(result) == 0 def test_disliked_topic_at_end(self, concrete_recommender): """Disliked topic at end of card topic is filtered.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Breaking News in Politics", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["politics"]} result = recommender.filter_cards([card], preferences) assert len(result) == 0 def test_disliked_topic_in_middle(self, concrete_recommender): """Disliked topic in middle of card topic is filtered.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Today's Politics Update", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["politics"]} result = recommender.filter_cards([card], preferences) assert len(result) == 0 def test_disliked_topic_with_punctuation(self, concrete_recommender): """Disliked topic adjacent to punctuation is filtered.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Politics!", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["politics"]} result = recommender.filter_cards([card], preferences) assert len(result) == 0 # ============================================================================= # Tests for multiple recommender instances # ============================================================================= class TestMultipleRecommenderInstances: """Tests for multiple recommender instances.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] return TestRecommender def test_instances_are_independent(self, concrete_recommender): """Multiple instances have independent state.""" mock_pref1 = Mock() mock_pref1.get_preferences.return_value = {"user": "1"} mock_pref2 = Mock() mock_pref2.get_preferences.return_value = {"user": "2"} r1 = concrete_recommender(preference_manager=mock_pref1) r2 = concrete_recommender(preference_manager=mock_pref2) prefs1 = r1._get_user_preferences("user") prefs2 = r2._get_user_preferences("user") assert prefs1["user"] == "1" assert prefs2["user"] == "2" def test_callback_is_instance_specific(self, concrete_recommender): """Callbacks are instance-specific.""" r1 = concrete_recommender() r2 = concrete_recommender() callback1 = Mock() callback2 = Mock() r1.set_progress_callback(callback1) r2.set_progress_callback(callback2) r1._update_progress("R1", 50) r2._update_progress("R2", 75) callback1.assert_called_once_with("R1", 50, {}) callback2.assert_called_once_with("R2", 75, {}) # ============================================================================= # Tests for recommendation generation contract # ============================================================================= class TestRecommendationGenerationContract: """Tests for the generate_recommendations contract.""" def test_must_accept_user_id(self): """generate_recommendations must accept user_id parameter.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class ValidRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [{"user": user_id}] recommender = ValidRecommender() result = recommender.generate_recommendations("user123") assert result[0]["user"] == "user123" def test_context_is_optional(self): """generate_recommendations context parameter is optional.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class ValidRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [{"context": context}] recommender = ValidRecommender() # Without context result1 = recommender.generate_recommendations("user123") assert result1[0]["context"] is None # With context result2 = recommender.generate_recommendations( "user123", {"key": "value"} ) assert result2[0]["context"]["key"] == "value" def test_can_return_empty_list(self): """generate_recommendations can return empty list.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class EmptyRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] recommender = EmptyRecommender() result = recommender.generate_recommendations("user123") assert result == [] def test_can_return_news_cards(self): """generate_recommendations can return NewsCard objects.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) from local_deep_research.news.core.base_card import NewsCard, CardSource class CardRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): source = CardSource(type="recommendation") return [ NewsCard( topic="Recommended", source=source, user_id=user_id, impact_score=8, ) ] recommender = CardRecommender() result = recommender.generate_recommendations("user123") assert len(result) == 1 assert result[0].topic == "Recommended" assert result[0].user_id == "user123" # ============================================================================= # Tests for boost value preservation # ============================================================================= class TestBoostValuePreservation: """Tests for boost value handling through the pipeline.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_boost_preserved_through_threshold_filter( self, concrete_recommender ): """Boost is preserved even when threshold filter removes other cards.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Low", source=source, user_id="user1", impact_score=3, category="tech", ), NewsCard( topic="High", source=source, user_id="user1", impact_score=8, category="tech", ), ] recommender = concrete_recommender() preferences = { "liked_categories": ["tech"], "impact_threshold": 5, } result = recommender.filter_cards(cards, preferences) # Only "High" remains, and it should have the boost assert len(result) == 1 assert result[0].metadata.get("preference_boost") == 1.2 def test_boost_affects_sort_order(self, concrete_recommender): """Boost affects the final sort order.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="No boost high impact", source=source, user_id="user1", impact_score=9, category="sports", ), NewsCard( topic="Boosted lower impact", source=source, user_id="user1", impact_score=8, category="tech", ), ] recommender = concrete_recommender() # Apply boost first preferences = {"liked_categories": ["tech"]} filtered = recommender.filter_cards(cards, preferences) # Then sort sorted_cards = recommender.sort_cards(filtered, "user123") # 8/10 * 1.2 = 0.96 > 9/10 * 1.0 = 0.9 assert sorted_cards[0].topic == "Boosted lower impact" assert sorted_cards[1].topic == "No boost high impact" # ============================================================================= # Tests for special characters in card data # ============================================================================= class TestSpecialCharactersInCards: """Tests for handling special characters in card data.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_topic_with_unicode(self, concrete_recommender): """Card with unicode topic is handled correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="日本語ニュース 🎉", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() result = recommender.filter_cards([card], {}) assert len(result) == 1 assert result[0].topic == "日本語ニュース 🎉" def test_topic_with_newlines(self, concrete_recommender): """Card with newlines in topic is handled correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Line 1\nLine 2\nLine 3", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() result = recommender.sort_cards([card], "user123") assert len(result) == 1 def test_disliked_topic_with_unicode(self, concrete_recommender): """Disliked topic matching works with unicode.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="日本語ニュース", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["日本語"]} result = recommender.filter_cards([card], preferences) assert len(result) == 0 def test_category_with_special_chars(self, concrete_recommender): """Category with special characters is handled correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, category="tech & science", ) recommender = concrete_recommender() preferences = {"liked_categories": ["tech & science"]} result = recommender.filter_cards([card], preferences) assert result[0].metadata.get("preference_boost") == 1.2 # ============================================================================= # Tests for empty and whitespace handling # ============================================================================= class TestEmptyAndWhitespaceHandling: """Tests for handling empty strings and whitespace.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_empty_topic_in_filter(self, concrete_recommender): """Card with empty topic passes through filter.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() result = recommender.filter_cards([card], {}) assert len(result) == 1 def test_whitespace_only_topic(self, concrete_recommender): """Card with whitespace-only topic is handled.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic=" ", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() result = recommender.sort_cards([card], "user123") assert len(result) == 1 def test_empty_disliked_topics_list(self, concrete_recommender): """Empty disliked_topics list doesn't filter anything.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Any topic", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": []} result = recommender.filter_cards([card], preferences) assert len(result) == 1 def test_whitespace_in_disliked_topic(self, concrete_recommender): """Whitespace in disliked topic is matched literally.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="topic with spaces", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"disliked_topics": ["with spaces"]} result = recommender.filter_cards([card], preferences) assert len(result) == 0 # ============================================================================= # Tests for impact score at boundaries # ============================================================================= class TestImpactScoreBoundaries: """Tests for impact score boundary conditions.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_impact_score_zero_in_sort(self, concrete_recommender): """Zero impact score results in zero score.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Zero", source=source, user_id="user1", impact_score=0, ), NewsCard( topic="Positive", source=source, user_id="user1", impact_score=1, ), ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") assert result[0].topic == "Positive" assert result[1].topic == "Zero" def test_impact_score_exactly_at_threshold(self, concrete_recommender): """Impact score exactly at threshold passes filter.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="At threshold", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = {"impact_threshold": 5} result = recommender.filter_cards([card], preferences) assert len(result) == 1 def test_impact_score_just_below_threshold(self, concrete_recommender): """Impact score just below threshold is filtered.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Just below", source=source, user_id="user1", impact_score=4.99, ) recommender = concrete_recommender() preferences = {"impact_threshold": 5} result = recommender.filter_cards([card], preferences) assert len(result) == 0 def test_impact_score_decimal_precision(self, concrete_recommender): """Impact score handles decimal precision correctly.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Score 5.001", source=source, user_id="user1", impact_score=5.001, ), NewsCard( topic="Score 5.002", source=source, user_id="user1", impact_score=5.002, ), ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") # Higher score should come first assert result[0].topic == "Score 5.002" assert result[1].topic == "Score 5.001" # ============================================================================= # Tests for preference key variations # ============================================================================= class TestPreferenceKeyVariations: """Tests for various preference key scenarios.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) return TestRecommender def test_unknown_preference_keys_ignored(self, concrete_recommender): """Unknown preference keys are ignored.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = { "unknown_key": "value", "another_unknown": [1, 2, 3], } result = recommender.filter_cards([card], preferences) assert len(result) == 1 def test_none_preference_values(self, concrete_recommender): """None values in preferences are handled.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") card = NewsCard( topic="Test", source=source, user_id="user1", impact_score=5, ) recommender = concrete_recommender() preferences = { "liked_categories": None, "disliked_topics": None, } result = recommender.filter_cards([card], preferences) assert len(result) == 1 def test_mixed_preference_types(self, concrete_recommender): """Mixed preference types work together.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic="Tech News High", source=source, user_id="user1", impact_score=8, category="tech", ), NewsCard( topic="Sports Low", source=source, user_id="user1", impact_score=3, category="sports", ), NewsCard( topic="Politics High", source=source, user_id="user1", impact_score=9, category="politics", ), ] recommender = concrete_recommender() preferences = { "liked_categories": ["tech"], "disliked_topics": ["politics"], "impact_threshold": 5, } result = recommender.filter_cards(cards, preferences) # Only "Tech News High" should remain assert len(result) == 1 assert result[0].topic == "Tech News High" assert result[0].metadata.get("preference_boost") == 1.2 # ============================================================================= # Tests for large datasets # ============================================================================= class TestLargeDatasets: """Tests for handling large numbers of cards.""" @pytest.fixture def concrete_recommender(self): """Create a concrete recommender class.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] def filter_cards(self, cards, preferences): return self._filter_by_preferences(cards, preferences) def sort_cards(self, cards, user_id): return self._sort_by_relevance(cards, user_id) return TestRecommender def test_filter_10000_cards(self, concrete_recommender): """Filter handles 10000 cards.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic=f"Topic {i}", source=source, user_id="user1", impact_score=i % 10, category="tech" if i % 2 == 0 else "sports", ) for i in range(10000) ] recommender = concrete_recommender() preferences = { "liked_categories": ["tech"], "impact_threshold": 5, } result = recommender.filter_cards(cards, preferences) # Should filter correctly assert all(card.impact_score >= 5 for card in result) def test_sort_10000_cards(self, concrete_recommender): """Sort handles 10000 cards.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic=f"Topic {i}", source=source, user_id="user1", impact_score=i % 100, ) for i in range(10000) ] recommender = concrete_recommender() result = recommender.sort_cards(cards, "user123") # Should be sorted in descending order for i in range(len(result) - 1): assert result[i].impact_score >= result[i + 1].impact_score def test_filter_all_cards_removed(self, concrete_recommender): """Filter can remove all cards.""" from local_deep_research.news.core.base_card import NewsCard, CardSource source = CardSource(type="test") cards = [ NewsCard( topic=f"Politics {i}", source=source, user_id="user1", impact_score=5, ) for i in range(100) ] recommender = concrete_recommender() preferences = {"disliked_topics": ["politics"]} result = recommender.filter_cards(cards, preferences) assert len(result) == 0 # ============================================================================= # Tests for strategy info completeness # ============================================================================= class TestStrategyInfoCompleteness: """Tests for get_strategy_info completeness.""" def test_strategy_info_with_all_dependencies(self): """Strategy info reflects all dependencies.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): """A test recommender for strategy info.""" def generate_recommendations(self, user_id, context=None): return [] mock_pref = Mock() mock_rating = Mock() mock_search = Mock() recommender = TestRecommender( preference_manager=mock_pref, rating_system=mock_rating, search_system=mock_search, ) info = recommender.get_strategy_info() assert info["has_preference_manager"] is True assert info["has_rating_system"] is True assert info["has_search_system"] is True def test_strategy_info_with_no_dependencies(self): """Strategy info reflects no dependencies.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class TestRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] recommender = TestRecommender() info = recommender.get_strategy_info() assert info["has_preference_manager"] is False assert info["has_rating_system"] is False assert info["has_search_system"] is False def test_strategy_info_name_matches_class(self): """Strategy info name matches class name.""" from local_deep_research.news.recommender.base_recommender import ( BaseRecommender, ) class MyCustomRecommender(BaseRecommender): def generate_recommendations(self, user_id, context=None): return [] recommender = MyCustomRecommender() info = recommender.get_strategy_info() assert info["name"] == "MyCustomRecommender"