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
1303 lines
44 KiB
Python
1303 lines
44 KiB
Python
"""
|
|
Tests for news/preference_manager/ modules.
|
|
|
|
Tests cover:
|
|
- BasePreferenceManager abstract class and methods
|
|
- TopicRegistry for global topic tracking
|
|
- SQLPreferenceStorage CRUD operations
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
# =============================================================================
|
|
# BasePreferenceManager Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestBasePreferenceManagerAbstract:
|
|
"""Tests for BasePreferenceManager abstract class enforcement."""
|
|
|
|
def test_cannot_instantiate_directly(self):
|
|
"""Test that BasePreferenceManager cannot be instantiated directly."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
with pytest.raises(TypeError):
|
|
BasePreferenceManager()
|
|
|
|
def test_requires_get_preferences_implementation(self):
|
|
"""Test that get_preferences must be implemented."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class PartialManager(BasePreferenceManager):
|
|
def update_preferences(self, user_id, preferences):
|
|
return preferences
|
|
|
|
with pytest.raises(TypeError):
|
|
PartialManager()
|
|
|
|
def test_requires_update_preferences_implementation(self):
|
|
"""Test that update_preferences must be implemented."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class PartialManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
with pytest.raises(TypeError):
|
|
PartialManager()
|
|
|
|
|
|
class TestBasePreferenceManagerInit:
|
|
"""Tests for BasePreferenceManager initialization."""
|
|
|
|
def test_init_with_storage_backend(self):
|
|
"""Test initialization with a storage backend."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
mock_storage = MagicMock()
|
|
|
|
class ConcreteManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, preferences):
|
|
return preferences
|
|
|
|
manager = ConcreteManager(storage_backend=mock_storage)
|
|
assert manager.storage_backend is mock_storage
|
|
|
|
def test_init_without_storage_backend(self):
|
|
"""Test initialization without a storage backend."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class ConcreteManager(BasePreferenceManager):
|
|
def get_preferences(self, user_id):
|
|
return {}
|
|
|
|
def update_preferences(self, user_id, preferences):
|
|
return preferences
|
|
|
|
manager = ConcreteManager()
|
|
assert manager.storage_backend is None
|
|
|
|
|
|
class ConcretePreferenceManager:
|
|
"""Helper class for testing BasePreferenceManager methods."""
|
|
|
|
@staticmethod
|
|
def create():
|
|
"""Create a concrete implementation for testing."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
BasePreferenceManager,
|
|
)
|
|
|
|
class Manager(BasePreferenceManager):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._prefs = {}
|
|
|
|
def get_preferences(self, user_id):
|
|
return self._prefs.get(user_id, {})
|
|
|
|
def update_preferences(self, user_id, preferences):
|
|
self._prefs[user_id] = preferences
|
|
return preferences
|
|
|
|
return Manager()
|
|
|
|
|
|
class TestAddInterest:
|
|
"""Tests for add_interest method."""
|
|
|
|
def test_add_interest_to_empty_preferences(self):
|
|
"""Test adding interest when preferences are empty."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.add_interest("user1", "technology", weight=1.0)
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert "interests" in prefs
|
|
assert prefs["interests"]["technology"] == 1.0
|
|
assert "interests_updated_at" in prefs
|
|
|
|
def test_add_interest_with_custom_weight(self):
|
|
"""Test adding interest with a custom weight."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.add_interest("user1", "sports", weight=2.5)
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert prefs["interests"]["sports"] == 2.5
|
|
|
|
def test_add_interest_overwrites_existing(self):
|
|
"""Test that adding an interest overwrites existing weight."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.add_interest("user1", "tech", weight=1.0)
|
|
manager.add_interest("user1", "tech", weight=3.0)
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert prefs["interests"]["tech"] == 3.0
|
|
|
|
def test_add_multiple_interests(self):
|
|
"""Test adding multiple interests."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.add_interest("user1", "tech", weight=1.0)
|
|
manager.add_interest("user1", "sports", weight=2.0)
|
|
manager.add_interest("user1", "politics", weight=0.5)
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert len(prefs["interests"]) == 3
|
|
assert prefs["interests"]["tech"] == 1.0
|
|
assert prefs["interests"]["sports"] == 2.0
|
|
assert prefs["interests"]["politics"] == 0.5
|
|
|
|
|
|
class TestRemoveInterest:
|
|
"""Tests for remove_interest method."""
|
|
|
|
def test_remove_existing_interest(self):
|
|
"""Test removing an existing interest."""
|
|
manager = ConcretePreferenceManager.create()
|
|
manager.add_interest("user1", "tech", weight=1.0)
|
|
manager.add_interest("user1", "sports", weight=2.0)
|
|
|
|
manager.remove_interest("user1", "tech")
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert "tech" not in prefs["interests"]
|
|
assert "sports" in prefs["interests"]
|
|
|
|
def test_remove_nonexistent_interest(self):
|
|
"""Test removing an interest that doesn't exist."""
|
|
manager = ConcretePreferenceManager.create()
|
|
manager.add_interest("user1", "tech", weight=1.0)
|
|
|
|
# Should not raise an error
|
|
manager.remove_interest("user1", "nonexistent")
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert "tech" in prefs["interests"]
|
|
|
|
def test_remove_interest_updates_timestamp(self):
|
|
"""Test that removing interest updates the timestamp."""
|
|
manager = ConcretePreferenceManager.create()
|
|
manager.add_interest("user1", "tech", weight=1.0)
|
|
|
|
# Get initial timestamp for later comparison
|
|
initial_prefs = manager.get_preferences("user1")
|
|
assert "interests_updated_at" in initial_prefs
|
|
|
|
with patch(
|
|
"local_deep_research.news.preference_manager.base_preference.utc_now"
|
|
) as mock_now:
|
|
from datetime import datetime, timezone
|
|
|
|
new_time = datetime(2025, 6, 15, 12, 0, 0, tzinfo=timezone.utc)
|
|
mock_now.return_value = new_time
|
|
|
|
manager.remove_interest("user1", "tech")
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
# Timestamp should be updated
|
|
assert "interests_updated_at" in prefs
|
|
|
|
|
|
class TestIgnoreTopic:
|
|
"""Tests for ignore_topic method."""
|
|
|
|
def test_ignore_topic_to_empty_list(self):
|
|
"""Test adding topic to ignore list when empty."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.ignore_topic("user1", "politics")
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert "disliked_topics" in prefs
|
|
assert "politics" in prefs["disliked_topics"]
|
|
|
|
def test_ignore_topic_no_duplicates(self):
|
|
"""Test that ignoring same topic twice doesn't create duplicates."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.ignore_topic("user1", "politics")
|
|
manager.ignore_topic("user1", "politics")
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert prefs["disliked_topics"].count("politics") == 1
|
|
|
|
def test_ignore_multiple_topics(self):
|
|
"""Test ignoring multiple topics."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.ignore_topic("user1", "politics")
|
|
manager.ignore_topic("user1", "gossip")
|
|
manager.ignore_topic("user1", "tabloid")
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert len(prefs["disliked_topics"]) == 3
|
|
|
|
|
|
class TestBoostSource:
|
|
"""Tests for boost_source method."""
|
|
|
|
def test_boost_source_empty_preferences(self):
|
|
"""Test boosting a source when preferences are empty."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.boost_source("user1", "example.com", weight=1.5)
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert "source_weights" in prefs
|
|
assert prefs["source_weights"]["example.com"] == 1.5
|
|
|
|
def test_boost_source_default_weight(self):
|
|
"""Test boost_source uses default weight of 1.5."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.boost_source("user1", "example.com")
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert prefs["source_weights"]["example.com"] == 1.5
|
|
|
|
def test_boost_source_custom_weight(self):
|
|
"""Test boost_source with custom weight."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.boost_source("user1", "trusted.com", weight=3.0)
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert prefs["source_weights"]["trusted.com"] == 3.0
|
|
|
|
def test_boost_source_updates_existing(self):
|
|
"""Test that boosting a source updates its weight."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
manager.boost_source("user1", "example.com", weight=1.5)
|
|
manager.boost_source("user1", "example.com", weight=2.0)
|
|
|
|
prefs = manager.get_preferences("user1")
|
|
assert prefs["source_weights"]["example.com"] == 2.0
|
|
|
|
|
|
class TestGetDefaultPreferences:
|
|
"""Tests for get_default_preferences method."""
|
|
|
|
def test_default_preferences_structure(self):
|
|
"""Test that default preferences have all required fields."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
defaults = manager.get_default_preferences()
|
|
|
|
required_fields = [
|
|
"liked_categories",
|
|
"disliked_categories",
|
|
"liked_topics",
|
|
"disliked_topics",
|
|
"interests",
|
|
"source_weights",
|
|
"impact_threshold",
|
|
"focus_preferences",
|
|
"custom_search_terms",
|
|
"search_strategy",
|
|
"created_at",
|
|
"preferences_updated_at",
|
|
]
|
|
|
|
for field in required_fields:
|
|
assert field in defaults, f"Missing field: {field}"
|
|
|
|
def test_default_preferences_types(self):
|
|
"""Test that default preferences have correct types."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
defaults = manager.get_default_preferences()
|
|
|
|
assert isinstance(defaults["liked_categories"], list)
|
|
assert isinstance(defaults["disliked_categories"], list)
|
|
assert isinstance(defaults["liked_topics"], list)
|
|
assert isinstance(defaults["disliked_topics"], list)
|
|
assert isinstance(defaults["interests"], dict)
|
|
assert isinstance(defaults["source_weights"], dict)
|
|
assert isinstance(defaults["impact_threshold"], int)
|
|
assert isinstance(defaults["focus_preferences"], dict)
|
|
assert isinstance(defaults["custom_search_terms"], str)
|
|
assert isinstance(defaults["search_strategy"], str)
|
|
|
|
def test_default_impact_threshold(self):
|
|
"""Test default impact threshold is 5."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
defaults = manager.get_default_preferences()
|
|
|
|
assert defaults["impact_threshold"] == 5
|
|
|
|
def test_default_focus_preferences(self):
|
|
"""Test default focus preferences structure."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
defaults = manager.get_default_preferences()
|
|
|
|
focus = defaults["focus_preferences"]
|
|
assert focus["surprising"] is False
|
|
assert focus["breaking"] is True
|
|
assert focus["positive"] is False
|
|
assert focus["local"] is False
|
|
|
|
def test_default_search_strategy(self):
|
|
"""Test default search strategy."""
|
|
manager = ConcretePreferenceManager.create()
|
|
|
|
defaults = manager.get_default_preferences()
|
|
|
|
assert defaults["search_strategy"] == "news_aggregation"
|
|
|
|
|
|
# =============================================================================
|
|
# TopicRegistry Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestTopicRegistryInit:
|
|
"""Tests for TopicRegistry initialization."""
|
|
|
|
def test_init_without_llm_client(self):
|
|
"""Test initialization without LLM client."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
assert registry.llm_client is None
|
|
assert registry.topics == {}
|
|
|
|
def test_init_with_llm_client(self):
|
|
"""Test initialization with LLM client."""
|
|
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
|
|
|
|
|
|
class TestRegisterTopic:
|
|
"""Tests for register_topic method."""
|
|
|
|
def test_register_new_topic(self):
|
|
"""Test registering a new topic."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
registry.register_topic("climate change")
|
|
|
|
assert "climate change" in registry.topics
|
|
assert registry.topics["climate change"]["count"] == 1
|
|
assert "first_seen" in registry.topics["climate change"]
|
|
assert "last_seen" in registry.topics["climate change"]
|
|
|
|
def test_register_topic_increments_count(self):
|
|
"""Test that registering same topic increments count."""
|
|
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_register_topic_updates_last_seen(self):
|
|
"""Test that registering updates last_seen timestamp."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
with patch(
|
|
"local_deep_research.news.preference_manager.base_preference.utc_now"
|
|
) as mock_now:
|
|
from datetime import datetime, timezone
|
|
|
|
time1 = datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
mock_now.return_value = time1
|
|
registry.register_topic("topic1")
|
|
|
|
time2 = datetime(2025, 1, 2, 12, 0, 0, tzinfo=timezone.utc)
|
|
mock_now.return_value = time2
|
|
registry.register_topic("topic1")
|
|
|
|
assert registry.topics["topic1"]["last_seen"] == time2
|
|
assert registry.topics["topic1"]["first_seen"] == time1
|
|
|
|
def test_register_multiple_topics(self):
|
|
"""Test registering multiple different topics."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
registry.register_topic("tech")
|
|
registry.register_topic("sports")
|
|
registry.register_topic("politics")
|
|
|
|
assert len(registry.topics) == 3
|
|
for topic in ["tech", "sports", "politics"]:
|
|
assert registry.topics[topic]["count"] == 1
|
|
|
|
|
|
class TestGetTrendingTopics:
|
|
"""Tests for get_trending_topics method."""
|
|
|
|
def test_trending_topics_empty_registry(self):
|
|
"""Test trending topics with empty registry."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
result = registry.get_trending_topics()
|
|
|
|
assert result == []
|
|
|
|
def test_trending_topics_filters_by_time(self):
|
|
"""Test that trending topics filters by time window."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
from datetime import datetime, timezone
|
|
|
|
registry = TopicRegistry()
|
|
|
|
# Register a topic with old timestamp
|
|
old_time = datetime(2020, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
registry.topics["old_topic"] = {
|
|
"first_seen": old_time,
|
|
"last_seen": old_time,
|
|
"count": 10,
|
|
}
|
|
|
|
# Register a recent topic
|
|
with patch(
|
|
"local_deep_research.news.preference_manager.base_preference.utc_now"
|
|
) as mock_now:
|
|
recent_time = datetime(2025, 6, 15, 12, 0, 0, tzinfo=timezone.utc)
|
|
mock_now.return_value = recent_time
|
|
registry.register_topic("recent_topic")
|
|
|
|
# Get trending with current time
|
|
result = registry.get_trending_topics(hours=24)
|
|
|
|
assert "recent_topic" in result
|
|
assert "old_topic" not in result
|
|
|
|
def test_trending_topics_sorted_by_count(self):
|
|
"""Test that trending topics are sorted by count."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
with patch(
|
|
"local_deep_research.news.preference_manager.base_preference.utc_now"
|
|
) as mock_now:
|
|
from datetime import datetime, timezone
|
|
|
|
now = datetime(2025, 6, 15, 12, 0, 0, tzinfo=timezone.utc)
|
|
mock_now.return_value = now
|
|
|
|
# Register with different counts
|
|
for _ in range(5):
|
|
registry.register_topic("popular")
|
|
for _ in range(3):
|
|
registry.register_topic("medium")
|
|
registry.register_topic("rare")
|
|
|
|
result = registry.get_trending_topics(hours=24)
|
|
|
|
assert result[0] == "popular"
|
|
assert result[1] == "medium"
|
|
assert result[2] == "rare"
|
|
|
|
def test_trending_topics_respects_limit(self):
|
|
"""Test that trending topics respects limit parameter."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
with patch(
|
|
"local_deep_research.news.preference_manager.base_preference.utc_now"
|
|
) as mock_now:
|
|
from datetime import datetime, timezone
|
|
|
|
now = datetime(2025, 6, 15, 12, 0, 0, tzinfo=timezone.utc)
|
|
mock_now.return_value = now
|
|
|
|
for i in range(20):
|
|
registry.register_topic(f"topic_{i}")
|
|
|
|
result = registry.get_trending_topics(hours=24, limit=5)
|
|
|
|
assert len(result) == 5
|
|
|
|
def test_trending_topics_custom_hours(self):
|
|
"""Test trending topics with custom hours parameter."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
from datetime import datetime, timezone
|
|
|
|
registry = TopicRegistry()
|
|
|
|
# Topic from 12 hours ago
|
|
time_12h_ago = datetime(2025, 6, 15, 0, 0, 0, tzinfo=timezone.utc)
|
|
registry.topics["topic_12h"] = {
|
|
"first_seen": time_12h_ago,
|
|
"last_seen": time_12h_ago,
|
|
"count": 5,
|
|
}
|
|
|
|
# Topic from 36 hours ago
|
|
time_36h_ago = datetime(2025, 6, 14, 0, 0, 0, tzinfo=timezone.utc)
|
|
registry.topics["topic_36h"] = {
|
|
"first_seen": time_36h_ago,
|
|
"last_seen": time_36h_ago,
|
|
"count": 5,
|
|
}
|
|
|
|
with patch(
|
|
"local_deep_research.news.preference_manager.base_preference.utc_now"
|
|
) as mock_now:
|
|
now = datetime(2025, 6, 15, 12, 0, 0, tzinfo=timezone.utc)
|
|
mock_now.return_value = now
|
|
|
|
result_24h = registry.get_trending_topics(hours=24)
|
|
result_48h = registry.get_trending_topics(hours=48)
|
|
|
|
assert "topic_12h" in result_24h
|
|
assert "topic_36h" not in result_24h
|
|
assert "topic_12h" in result_48h
|
|
assert "topic_36h" in result_48h
|
|
|
|
|
|
class TestGetTopicInfo:
|
|
"""Tests for get_topic_info method."""
|
|
|
|
def test_get_existing_topic_info(self):
|
|
"""Test getting info for existing topic."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
registry.register_topic("AI")
|
|
registry.register_topic("AI")
|
|
|
|
info = registry.get_topic_info("AI")
|
|
|
|
assert info is not None
|
|
assert info["count"] == 2
|
|
assert "first_seen" in info
|
|
assert "last_seen" in info
|
|
|
|
def test_get_nonexistent_topic_info(self):
|
|
"""Test getting info for nonexistent topic."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
info = registry.get_topic_info("nonexistent")
|
|
|
|
assert info is None
|
|
|
|
|
|
class TestExtractTopics:
|
|
"""Tests for extract_topics method."""
|
|
|
|
def test_extract_topics_uses_topic_generator(self):
|
|
"""Test that extract_topics uses topic generator."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
with patch(
|
|
"local_deep_research.news.utils.topic_generator.generate_topics"
|
|
) as mock_gen:
|
|
mock_gen.return_value = ["AI", "Machine Learning"]
|
|
|
|
result = registry.extract_topics("Content about AI and ML")
|
|
|
|
mock_gen.assert_called_once()
|
|
assert "AI" in result
|
|
assert "Machine Learning" in result
|
|
|
|
def test_extract_topics_registers_discovered(self):
|
|
"""Test that extracted topics are registered."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
with patch(
|
|
"local_deep_research.news.utils.topic_generator.generate_topics"
|
|
) as mock_gen:
|
|
mock_gen.return_value = ["Climate", "Renewable Energy"]
|
|
|
|
registry.extract_topics("Content about climate")
|
|
|
|
assert "Climate" in registry.topics
|
|
assert "Renewable Energy" in registry.topics
|
|
|
|
def test_extract_topics_respects_max_topics(self):
|
|
"""Test that extract_topics passes max_topics parameter."""
|
|
from local_deep_research.news.preference_manager.base_preference import (
|
|
TopicRegistry,
|
|
)
|
|
|
|
registry = TopicRegistry()
|
|
|
|
with patch(
|
|
"local_deep_research.news.utils.topic_generator.generate_topics"
|
|
) as mock_gen:
|
|
mock_gen.return_value = ["Topic1", "Topic2", "Topic3"]
|
|
|
|
registry.extract_topics("Some content", max_topics=3)
|
|
|
|
call_kwargs = mock_gen.call_args[1]
|
|
assert call_kwargs["max_topics"] == 3
|
|
|
|
|
|
# =============================================================================
|
|
# SQLPreferenceStorage Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestSQLPreferenceStorageInit:
|
|
"""Tests for SQLPreferenceStorage initialization."""
|
|
|
|
def test_init_with_valid_session(self):
|
|
"""Test initialization with a valid session."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
|
|
assert storage._session is mock_session
|
|
|
|
def test_init_without_session_raises_error(self):
|
|
"""Test that initialization without session raises ValueError."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Session is required"):
|
|
SQLPreferenceStorage(None)
|
|
|
|
def test_session_property(self):
|
|
"""Test session property returns correct session."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
|
|
assert storage.session is mock_session
|
|
|
|
|
|
class TestSQLPreferenceStorageCreate:
|
|
"""Tests for SQLPreferenceStorage create method."""
|
|
|
|
@patch("local_deep_research.news.preference_manager.storage.UserPreference")
|
|
def test_create_preferences(self, mock_model_class):
|
|
"""Test creating new preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.id = 123
|
|
mock_model_class.return_value = mock_pref
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
|
|
data = {
|
|
"user_id": "user123",
|
|
"liked_categories": ["tech"],
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
result = storage.create(data)
|
|
|
|
assert result == "123"
|
|
mock_session.add.assert_called_once()
|
|
mock_session.commit.assert_called_once()
|
|
|
|
|
|
class TestSQLPreferenceStorageGet:
|
|
"""Tests for SQLPreferenceStorage get method."""
|
|
|
|
def test_get_existing_preferences(self):
|
|
"""Test getting existing preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.to_dict.return_value = {"id": 123, "user_id": "user123"}
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.get("123")
|
|
|
|
assert result is not None
|
|
assert result["id"] == 123
|
|
|
|
def test_get_nonexistent_preferences(self):
|
|
"""Test getting nonexistent preferences returns None."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = None
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.get("999")
|
|
|
|
assert result is None
|
|
|
|
|
|
class TestSQLPreferenceStorageUpdate:
|
|
"""Tests for SQLPreferenceStorage update method."""
|
|
|
|
def test_update_existing_preferences(self):
|
|
"""Test updating existing preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.impact_threshold = 5
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.update("123", {"impact_threshold": 8})
|
|
|
|
assert result is True
|
|
mock_session.commit.assert_called_once()
|
|
|
|
def test_update_nonexistent_preferences(self):
|
|
"""Test updating nonexistent preferences returns False."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = None
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.update("999", {"impact_threshold": 8})
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestSQLPreferenceStorageDelete:
|
|
"""Tests for SQLPreferenceStorage delete method."""
|
|
|
|
def test_delete_existing_preferences(self):
|
|
"""Test deleting existing preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.delete("123")
|
|
|
|
assert result is True
|
|
mock_session.delete.assert_called_once_with(mock_pref)
|
|
mock_session.commit.assert_called_once()
|
|
|
|
def test_delete_nonexistent_preferences(self):
|
|
"""Test deleting nonexistent preferences returns False."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = None
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.delete("999")
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestSQLPreferenceStorageList:
|
|
"""Tests for SQLPreferenceStorage list method."""
|
|
|
|
def test_list_all_preferences(self):
|
|
"""Test listing all preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.to_dict.return_value = {"id": 1, "user_id": "user1"}
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.limit.return_value.offset.return_value.all.return_value = [
|
|
mock_pref
|
|
]
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.list()
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_list_with_user_filter(self):
|
|
"""Test listing preferences with user filter."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value = mock_query
|
|
mock_query.limit.return_value.offset.return_value.all.return_value = []
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
storage.list(filters={"user_id": "user123"})
|
|
|
|
mock_query.filter_by.assert_called_once_with(user_id="user123")
|
|
|
|
|
|
class TestGetUserPreferences:
|
|
"""Tests for get_user_preferences method."""
|
|
|
|
def test_get_user_preferences_found(self):
|
|
"""Test getting existing user preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.to_dict.return_value = {
|
|
"user_id": "user123",
|
|
"impact_threshold": 5,
|
|
}
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.get_user_preferences("user123")
|
|
|
|
assert result is not None
|
|
assert result["user_id"] == "user123"
|
|
|
|
def test_get_user_preferences_not_found(self):
|
|
"""Test getting nonexistent user preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = None
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.get_user_preferences("nonexistent")
|
|
|
|
assert result is None
|
|
|
|
|
|
class TestUpsertPreferences:
|
|
"""Tests for upsert_preferences method."""
|
|
|
|
def test_upsert_creates_new_preferences(self):
|
|
"""Test upsert creates new preferences when none exist."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
|
|
storage.get_user_preferences = MagicMock(return_value=None)
|
|
storage.create = MagicMock(return_value="new-id")
|
|
|
|
result = storage.upsert_preferences("user123", {"impact_threshold": 7})
|
|
|
|
assert result == "new-id"
|
|
storage.create.assert_called_once()
|
|
call_args = storage.create.call_args[0][0]
|
|
assert call_args["user_id"] == "user123"
|
|
|
|
def test_upsert_updates_existing_preferences(self):
|
|
"""Test upsert updates existing preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.id = 123
|
|
mock_pref.impact_threshold = 5
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
storage.get_user_preferences = MagicMock(
|
|
return_value={"id": 123, "user_id": "user123"}
|
|
)
|
|
|
|
result = storage.upsert_preferences("user123", {"impact_threshold": 8})
|
|
|
|
assert result == "123"
|
|
mock_session.commit.assert_called()
|
|
|
|
|
|
class TestAddLikedItem:
|
|
"""Tests for add_liked_item method."""
|
|
|
|
def test_add_liked_item_creates_preferences(self):
|
|
"""Test adding liked item creates preferences if not exist."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = None
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
|
|
with patch(
|
|
"local_deep_research.news.preference_manager.storage.UserPreference"
|
|
):
|
|
result = storage.add_liked_item("user123", "news_item_1", "news")
|
|
|
|
assert result is True
|
|
mock_session.add.assert_called_once()
|
|
mock_session.commit.assert_called_once()
|
|
|
|
def test_add_liked_item_updates_existing(self):
|
|
"""Test adding liked item to existing preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.liked_news_ids = ["existing_item"]
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.add_liked_item("user123", "new_item", "news")
|
|
|
|
assert result is True
|
|
assert "new_item" in mock_pref.liked_news_ids
|
|
|
|
def test_add_liked_item_no_duplicate(self):
|
|
"""Test that adding same item doesn't create duplicate."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.liked_news_ids = ["item1"]
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
storage.add_liked_item("user123", "item1", "news")
|
|
|
|
assert mock_pref.liked_news_ids.count("item1") == 1
|
|
|
|
def test_add_liked_item_100_item_limit(self):
|
|
"""Test that liked items are limited to 100."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.liked_news_ids = [f"item_{i}" for i in range(100)]
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
storage.add_liked_item("user123", "new_item", "news")
|
|
|
|
# Should keep last 100 items
|
|
assert len(mock_pref.liked_news_ids) == 100
|
|
assert "new_item" in mock_pref.liked_news_ids
|
|
|
|
|
|
class TestAddDislikedItem:
|
|
"""Tests for add_disliked_item method."""
|
|
|
|
def test_add_disliked_item_creates_preferences(self):
|
|
"""Test adding disliked item creates preferences if not exist."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = None
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
|
|
with patch(
|
|
"local_deep_research.news.preference_manager.storage.UserPreference"
|
|
):
|
|
result = storage.add_disliked_item("user123", "news_item_1", "news")
|
|
|
|
assert result is True
|
|
mock_session.add.assert_called_once()
|
|
|
|
def test_add_disliked_item_updates_existing(self):
|
|
"""Test adding disliked item to existing preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.disliked_news_ids = ["existing_item"]
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
result = storage.add_disliked_item("user123", "new_item", "news")
|
|
|
|
assert result is True
|
|
assert "new_item" in mock_pref.disliked_news_ids
|
|
|
|
def test_add_disliked_item_100_item_limit(self):
|
|
"""Test that disliked items are limited to 100."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.disliked_news_ids = [f"item_{i}" for i in range(100)]
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
storage.add_disliked_item("user123", "new_item", "news")
|
|
|
|
# Should keep last 100 items
|
|
assert len(mock_pref.disliked_news_ids) == 100
|
|
assert "new_item" in mock_pref.disliked_news_ids
|
|
|
|
|
|
class TestUpdatePreferenceEmbedding:
|
|
"""Tests for update_preference_embedding method."""
|
|
|
|
def test_update_embedding_creates_preferences(self):
|
|
"""Test updating embedding creates preferences if not exist."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = None
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
|
|
with patch(
|
|
"local_deep_research.news.preference_manager.storage.UserPreference"
|
|
):
|
|
result = storage.update_preference_embedding(
|
|
"user123", [0.1, 0.2, 0.3]
|
|
)
|
|
|
|
assert result is True
|
|
mock_session.add.assert_called_once()
|
|
|
|
def test_update_embedding_on_existing_preferences(self):
|
|
"""Test updating embedding on existing preferences."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.__enter__ = MagicMock(return_value=mock_session)
|
|
mock_session.__exit__ = MagicMock(return_value=False)
|
|
|
|
mock_pref = MagicMock()
|
|
mock_pref.preference_embedding = None
|
|
|
|
mock_query = MagicMock()
|
|
mock_query.filter_by.return_value.first.return_value = mock_pref
|
|
mock_session.query.return_value = mock_query
|
|
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
embedding = [0.1, 0.2, 0.3, 0.4, 0.5]
|
|
result = storage.update_preference_embedding("user123", embedding)
|
|
|
|
assert result is True
|
|
assert mock_pref.preference_embedding == embedding
|
|
mock_session.commit.assert_called_once()
|
|
|
|
|
|
class TestInheritance:
|
|
"""Tests for SQLPreferenceStorage inheritance."""
|
|
|
|
def test_inherits_from_preference_storage(self):
|
|
"""Test that SQLPreferenceStorage inherits from PreferenceStorage."""
|
|
from local_deep_research.news.preference_manager.storage import (
|
|
SQLPreferenceStorage,
|
|
)
|
|
from local_deep_research.news.core.storage import PreferenceStorage
|
|
|
|
mock_session = MagicMock()
|
|
storage = SQLPreferenceStorage(mock_session)
|
|
|
|
assert isinstance(storage, PreferenceStorage)
|