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
981 lines
30 KiB
Python
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
|