Files
learningcircuit--local-deep…/tests/news/test_base_preference_extended.py
wehub-resource-sync 7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:08:55 +08:00

966 lines
31 KiB
Python

"""
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