""" Extended tests for news/preference_manager/base_preference.py Tests cover: - BasePreferenceManager abstract class - add_interest() - interest management - remove_interest() - interest removal - ignore_topic() - topic ignore list - boost_source() - source boosting - get_default_preferences() - default preferences - TopicRegistry class - extract_topics() - topic extraction - register_topic() - topic registration - get_trending_topics() - trending topics - get_topic_info() - topic information """ from unittest.mock import MagicMock from datetime import timedelta class ConcretePreferenceManager: """Concrete implementation for testing abstract class.""" def __init__(self, storage_backend=None): from local_deep_research.news.preference_manager.base_preference import ( BasePreferenceManager, ) # Create a test subclass class TestManager(BasePreferenceManager): def __init__(self, storage): super().__init__(storage) self._prefs = {} def get_preferences(self, user_id): return self._prefs.get(user_id, self.get_default_preferences()) def update_preferences(self, user_id, preferences): self._prefs[user_id] = preferences return preferences self.manager = TestManager(storage_backend) class TestBasePreferenceManagerInit: """Tests for BasePreferenceManager initialization.""" def test_stores_storage_backend(self): """Stores the storage backend.""" mock_storage = MagicMock() helper = ConcretePreferenceManager(mock_storage) assert helper.manager.storage_backend is mock_storage def test_none_storage_backend(self): """Accepts None as storage backend.""" helper = ConcretePreferenceManager(None) assert helper.manager.storage_backend is None class TestAddInterest: """Tests for BasePreferenceManager.add_interest().""" def test_adds_interest_to_empty_prefs(self): """Adds interest when no interests exist.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "technology") prefs = manager.get_preferences("user1") assert "technology" in prefs["interests"] def test_adds_interest_with_weight(self): """Adds interest with specified weight.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "AI", weight=1.5) prefs = manager.get_preferences("user1") assert prefs["interests"]["AI"] == 1.5 def test_default_weight_is_1(self): """Default weight is 1.0.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "science") prefs = manager.get_preferences("user1") assert prefs["interests"]["science"] == 1.0 def test_overwrites_existing_interest(self): """Overwrites weight for existing interest.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "tech", weight=1.0) manager.add_interest("user1", "tech", weight=2.0) prefs = manager.get_preferences("user1") assert prefs["interests"]["tech"] == 2.0 def test_updates_interests_updated_at(self): """Updates interests_updated_at timestamp.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "test") prefs = manager.get_preferences("user1") assert "interests_updated_at" in prefs def test_multiple_interests(self): """Can add multiple interests.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "AI") manager.add_interest("user1", "ML") manager.add_interest("user1", "robotics") prefs = manager.get_preferences("user1") assert len(prefs["interests"]) == 3 class TestRemoveInterest: """Tests for BasePreferenceManager.remove_interest().""" def test_removes_existing_interest(self): """Removes an existing interest.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "technology") manager.remove_interest("user1", "technology") prefs = manager.get_preferences("user1") assert "technology" not in prefs["interests"] def test_no_error_for_nonexistent_interest(self): """Does not error when interest doesn't exist.""" helper = ConcretePreferenceManager() manager = helper.manager # Should not raise manager.remove_interest("user1", "nonexistent") def test_updates_interests_updated_at(self): """Updates interests_updated_at when removing.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "tech") prefs_before = manager.get_preferences("user1") timestamp_before = prefs_before["interests_updated_at"] import time time.sleep(0.01) manager.remove_interest("user1", "tech") prefs_after = manager.get_preferences("user1") assert prefs_after["interests_updated_at"] >= timestamp_before def test_preserves_other_interests(self): """Preserves other interests when removing one.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "AI") manager.add_interest("user1", "ML") manager.remove_interest("user1", "AI") prefs = manager.get_preferences("user1") assert "AI" not in prefs["interests"] assert "ML" in prefs["interests"] class TestIgnoreTopic: """Tests for BasePreferenceManager.ignore_topic().""" def test_adds_topic_to_disliked_list(self): """Adds topic to disliked_topics list.""" helper = ConcretePreferenceManager() manager = helper.manager manager.ignore_topic("user1", "politics") prefs = manager.get_preferences("user1") assert "politics" in prefs["disliked_topics"] def test_does_not_add_duplicate(self): """Does not add duplicate topics.""" helper = ConcretePreferenceManager() manager = helper.manager manager.ignore_topic("user1", "sports") manager.ignore_topic("user1", "sports") prefs = manager.get_preferences("user1") assert prefs["disliked_topics"].count("sports") == 1 def test_updates_preferences_updated_at(self): """Updates preferences_updated_at timestamp.""" helper = ConcretePreferenceManager() manager = helper.manager manager.ignore_topic("user1", "topic") prefs = manager.get_preferences("user1") assert "preferences_updated_at" in prefs def test_multiple_ignored_topics(self): """Can ignore multiple topics.""" helper = ConcretePreferenceManager() manager = helper.manager manager.ignore_topic("user1", "topic1") manager.ignore_topic("user1", "topic2") manager.ignore_topic("user1", "topic3") prefs = manager.get_preferences("user1") assert len(prefs["disliked_topics"]) == 3 class TestBoostSource: """Tests for BasePreferenceManager.boost_source().""" def test_adds_source_weight(self): """Adds source to source_weights.""" helper = ConcretePreferenceManager() manager = helper.manager manager.boost_source("user1", "reuters.com") prefs = manager.get_preferences("user1") assert "reuters.com" in prefs["source_weights"] def test_default_weight_is_1_5(self): """Default boost weight is 1.5.""" helper = ConcretePreferenceManager() manager = helper.manager manager.boost_source("user1", "example.com") prefs = manager.get_preferences("user1") assert prefs["source_weights"]["example.com"] == 1.5 def test_custom_weight(self): """Accepts custom weight.""" helper = ConcretePreferenceManager() manager = helper.manager manager.boost_source("user1", "news.com", weight=2.0) prefs = manager.get_preferences("user1") assert prefs["source_weights"]["news.com"] == 2.0 def test_overwrites_existing_weight(self): """Overwrites existing source weight.""" helper = ConcretePreferenceManager() manager = helper.manager manager.boost_source("user1", "site.com", weight=1.5) manager.boost_source("user1", "site.com", weight=3.0) prefs = manager.get_preferences("user1") assert prefs["source_weights"]["site.com"] == 3.0 def test_can_set_negative_weight(self): """Can set negative weight to demote source.""" helper = ConcretePreferenceManager() manager = helper.manager manager.boost_source("user1", "tabloid.com", weight=0.5) prefs = manager.get_preferences("user1") assert prefs["source_weights"]["tabloid.com"] == 0.5 class TestGetDefaultPreferences: """Tests for BasePreferenceManager.get_default_preferences().""" def test_returns_dict(self): """Returns a dictionary.""" helper = ConcretePreferenceManager() manager = helper.manager result = manager.get_default_preferences() assert isinstance(result, dict) def test_includes_liked_categories(self): """Includes liked_categories list.""" helper = ConcretePreferenceManager() manager = helper.manager result = manager.get_default_preferences() assert "liked_categories" in result assert isinstance(result["liked_categories"], list) def test_includes_disliked_categories(self): """Includes disliked_categories list.""" helper = ConcretePreferenceManager() manager = helper.manager result = manager.get_default_preferences() assert "disliked_categories" in result def test_includes_interests(self): """Includes interests dict.""" helper = ConcretePreferenceManager() manager = helper.manager result = manager.get_default_preferences() assert "interests" in result assert isinstance(result["interests"], dict) def test_includes_impact_threshold(self): """Includes default impact_threshold of 5.""" helper = ConcretePreferenceManager() manager = helper.manager result = manager.get_default_preferences() assert result["impact_threshold"] == 5 def test_includes_focus_preferences(self): """Includes focus_preferences dict.""" helper = ConcretePreferenceManager() manager = helper.manager result = manager.get_default_preferences() assert "focus_preferences" in result assert "breaking" in result["focus_preferences"] def test_includes_search_strategy(self): """Includes default search_strategy.""" helper = ConcretePreferenceManager() manager = helper.manager result = manager.get_default_preferences() assert result["search_strategy"] == "news_aggregation" def test_includes_timestamps(self): """Includes created_at and preferences_updated_at.""" helper = ConcretePreferenceManager() manager = helper.manager result = manager.get_default_preferences() assert "created_at" in result assert "preferences_updated_at" in result class TestTopicRegistryInit: """Tests for TopicRegistry initialization.""" def test_initializes_empty_topics(self): """Initializes with empty topics dict.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() assert registry.topics == {} def test_stores_llm_client(self): """Stores LLM client if provided.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) mock_llm = MagicMock() registry = TopicRegistry(llm_client=mock_llm) assert registry.llm_client is mock_llm def test_none_llm_client(self): """Accepts None as LLM client.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry(llm_client=None) assert registry.llm_client is None class TestTopicRegistryRegisterTopic: """Tests for TopicRegistry.register_topic().""" def test_adds_new_topic(self): """Adds new topic to registry.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("AI") assert "AI" in registry.topics def test_sets_first_seen(self): """Sets first_seen timestamp for new topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("topic") assert "first_seen" in registry.topics["topic"] def test_sets_count_to_1(self): """Sets count to 1 for new topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("topic") assert registry.topics["topic"]["count"] == 1 def test_increments_count_for_existing(self): """Increments count for existing topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("topic") registry.register_topic("topic") registry.register_topic("topic") assert registry.topics["topic"]["count"] == 3 def test_updates_last_seen(self): """Updates last_seen on each registration.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("topic") first_seen = registry.topics["topic"]["last_seen"] import time time.sleep(0.01) registry.register_topic("topic") second_seen = registry.topics["topic"]["last_seen"] assert second_seen >= first_seen class TestTopicRegistryGetTrendingTopics: """Tests for TopicRegistry.get_trending_topics().""" def test_returns_list(self): """Returns a list.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() result = registry.get_trending_topics() assert isinstance(result, list) def test_returns_empty_for_no_topics(self): """Returns empty list when no topics registered.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() result = registry.get_trending_topics() assert result == [] def test_filters_by_hours(self): """Filters topics by hours lookback.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) from local_deep_research.news.core.utils import utc_now registry = TopicRegistry() # Add recent topic registry.register_topic("recent") # Add old topic registry.topics["old"] = { "first_seen": utc_now() - timedelta(hours=48), "last_seen": utc_now() - timedelta(hours=48), "count": 5, } result = registry.get_trending_topics(hours=24) assert "recent" in result assert "old" not in result def test_sorts_by_count(self): """Sorts topics by count descending.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("low") for _ in range(5): registry.register_topic("high") for _ in range(2): registry.register_topic("medium") result = registry.get_trending_topics() assert result[0] == "high" assert result[1] == "medium" assert result[2] == "low" def test_respects_limit(self): """Respects the limit parameter.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() for i in range(20): registry.register_topic(f"topic{i}") result = registry.get_trending_topics(limit=5) assert len(result) == 5 def test_default_hours_is_24(self): """Default hours is 24.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) from local_deep_research.news.core.utils import utc_now registry = TopicRegistry() # Add topic from 25 hours ago registry.topics["old"] = { "first_seen": utc_now() - timedelta(hours=25), "last_seen": utc_now() - timedelta(hours=25), "count": 10, } result = registry.get_trending_topics() assert "old" not in result def test_default_limit_is_10(self): """Default limit is 10.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() for i in range(20): registry.register_topic(f"topic{i}") result = registry.get_trending_topics() assert len(result) == 10 class TestTopicRegistryGetTopicInfo: """Tests for TopicRegistry.get_topic_info().""" def test_returns_topic_data(self): """Returns topic data for existing topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("AI") result = registry.get_topic_info("AI") assert result is not None assert "count" in result assert "first_seen" in result def test_returns_none_for_unknown(self): """Returns None for unknown topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() result = registry.get_topic_info("unknown") assert result is None def test_includes_count(self): """Topic info includes count.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("topic") registry.register_topic("topic") result = registry.get_topic_info("topic") assert result["count"] == 2 def test_includes_first_seen(self): """Topic info includes first_seen.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("topic") result = registry.get_topic_info("topic") assert "first_seen" in result def test_includes_last_seen(self): """Topic info includes last_seen.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("topic") result = registry.get_topic_info("topic") assert "last_seen" in result class TestPreferenceManagerEdgeCases: """Edge case tests for BasePreferenceManager.""" def test_add_interest_with_zero_weight(self): """Adding interest with zero weight is allowed.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "topic", weight=0) prefs = manager.get_preferences("user1") assert prefs["interests"]["topic"] == 0 def test_add_interest_with_negative_weight(self): """Adding interest with negative weight is allowed.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "topic", weight=-1) prefs = manager.get_preferences("user1") assert prefs["interests"]["topic"] == -1 def test_add_interest_with_very_high_weight(self): """Adding interest with very high weight is allowed.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "topic", weight=1000) prefs = manager.get_preferences("user1") assert prefs["interests"]["topic"] == 1000 def test_remove_all_interests(self): """Can remove all interests from user.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "topic1") manager.add_interest("user1", "topic2") manager.add_interest("user1", "topic3") manager.remove_interest("user1", "topic1") manager.remove_interest("user1", "topic2") manager.remove_interest("user1", "topic3") prefs = manager.get_preferences("user1") assert len(prefs["interests"]) == 0 def test_preferences_isolated_between_users(self): """Preferences are isolated between users.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "topic_a") manager.add_interest("user2", "topic_b") prefs1 = manager.get_preferences("user1") prefs2 = manager.get_preferences("user2") assert "topic_a" in prefs1["interests"] assert "topic_a" not in prefs2["interests"] assert "topic_b" in prefs2["interests"] assert "topic_b" not in prefs1["interests"] def test_empty_user_id_handled(self): """Empty user_id is handled without error.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("", "topic") prefs = manager.get_preferences("") assert "topic" in prefs["interests"] def test_special_characters_in_topic(self): """Topics with special characters are handled.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "AI & ML") manager.add_interest("user1", "C++") manager.add_interest("user1", "email@domain.com") prefs = manager.get_preferences("user1") assert "AI & ML" in prefs["interests"] assert "C++" in prefs["interests"] assert "email@domain.com" in prefs["interests"] def test_unicode_topic_names(self): """Unicode topic names are handled.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "日本語") manager.add_interest("user1", "émoji 🎉") prefs = manager.get_preferences("user1") assert "日本語" in prefs["interests"] assert "émoji 🎉" in prefs["interests"] def test_very_long_topic_name(self): """Very long topic names are handled.""" helper = ConcretePreferenceManager() manager = helper.manager long_topic = "a" * 1000 manager.add_interest("user1", long_topic) prefs = manager.get_preferences("user1") assert long_topic in prefs["interests"] class TestTopicRegistryEdgeCases: """Edge case tests for TopicRegistry.""" def test_register_empty_topic_name(self): """Registering empty string topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("") assert "" in registry.topics def test_register_whitespace_topic(self): """Registering whitespace-only topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic(" ") assert " " in registry.topics def test_topics_case_sensitive(self): """Topics are case sensitive.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("AI") registry.register_topic("ai") registry.register_topic("Ai") assert "AI" in registry.topics assert "ai" in registry.topics assert "Ai" in registry.topics assert len(registry.topics) == 3 def test_get_trending_with_mixed_ages(self): """Get trending with mix of old and new topics.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) from local_deep_research.news.core.utils import utc_now registry = TopicRegistry() # Add recent topics registry.register_topic("recent1") registry.register_topic("recent2") # Add old topic directly registry.topics["old"] = { "first_seen": utc_now() - timedelta(hours=100), "last_seen": utc_now() - timedelta(hours=100), "count": 50, } result = registry.get_trending_topics(hours=24) assert "recent1" in result assert "recent2" in result assert "old" not in result def test_get_topic_info_preserves_first_seen(self): """First seen timestamp doesn't change on subsequent registrations.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("topic") first_seen_1 = registry.topics["topic"]["first_seen"] import time time.sleep(0.01) registry.register_topic("topic") first_seen_2 = registry.topics["topic"]["first_seen"] assert first_seen_1 == first_seen_2 def test_trending_topics_sorted_by_count(self): """Trending topics are sorted by count descending.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() # Register topics with different counts for _ in range(5): registry.register_topic("popular") for _ in range(3): registry.register_topic("medium") registry.register_topic("unpopular") result = registry.get_trending_topics(limit=3) # Should be sorted by count assert result[0] == "popular" def test_large_number_of_topics(self): """Handles large number of topics.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() for i in range(1000): registry.register_topic(f"topic_{i}") assert len(registry.topics) == 1000 result = registry.get_trending_topics(limit=50) assert len(result) == 50 class TestPreferenceManagerConcurrentUsers: """Tests for multiple user scenarios.""" def test_many_users_with_preferences(self): """Many users can have separate preferences.""" helper = ConcretePreferenceManager() manager = helper.manager for i in range(100): manager.add_interest(f"user{i}", f"topic{i}") for i in range(100): prefs = manager.get_preferences(f"user{i}") assert f"topic{i}" in prefs["interests"] def test_user_preference_modification_isolated(self): """Modifying one user's preferences doesn't affect others.""" helper = ConcretePreferenceManager() manager = helper.manager manager.add_interest("user1", "shared_topic") manager.add_interest("user2", "shared_topic") # Modify user1's preference manager.add_interest("user1", "shared_topic", weight=10) prefs1 = manager.get_preferences("user1") prefs2 = manager.get_preferences("user2") assert prefs1["interests"]["shared_topic"] == 10 assert prefs2["interests"]["shared_topic"] == 1 # Default weight class TestBoostSourceEdgeCases: """Edge cases for boost_source method.""" def test_boost_source_with_empty_string(self): """Can boost empty string source.""" helper = ConcretePreferenceManager() manager = helper.manager manager.boost_source("user1", "") prefs = manager.get_preferences("user1") assert "" in prefs["source_weights"] def test_boost_source_with_url(self): """Can boost full URL as source.""" helper = ConcretePreferenceManager() manager = helper.manager manager.boost_source("user1", "https://example.com/news") prefs = manager.get_preferences("user1") assert "https://example.com/news" in prefs["source_weights"] def test_boost_multiple_sources(self): """Can boost multiple sources for same user.""" helper = ConcretePreferenceManager() manager = helper.manager sources = ["reuters.com", "bbc.com", "cnn.com", "nyt.com"] for source in sources: manager.boost_source("user1", source, weight=2.0) prefs = manager.get_preferences("user1") for source in sources: assert prefs["source_weights"][source] == 2.0 def test_boost_source_weight_precision(self): """Boost weight maintains float precision.""" helper = ConcretePreferenceManager() manager = helper.manager manager.boost_source("user1", "site.com", weight=1.23456789) prefs = manager.get_preferences("user1") assert prefs["source_weights"]["site.com"] == 1.23456789