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
729 lines
22 KiB
Python
729 lines
22 KiB
Python
"""
|
|
Comprehensive tests for base_preference module.
|
|
Tests BasePreferenceManager and TopicRegistry classes.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch
|
|
from datetime import timedelta
|
|
|
|
|
|
class TestBasePreferenceManagerInit:
|
|
"""Tests for BasePreferenceManager initialization."""
|
|
|
|
def test_stores_storage_backend(self):
|
|
"""Test stores storage_backend reference."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
# Create a concrete implementation for testing
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, preferences):
|
|
return preferences
|
|
|
|
mock_storage = Mock()
|
|
manager = ConcretePreferenceManager(storage_backend=mock_storage)
|
|
|
|
assert manager.storage_backend is mock_storage
|
|
|
|
def test_accepts_none_storage(self):
|
|
"""Test accepts None for storage_backend."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, preferences):
|
|
return preferences
|
|
|
|
manager = ConcretePreferenceManager(storage_backend=None)
|
|
|
|
assert manager.storage_backend is None
|
|
|
|
|
|
class TestAddInterest:
|
|
"""Tests for add_interest method."""
|
|
|
|
def test_adds_interest_to_preferences(self):
|
|
"""Test adds interest to user preferences."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.add_interest("user123", "AI")
|
|
|
|
assert "interests" in preferences
|
|
assert "AI" in preferences["interests"]
|
|
|
|
def test_sets_weight_for_interest(self):
|
|
"""Test sets weight for interest."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.add_interest("user123", "Machine Learning", weight=2.0)
|
|
|
|
assert preferences["interests"]["Machine Learning"] == 2.0
|
|
|
|
def test_default_weight_is_1(self):
|
|
"""Test default weight is 1.0."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.add_interest("user123", "Tech")
|
|
|
|
assert preferences["interests"]["Tech"] == 1.0
|
|
|
|
def test_updates_timestamp(self):
|
|
"""Test updates interests_updated_at timestamp."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.add_interest("user123", "Science")
|
|
|
|
assert "interests_updated_at" in preferences
|
|
|
|
|
|
class TestRemoveInterest:
|
|
"""Tests for remove_interest method."""
|
|
|
|
def test_removes_interest_from_preferences(self):
|
|
"""Test removes interest from preferences."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {"interests": {"AI": 1.0, "Tech": 1.5}}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.remove_interest("user123", "AI")
|
|
|
|
assert "AI" not in preferences["interests"]
|
|
assert "Tech" in preferences["interests"]
|
|
|
|
def test_does_nothing_if_interest_not_found(self):
|
|
"""Test does nothing if interest doesn't exist."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {"interests": {"AI": 1.0}}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
|
|
# Should not raise
|
|
manager.remove_interest("user123", "NonExistent")
|
|
|
|
|
|
class TestIgnoreTopic:
|
|
"""Tests for ignore_topic method."""
|
|
|
|
def test_adds_topic_to_disliked_list(self):
|
|
"""Test adds topic to disliked_topics list."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.ignore_topic("user123", "Politics")
|
|
|
|
assert "Politics" in preferences["disliked_topics"]
|
|
|
|
def test_does_not_add_duplicate_topic(self):
|
|
"""Test does not add duplicate topic."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {"disliked_topics": ["Politics"]}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.ignore_topic("user123", "Politics")
|
|
|
|
assert preferences["disliked_topics"].count("Politics") == 1
|
|
|
|
def test_updates_timestamp(self):
|
|
"""Test updates preferences_updated_at timestamp."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.ignore_topic("user123", "Sports")
|
|
|
|
assert "preferences_updated_at" in preferences
|
|
|
|
|
|
class TestBoostSource:
|
|
"""Tests for boost_source method."""
|
|
|
|
def test_adds_source_weight(self):
|
|
"""Test adds source to source_weights."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.boost_source("user123", "reuters.com", weight=2.0)
|
|
|
|
assert "reuters.com" in preferences["source_weights"]
|
|
assert preferences["source_weights"]["reuters.com"] == 2.0
|
|
|
|
def test_default_weight_is_1_5(self):
|
|
"""Test default weight is 1.5."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
preferences = {}
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return preferences
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
preferences.update(prefs)
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
manager.boost_source("user123", "bbc.com")
|
|
|
|
assert preferences["source_weights"]["bbc.com"] == 1.5
|
|
|
|
|
|
class TestGetDefaultPreferences:
|
|
"""Tests for get_default_preferences method."""
|
|
|
|
def test_returns_dict(self):
|
|
"""Test returns a dictionary."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
result = manager.get_default_preferences()
|
|
|
|
assert isinstance(result, dict)
|
|
|
|
def test_includes_liked_categories(self):
|
|
"""Test includes liked_categories list."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
result = manager.get_default_preferences()
|
|
|
|
assert "liked_categories" in result
|
|
assert isinstance(result["liked_categories"], list)
|
|
|
|
def test_includes_impact_threshold(self):
|
|
"""Test includes impact_threshold with default 5."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
result = manager.get_default_preferences()
|
|
|
|
assert result["impact_threshold"] == 5
|
|
|
|
def test_includes_focus_preferences(self):
|
|
"""Test includes focus_preferences dict."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
result = manager.get_default_preferences()
|
|
|
|
assert "focus_preferences" in result
|
|
assert "breaking" in result["focus_preferences"]
|
|
assert result["focus_preferences"]["breaking"] is True
|
|
|
|
def test_includes_search_strategy(self):
|
|
"""Test includes search_strategy."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
result = manager.get_default_preferences()
|
|
|
|
assert "search_strategy" in result
|
|
|
|
def test_includes_timestamps(self):
|
|
"""Test includes created_at and preferences_updated_at."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class ConcretePreferenceManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, prefs):
|
|
return prefs
|
|
|
|
manager = ConcretePreferenceManager()
|
|
result = manager.get_default_preferences()
|
|
|
|
assert "created_at" in result
|
|
assert "preferences_updated_at" in result
|
|
|
|
|
|
class TestTopicRegistryInit:
|
|
"""Tests for TopicRegistry initialization."""
|
|
|
|
def test_creates_instance(self):
|
|
"""Test creates instance successfully."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
assert registry is not None
|
|
|
|
def test_accepts_llm_client(self):
|
|
"""Test accepts optional llm_client."""
|
|
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_initializes_empty_topics_dict(self):
|
|
"""Test initializes empty topics dictionary."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
assert registry.topics == {}
|
|
|
|
|
|
class TestExtractTopics:
|
|
"""Tests for extract_topics method."""
|
|
|
|
def test_extract_topics_returns_list(self):
|
|
"""Test extract_topics returns a list."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
# Mock the internal call to generate_topics via patching the import
|
|
with patch(
|
|
"local_deep_research.news.utils.topic_generator.generate_topics"
|
|
) as mock_gen:
|
|
mock_gen.return_value = ["AI", "Tech"]
|
|
|
|
result = registry.extract_topics("Some content")
|
|
|
|
assert isinstance(result, list)
|
|
|
|
def test_extract_topics_registers_topics(self):
|
|
"""Test extract_topics registers discovered topics."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
# Manually add topics via register_topic to test registration
|
|
registry.register_topic("ManualTopic")
|
|
|
|
assert "ManualTopic" in registry.topics
|
|
|
|
def test_extract_topics_accepts_max_topics(self):
|
|
"""Test extract_topics accepts max_topics parameter."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
import inspect
|
|
|
|
registry = TopicRegistry()
|
|
|
|
# Check method signature
|
|
sig = inspect.signature(registry.extract_topics)
|
|
params = list(sig.parameters.keys())
|
|
|
|
assert "max_topics" in params
|
|
|
|
def test_extract_topics_default_max_is_5(self):
|
|
"""Test extract_topics default max_topics is 5."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
import inspect
|
|
|
|
registry = TopicRegistry()
|
|
|
|
sig = inspect.signature(registry.extract_topics)
|
|
max_topics_param = sig.parameters.get("max_topics")
|
|
|
|
assert max_topics_param.default == 5
|
|
|
|
|
|
class TestRegisterTopic:
|
|
"""Tests for register_topic method."""
|
|
|
|
def test_adds_new_topic(self):
|
|
"""Test adds new topic to registry."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
registry.register_topic("NewTopic")
|
|
|
|
assert "NewTopic" in registry.topics
|
|
|
|
def test_sets_first_seen_for_new_topic(self):
|
|
"""Test sets first_seen for new topic."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
registry.register_topic("Fresh")
|
|
|
|
assert "first_seen" in registry.topics["Fresh"]
|
|
|
|
def test_initializes_count_to_1(self):
|
|
"""Test initializes count to 1 for new topic."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
registry.register_topic("New")
|
|
|
|
assert registry.topics["New"]["count"] == 1
|
|
|
|
def test_increments_count_for_existing_topic(self):
|
|
"""Test increments count for existing topic."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
registry.register_topic("Existing")
|
|
registry.register_topic("Existing")
|
|
|
|
assert registry.topics["Existing"]["count"] == 2
|
|
|
|
def test_updates_last_seen(self):
|
|
"""Test updates last_seen timestamp."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
registry.register_topic("Topic")
|
|
|
|
first_seen = registry.topics["Topic"]["last_seen"]
|
|
|
|
registry.register_topic("Topic")
|
|
|
|
# Should be updated (or same if executed very quickly)
|
|
assert registry.topics["Topic"]["last_seen"] >= first_seen
|
|
|
|
|
|
class TestGetTrendingTopics:
|
|
"""Tests for get_trending_topics method."""
|
|
|
|
def test_returns_list(self):
|
|
"""Test 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):
|
|
"""Test 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_filters_by_hours(self):
|
|
"""Test filters topics by hours parameter."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
from local_deep_research.news.core.utils import utc_now
|
|
|
|
registry = TopicRegistry()
|
|
|
|
# Add old topic
|
|
old_time = utc_now() - timedelta(hours=48)
|
|
registry.topics["OldTopic"] = {
|
|
"first_seen": old_time,
|
|
"last_seen": old_time,
|
|
"count": 10,
|
|
}
|
|
|
|
# Add recent topic
|
|
registry.register_topic("RecentTopic")
|
|
|
|
result = registry.get_trending_topics(hours=24)
|
|
|
|
assert "RecentTopic" in result
|
|
assert "OldTopic" not in result
|
|
|
|
def test_sorts_by_count(self):
|
|
"""Test sorts topics by count (highest first)."""
|
|
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("HighCount")
|
|
for _ in range(2):
|
|
registry.register_topic("LowCount")
|
|
|
|
result = registry.get_trending_topics()
|
|
|
|
assert result.index("HighCount") < result.index("LowCount")
|
|
|
|
def test_respects_limit(self):
|
|
"""Test 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_default_limit_is_10(self):
|
|
"""Test 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 TestGetTopicInfo:
|
|
"""Tests for get_topic_info method."""
|
|
|
|
def test_returns_topic_data(self):
|
|
"""Test returns topic data dictionary."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
registry.register_topic("MyTopic")
|
|
|
|
result = registry.get_topic_info("MyTopic")
|
|
|
|
assert isinstance(result, dict)
|
|
assert "count" in result
|
|
assert "first_seen" in result
|
|
|
|
def test_returns_none_for_unknown_topic(self):
|
|
"""Test 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
|
|
|
|
|
|
class TestBasePreferenceAbstract:
|
|
"""Tests for abstract base class behavior."""
|
|
|
|
def test_is_abstract(self):
|
|
"""Test BasePreferenceManager is abstract."""
|
|
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):
|
|
"""Test cannot instantiate BasePreferenceManager directly."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
with pytest.raises(TypeError):
|
|
BasePreferenceManager()
|