Files
learningcircuit--local-deep…/tests/news/test_preference_manager_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

981 lines
30 KiB
Python

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