""" Extended tests for news/preference_manager/base_preference.py Tests cover: - BasePreferenceManager abstract class - add_interest(), remove_interest() methods - ignore_topic(), boost_source() methods - get_default_preferences() method - TopicRegistry class - extract_topics(), register_topic() methods - get_trending_topics(), get_topic_info() methods - Edge cases and error handling """ import pytest from unittest.mock import Mock, patch from datetime import timedelta class TestBasePreferenceManagerAbstract: """Tests for BasePreferenceManager abstract class.""" def test_is_abstract_class(self): """BasePreferenceManager is an abstract class.""" from local_deep_research.news.preference_manager.base_preference import ( BasePreferenceManager, ) from abc import ABC assert issubclass(BasePreferenceManager, ABC) def test_cannot_instantiate_directly(self): """Cannot directly instantiate BasePreferenceManager.""" from local_deep_research.news.preference_manager.base_preference import ( BasePreferenceManager, ) with pytest.raises(TypeError): BasePreferenceManager() def test_has_get_preferences_abstract_method(self): """Has abstract get_preferences method.""" from local_deep_research.news.preference_manager.base_preference import ( BasePreferenceManager, ) assert hasattr(BasePreferenceManager, "get_preferences") def test_has_update_preferences_abstract_method(self): """Has abstract update_preferences method.""" from local_deep_research.news.preference_manager.base_preference import ( BasePreferenceManager, ) assert hasattr(BasePreferenceManager, "update_preferences") class ConcretePreferenceManager: """Factory for creating concrete preference manager instances.""" @staticmethod def create(storage_backend=None, initial_prefs=None): """Create a concrete preference manager for testing.""" from local_deep_research.news.preference_manager.base_preference import ( BasePreferenceManager, ) class TestPreferenceManager(BasePreferenceManager): def __init__(self, storage_backend=None, initial_prefs=None): super().__init__(storage_backend) self._prefs = initial_prefs or {} 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 return TestPreferenceManager(storage_backend, initial_prefs) class TestBasePreferenceManagerInit: """Tests for BasePreferenceManager initialization.""" def test_accepts_storage_backend(self): """Accepts storage_backend parameter.""" mock_storage = Mock() manager = ConcretePreferenceManager.create(storage_backend=mock_storage) assert manager.storage_backend is mock_storage def test_storage_backend_defaults_to_none(self): """Storage backend defaults to None.""" manager = ConcretePreferenceManager.create() assert manager.storage_backend is None class TestAddInterest: """Tests for add_interest() method.""" def test_adds_interest_to_preferences(self): """add_interest adds interest to user preferences.""" manager = ConcretePreferenceManager.create() manager.add_interest("user1", "AI") prefs = manager.get_preferences("user1") assert "AI" in prefs["interests"] def test_adds_interest_with_default_weight(self): """add_interest uses default weight of 1.0.""" manager = ConcretePreferenceManager.create() manager.add_interest("user1", "technology") prefs = manager.get_preferences("user1") assert prefs["interests"]["technology"] == 1.0 def test_adds_interest_with_custom_weight(self): """add_interest accepts custom weight.""" manager = ConcretePreferenceManager.create() manager.add_interest("user1", "science", weight=2.5) prefs = manager.get_preferences("user1") assert prefs["interests"]["science"] == 2.5 def test_creates_interests_dict_if_missing(self): """add_interest creates interests dict if missing.""" initial_prefs = {"user1": {"liked_categories": []}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) manager.add_interest("user1", "sports") prefs = manager.get_preferences("user1") assert "interests" in prefs assert "sports" in prefs["interests"] def test_updates_interests_updated_at(self): """add_interest updates interests_updated_at timestamp.""" manager = ConcretePreferenceManager.create() manager.add_interest("user1", "finance") prefs = manager.get_preferences("user1") assert "interests_updated_at" in prefs assert prefs["interests_updated_at"] is not None def test_overwrites_existing_interest_weight(self): """add_interest overwrites existing interest weight.""" initial_prefs = {"user1": {"interests": {"AI": 1.0}}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) manager.add_interest("user1", "AI", weight=3.0) prefs = manager.get_preferences("user1") assert prefs["interests"]["AI"] == 3.0 def test_multiple_interests_for_same_user(self): """Can add multiple interests for same user.""" manager = ConcretePreferenceManager.create() manager.add_interest("user1", "AI") manager.add_interest("user1", "robotics") manager.add_interest("user1", "machine learning") prefs = manager.get_preferences("user1") assert len(prefs["interests"]) == 3 class TestRemoveInterest: """Tests for remove_interest() method.""" def test_removes_existing_interest(self): """remove_interest removes existing interest.""" initial_prefs = {"user1": {"interests": {"AI": 1.0, "tech": 1.5}}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) manager.remove_interest("user1", "AI") prefs = manager.get_preferences("user1") assert "AI" not in prefs["interests"] assert "tech" in prefs["interests"] def test_updates_timestamp_on_removal(self): """remove_interest updates interests_updated_at timestamp.""" initial_prefs = {"user1": {"interests": {"AI": 1.0}}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) manager.remove_interest("user1", "AI") prefs = manager.get_preferences("user1") assert "interests_updated_at" in prefs def test_no_error_removing_nonexistent_interest(self): """remove_interest does not error for nonexistent interest.""" initial_prefs = {"user1": {"interests": {"AI": 1.0}}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) # Should not raise manager.remove_interest("user1", "nonexistent") def test_no_error_when_interests_missing(self): """remove_interest handles missing interests dict.""" initial_prefs = {"user1": {"liked_categories": []}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) # Should not raise manager.remove_interest("user1", "AI") class TestIgnoreTopic: """Tests for ignore_topic() method.""" def test_adds_topic_to_disliked(self): """ignore_topic adds topic to disliked_topics.""" manager = ConcretePreferenceManager.create() manager.ignore_topic("user1", "politics") prefs = manager.get_preferences("user1") assert "politics" in prefs["disliked_topics"] def test_creates_disliked_topics_if_missing(self): """ignore_topic creates disliked_topics list if missing.""" initial_prefs = {"user1": {"interests": {}}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) manager.ignore_topic("user1", "gossip") prefs = manager.get_preferences("user1") assert "disliked_topics" in prefs assert "gossip" in prefs["disliked_topics"] def test_no_duplicate_topics(self): """ignore_topic doesn't add duplicate topics.""" initial_prefs = {"user1": {"disliked_topics": ["politics"]}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) manager.ignore_topic("user1", "politics") prefs = manager.get_preferences("user1") assert prefs["disliked_topics"].count("politics") == 1 def test_updates_preferences_updated_at(self): """ignore_topic updates preferences_updated_at timestamp.""" manager = ConcretePreferenceManager.create() manager.ignore_topic("user1", "sports") prefs = manager.get_preferences("user1") assert "preferences_updated_at" in prefs def test_multiple_ignored_topics(self): """Can ignore multiple topics.""" manager = ConcretePreferenceManager.create() manager.ignore_topic("user1", "politics") manager.ignore_topic("user1", "gossip") manager.ignore_topic("user1", "celebrity") prefs = manager.get_preferences("user1") assert len(prefs["disliked_topics"]) == 3 class TestBoostSource: """Tests for boost_source() method.""" def test_boosts_source_with_default_weight(self): """boost_source uses default weight of 1.5.""" manager = ConcretePreferenceManager.create() manager.boost_source("user1", "nytimes.com") prefs = manager.get_preferences("user1") assert prefs["source_weights"]["nytimes.com"] == 1.5 def test_boosts_source_with_custom_weight(self): """boost_source accepts custom weight.""" manager = ConcretePreferenceManager.create() manager.boost_source("user1", "reuters.com", weight=2.0) prefs = manager.get_preferences("user1") assert prefs["source_weights"]["reuters.com"] == 2.0 def test_creates_source_weights_if_missing(self): """boost_source creates source_weights dict if missing.""" initial_prefs = {"user1": {"interests": {}}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) manager.boost_source("user1", "bbc.com") prefs = manager.get_preferences("user1") assert "source_weights" in prefs assert "bbc.com" in prefs["source_weights"] def test_overwrites_existing_source_weight(self): """boost_source overwrites existing source weight.""" initial_prefs = {"user1": {"source_weights": {"nytimes.com": 1.5}}} manager = ConcretePreferenceManager.create(initial_prefs=initial_prefs) manager.boost_source("user1", "nytimes.com", weight=3.0) prefs = manager.get_preferences("user1") assert prefs["source_weights"]["nytimes.com"] == 3.0 def test_updates_preferences_updated_at(self): """boost_source updates preferences_updated_at timestamp.""" manager = ConcretePreferenceManager.create() manager.boost_source("user1", "cnn.com") prefs = manager.get_preferences("user1") assert "preferences_updated_at" in prefs class TestGetDefaultPreferences: """Tests for get_default_preferences() method.""" def test_returns_dict(self): """get_default_preferences returns dictionary.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert isinstance(result, dict) def test_has_liked_categories(self): """Default preferences has liked_categories.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "liked_categories" in result assert result["liked_categories"] == [] def test_has_disliked_categories(self): """Default preferences has disliked_categories.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "disliked_categories" in result assert result["disliked_categories"] == [] def test_has_liked_topics(self): """Default preferences has liked_topics.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "liked_topics" in result assert result["liked_topics"] == [] def test_has_disliked_topics(self): """Default preferences has disliked_topics.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "disliked_topics" in result assert result["disliked_topics"] == [] def test_has_interests_dict(self): """Default preferences has interests dict.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "interests" in result assert result["interests"] == {} def test_has_source_weights_dict(self): """Default preferences has source_weights dict.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "source_weights" in result assert result["source_weights"] == {} def test_has_impact_threshold(self): """Default preferences has impact_threshold of 5.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "impact_threshold" in result assert result["impact_threshold"] == 5 def test_has_focus_preferences(self): """Default preferences has focus_preferences.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "focus_preferences" in result assert result["focus_preferences"]["surprising"] is False assert result["focus_preferences"]["breaking"] is True assert result["focus_preferences"]["positive"] is False assert result["focus_preferences"]["local"] is False def test_has_custom_search_terms(self): """Default preferences has empty custom_search_terms.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "custom_search_terms" in result assert result["custom_search_terms"] == "" def test_has_search_strategy(self): """Default preferences has search_strategy.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "search_strategy" in result assert result["search_strategy"] == "news_aggregation" def test_has_created_at(self): """Default preferences has created_at timestamp.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "created_at" in result assert result["created_at"] is not None def test_has_preferences_updated_at(self): """Default preferences has preferences_updated_at timestamp.""" manager = ConcretePreferenceManager.create() result = manager.get_default_preferences() assert "preferences_updated_at" in result assert result["preferences_updated_at"] is not None class TestTopicRegistryInit: """Tests for TopicRegistry initialization.""" def test_creates_empty_topics_dict(self): """TopicRegistry starts with empty topics dict.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() assert registry.topics == {} def test_accepts_llm_client(self): """TopicRegistry accepts llm_client parameter.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) mock_llm = Mock() registry = TopicRegistry(llm_client=mock_llm) assert registry.llm_client is mock_llm def test_llm_client_defaults_to_none(self): """llm_client defaults to None.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() assert registry.llm_client is None class TestRegisterTopic: """Tests for register_topic() method.""" def test_registers_new_topic(self): """register_topic 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_timestamp(self): """register_topic sets first_seen timestamp for new topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("technology") assert "first_seen" in registry.topics["technology"] assert registry.topics["technology"]["first_seen"] is not None def test_sets_initial_count_to_one(self): """register_topic sets count to 1 for new topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("science") assert registry.topics["science"]["count"] == 1 def test_increments_count_for_existing_topic(self): """register_topic increments count for existing topic.""" 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 registry.topics["AI"]["count"] == 3 def test_updates_last_seen_timestamp(self): """register_topic updates last_seen timestamp.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("robotics") assert "last_seen" in registry.topics["robotics"] def test_preserves_first_seen_on_reregister(self): """register_topic preserves first_seen on re-registration.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("data science") first_seen = registry.topics["data science"]["first_seen"] registry.register_topic("data science") assert registry.topics["data science"]["first_seen"] == first_seen class TestExtractTopics: """Tests for extract_topics() method.""" def test_returns_list(self): """extract_topics returns a list.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) with patch( "local_deep_research.news.utils.topic_generator.generate_topics" ) as mock_gen: mock_gen.return_value = ["AI", "technology"] registry = TopicRegistry() result = registry.extract_topics("Some content about AI and tech") assert isinstance(result, list) def test_calls_generate_topics(self): """extract_topics calls topic generator.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) with patch( "local_deep_research.news.utils.topic_generator.generate_topics" ) as mock_gen: mock_gen.return_value = ["science"] registry = TopicRegistry() registry.extract_topics("Scientific discovery content") mock_gen.assert_called_once() def test_respects_max_topics_parameter(self): """extract_topics passes max_topics to generator.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) with patch( "local_deep_research.news.utils.topic_generator.generate_topics" ) as mock_gen: mock_gen.return_value = ["topic1", "topic2"] registry = TopicRegistry() registry.extract_topics("Content", max_topics=3) call_kwargs = mock_gen.call_args[1] assert call_kwargs["max_topics"] == 3 def test_registers_extracted_topics(self): """extract_topics registers all extracted topics.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) with patch( "local_deep_research.news.utils.topic_generator.generate_topics" ) as mock_gen: mock_gen.return_value = ["AI", "robotics", "automation"] registry = TopicRegistry() registry.extract_topics("Content about various topics") assert "AI" in registry.topics assert "robotics" in registry.topics assert "automation" in registry.topics def test_returns_extracted_topics(self): """extract_topics returns the extracted topics.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) with patch( "local_deep_research.news.utils.topic_generator.generate_topics" ) as mock_gen: mock_gen.return_value = ["machine learning", "neural networks"] registry = TopicRegistry() result = registry.extract_topics("ML content") assert "machine learning" in result assert "neural networks" in result class TestGetTrendingTopics: """Tests for get_trending_topics() method.""" def test_returns_list(self): """get_trending_topics 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): """get_trending_topics returns empty list when no topics.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() result = registry.get_trending_topics() assert result == [] def test_returns_recent_topics(self): """get_trending_topics returns recently seen topics.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("AI") registry.register_topic("tech") result = registry.get_trending_topics(hours=24) assert "AI" in result assert "tech" in result def test_sorts_by_count_descending(self): """get_trending_topics sorts by count (most frequent first).""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("rare") for _ in range(5): registry.register_topic("popular") for _ in range(3): registry.register_topic("medium") result = registry.get_trending_topics() assert result[0] == "popular" assert result[1] == "medium" assert result[2] == "rare" def test_respects_limit_parameter(self): """get_trending_topics respects 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_filters_by_hours_parameter(self): """get_trending_topics filters 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() registry.register_topic("recent") # Manually set an old topic old_time = utc_now() - timedelta(hours=48) registry.topics["old_topic"] = { "first_seen": old_time, "last_seen": old_time, "count": 10, } result = registry.get_trending_topics(hours=24) assert "recent" in result assert "old_topic" not in result def test_default_hours_is_24(self): """get_trending_topics defaults to 24 hours lookback.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) import inspect sig = inspect.signature(TopicRegistry.get_trending_topics) assert sig.parameters["hours"].default == 24 def test_default_limit_is_10(self): """get_trending_topics defaults to limit of 10.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) import inspect sig = inspect.signature(TopicRegistry.get_trending_topics) assert sig.parameters["limit"].default == 10 class TestGetTopicInfo: """Tests for get_topic_info() method.""" def test_returns_dict_for_existing_topic(self): """get_topic_info returns dict 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 isinstance(result, dict) def test_returns_none_for_unknown_topic(self): """get_topic_info returns None for unknown topic.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() result = registry.get_topic_info("nonexistent") assert result is None def test_info_contains_first_seen(self): """Topic info contains first_seen timestamp.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("science") result = registry.get_topic_info("science") assert "first_seen" in result def test_info_contains_count(self): """Topic info contains count.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("tech") registry.register_topic("tech") result = registry.get_topic_info("tech") assert "count" in result assert result["count"] == 2 def test_info_contains_last_seen(self): """Topic info contains last_seen timestamp.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("robotics") result = registry.get_topic_info("robotics") assert "last_seen" in result class TestPreferenceManagerEdgeCases: """Edge case tests for preference management.""" def test_empty_user_id(self): """Handles empty user_id.""" manager = ConcretePreferenceManager.create() manager.add_interest("", "AI") prefs = manager.get_preferences("") assert "AI" in prefs["interests"] def test_unicode_interests(self): """Handles unicode interests.""" manager = ConcretePreferenceManager.create() manager.add_interest("user1", "日本語") prefs = manager.get_preferences("user1") assert "日本語" in prefs["interests"] def test_special_characters_in_topic(self): """Handles special characters in topic.""" manager = ConcretePreferenceManager.create() manager.ignore_topic("user1", "C++ & Python") prefs = manager.get_preferences("user1") assert "C++ & Python" in prefs["disliked_topics"] def test_very_long_interest_name(self): """Handles very long interest names.""" manager = ConcretePreferenceManager.create() long_interest = "x" * 1000 manager.add_interest("user1", long_interest) prefs = manager.get_preferences("user1") assert long_interest in prefs["interests"] def test_zero_weight_interest(self): """Handles zero weight for interest.""" manager = ConcretePreferenceManager.create() manager.add_interest("user1", "low_priority", weight=0.0) prefs = manager.get_preferences("user1") assert prefs["interests"]["low_priority"] == 0.0 def test_negative_weight_interest(self): """Handles negative weight for interest.""" manager = ConcretePreferenceManager.create() manager.add_interest("user1", "disliked", weight=-1.0) prefs = manager.get_preferences("user1") assert prefs["interests"]["disliked"] == -1.0 class TestTopicRegistryEdgeCases: """Edge case tests for TopicRegistry.""" def test_empty_topic_name(self): """Handles empty topic name.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("") assert "" in registry.topics def test_unicode_topic_name(self): """Handles unicode topic names.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("科技新闻") result = registry.get_topic_info("科技新闻") assert result is not None 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}") result = registry.get_trending_topics(limit=100) assert len(result) == 100 def test_zero_hours_lookback(self): """Handles zero hours lookback.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("test") # Zero hours means nothing is recent result = registry.get_trending_topics(hours=0) # All topics were registered "now", so with 0 hours # lookback they might not match depending on timing assert isinstance(result, list) def test_very_large_hours_lookback(self): """Handles very large hours lookback.""" from local_deep_research.news.preference_manager.base_preference import ( TopicRegistry, ) registry = TopicRegistry() registry.register_topic("test") result = registry.get_trending_topics(hours=10000) assert "test" in result